From c4c31a529cc06f8228024dfd52382362be099305 Mon Sep 17 00:00:00 2001 From: Eugene Date: Mon, 5 Oct 2026 11:59:50 +0000 Subject: [PATCH] feat(core): Add parent-Agent plan tools (no-changelog) (#39542) --- .../configs/__tests__/agents.config.test.ts | 20 ++ .../@n8n/config/src/configs/agents.config.ts | 3 + packages/@n8n/config/test/config.test.ts | 1 + .../agent-workflow-execution.service.test.ts | 4 + .../agents-service-reconstruct-gating.test.ts | 97 ++++++ .../src/modules/agents/agent-plan.service.ts | 15 +- .../agent-runtime-reconstruction.service.ts | 12 +- .../agent-workflow-execution.service.ts | 2 +- .../plans/__tests__/agent-plan-tools.test.ts | 313 ++++++++++++++++++ .../__tests__/agent-plan.validation.test.ts | 25 ++ .../modules/agents/plans/agent-plan-tools.ts | 265 +++++++++++++++ .../modules/agents/plans/agent-plan.schema.ts | 7 + .../integration/agent-plan.service.test.ts | 129 +++++++- 13 files changed, 886 insertions(+), 7 deletions(-) create mode 100644 packages/@n8n/config/src/configs/__tests__/agents.config.test.ts create mode 100644 packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts create mode 100644 packages/cli/src/modules/agents/plans/agent-plan-tools.ts diff --git a/packages/@n8n/config/src/configs/__tests__/agents.config.test.ts b/packages/@n8n/config/src/configs/__tests__/agents.config.test.ts new file mode 100644 index 00000000000..bd4b77b528a --- /dev/null +++ b/packages/@n8n/config/src/configs/__tests__/agents.config.test.ts @@ -0,0 +1,20 @@ +import { Container } from '@n8n/di'; + +import { AgentsConfig } from '../agents.config'; + +describe('AgentsConfig plan tools', () => { + afterEach(() => { + vi.unstubAllEnvs(); + Container.reset(); + }); + + it.each([ + ['', false], + ['false', false], + ['true', true], + ])('parses N8N_AGENTS_PLAN_TOOLS_ENABLED=%s as %s', (value, expected) => { + vi.stubEnv('N8N_AGENTS_PLAN_TOOLS_ENABLED', value); + Container.reset(); + expect(Container.get(AgentsConfig).planToolsEnabled).toBe(expected); + }); +}); diff --git a/packages/@n8n/config/src/configs/agents.config.ts b/packages/@n8n/config/src/configs/agents.config.ts index 69a98a74ae5..ef77d9a0c62 100644 --- a/packages/@n8n/config/src/configs/agents.config.ts +++ b/packages/@n8n/config/src/configs/agents.config.ts @@ -34,6 +34,9 @@ class AgentsModuleArray extends CommaSeparatedStringArray { @Config export class AgentsConfig { + @Env('N8N_AGENTS_PLAN_TOOLS_ENABLED') + planToolsEnabled: boolean = false; + /** TTL in seconds for agent checkpoint records. Stale checkpoints older than this are pruned. */ @Env('N8N_AGENTS_CHECKPOINT_TTL') checkpointTtlSeconds: number = 96 * Time.hours.toSeconds; diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index e66f9105efe..e76704325c2 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -724,6 +724,7 @@ describe('GlobalConfig', () => { communityPackages: '', }, agents: { + planToolsEnabled: false, checkpointTtlSeconds: 345600, tracingEnabled: true, tracingRecordInputs: true, diff --git a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts index 825949a501d..e9d2e6f0112 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts @@ -264,6 +264,10 @@ describe('AgentWorkflowExecutionService', () => { userId, ); + expect(reconstructionService.reconstructFromAgentEntity.mock.calls[0][8]).toEqual({ + supportsHitl: false, + allowPlanTools: false, + }); expect(runtime.agent.stream).toHaveBeenCalledWith( [{ id: 'message-1', role: 'user', content: [{ type: 'text', text: 'hello' }] }], expect.objectContaining({ diff --git a/packages/cli/src/modules/agents/__tests__/agents-service-reconstruct-gating.test.ts b/packages/cli/src/modules/agents/__tests__/agents-service-reconstruct-gating.test.ts index 18bdc8ccafd..777c4141c7b 100644 --- a/packages/cli/src/modules/agents/__tests__/agents-service-reconstruct-gating.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agents-service-reconstruct-gating.test.ts @@ -30,6 +30,8 @@ import type { WorkflowFinderService } from '@/workflows/workflow-finder.service' import type { AgentChatAttachmentService } from '../agent-chat-attachment.service'; import type { AgentKnowledgeMirrorService } from '../agent-knowledge-mirror.service'; +import { AgentPlanService } from '../agent-plan.service'; +import { AgentWakeService } from '../background/agent-wake.service'; import { AgentBackgroundJobService } from '../background/agent-background-job.service'; import { SubAgentBackgroundRunner } from '../background/sub-agent-background-runner'; import { AgentRuntimeReconstructionService } from '../agent-runtime-reconstruction.service'; @@ -859,6 +861,101 @@ describe('AgentRuntimeReconstructionService.reconstructFromResolvedSource — su }); }); +describe('AgentRuntimeReconstructionService — plan tools gating', () => { + const planToolNames = ['create_plan', 'read_plan', 'update_plan', 'close_plan']; + const wakeService = mock(); + + beforeEach(() => { + vi.clearAllMocks(); + builtAgent.hasCheckpointStorage.mockReturnValue(true); + Container.set(AgentPlanService, mock()); + Container.set(AgentBackgroundJobService, mock()); + Container.set(SubAgentBackgroundRunner, mock()); + Container.set(AgentWakeService, wakeService); + }); + + afterEach(() => { + Container.get(AgentsConfig).planToolsEnabled = false; + Container.get(AgentsConfig).backgroundTasksEnabled = false; + }); + + it('keeps write_todos and omits planning instructions when disabled', async () => { + Container.get(AgentsConfig).planToolsEnabled = false; + await makeReconstructionService().reconstructFromAgentEntity( + makeAgentEntity(), + mock(), + 'production', + ); + expect(getInjectedToolNames()).toContain(WRITE_TODOS_TOOL_NAME); + for (const name of planToolNames) expect(getInjectedToolNames()).not.toContain(name); + expect(builtAgent.volatileInstructionsProvider).not.toHaveBeenCalled(); + }); + + it.each([false, true])( + 'replaces write_todos when background tasks are %s', + async (backgroundTasksEnabled) => { + Container.get(AgentsConfig).planToolsEnabled = true; + Container.get(AgentsConfig).backgroundTasksEnabled = backgroundTasksEnabled; + await makeReconstructionService().reconstructFromAgentEntity( + makeAgentEntity(), + mock(), + 'production', + ); + expect(getInjectedToolNames()).toEqual(expect.arrayContaining(planToolNames)); + expect(getInjectedToolNames()).not.toContain(WRITE_TODOS_TOOL_NAME); + if (backgroundTasksEnabled) { + expect(getInjectedToolNames()).toContain('check_background_jobs'); + expect(builtAgent.volatileInstructionsProvider).toHaveBeenCalledTimes(1); + const provider = builtAgent.volatileInstructionsProvider.mock.calls[0][0]; + wakeService.getBackgroundUpdates.mockResolvedValue('Background result'); + expect( + await provider({ persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }), + ).toBe('Background result'); + expect(wakeService.getBackgroundUpdates).toHaveBeenCalledWith('thread-1', 'resource-1'); + } else { + expect(builtAgent.volatileInstructionsProvider).not.toHaveBeenCalled(); + } + }, + ); + + it('keeps write_todos when the caller disables plan tools', async () => { + Container.get(AgentsConfig).planToolsEnabled = true; + await makeReconstructionService().reconstructFromAgentEntity( + makeAgentEntity(), + mock(), + 'production', + undefined, + undefined, + undefined, + 'manual', + undefined, + { allowPlanTools: false }, + ); + expect(getInjectedToolNames()).toContain(WRITE_TODOS_TOOL_NAME); + for (const name of planToolNames) expect(getInjectedToolNames()).not.toContain(name); + }); + + it.each(['sub-agent', 'inline'] as const)( + 'omits plan tools for %s runtimes', + async (runtimeProfile) => { + Container.get(AgentsConfig).planToolsEnabled = true; + await makeReconstructionService().reconstructFromResolvedSource({ + config: { name: 'Child', model: 'anthropic/claude-sonnet-4-5', instructions: 'Help' }, + memoryOwnerAgentId: 'child-agent-1', + projectId: 'project-1', + credentialProvider: mock(), + toolDescriptors: {}, + toolCodeByName: {}, + skills: {}, + runtimeProfile, + runType: 'production', + parentAgentIdForDelegation: 'parent-agent-1', + }); + for (const name of planToolNames) expect(getInjectedToolNames()).not.toContain(name); + }, + ); +}); + describe('AgentRuntimeReconstructionService.reconstructFromAgentEntity — background job tools gating', () => { const BACKGROUND_TOOL_NAMES = [ 'spawn_background_subagent', diff --git a/packages/cli/src/modules/agents/agent-plan.service.ts b/packages/cli/src/modules/agents/agent-plan.service.ts index 8df6f057bc9..010f05a51e9 100644 --- a/packages/cli/src/modules/agents/agent-plan.service.ts +++ b/packages/cli/src/modules/agents/agent-plan.service.ts @@ -1,8 +1,12 @@ import type { OperationContext } from '@n8n/db'; import { Service } from '@n8n/di'; -import { parseAgentPlan } from './plans/agent-plan.schema'; -import { getAgentPlanReadiness, prepareAgentPlan } from './plans/agent-plan.validation'; +import { parseAgentPlan, type AgentPlanDocument } from './plans/agent-plan.schema'; +import { + getAgentPlanReadiness, + prepareAgentPlan, + type AgentPlanReadiness, +} from './plans/agent-plan.validation'; import { AgentPlanRepository, AgentPlanWriteConflictError, @@ -13,6 +17,11 @@ import { type PlanDocumentInput = { data: unknown; formatVersion: number }; type PlanWrite = { threadId: string; planId: string; expectedRevision: number }; +export type AgentPlanSnapshot = Omit & { + data: AgentPlanDocument; + readiness: AgentPlanReadiness; +}; + @Service() export class AgentPlanService { constructor(private readonly repository: AgentPlanRepository) {} @@ -25,7 +34,7 @@ export class AgentPlanService { return this.decode(await this.repository.createActivePlan({ ...input, data }, ctx)); } - async findActivePlan(threadId: string, ctx: OperationContext) { + async findActivePlan(threadId: string, ctx: OperationContext): Promise { const plan = await this.repository.findActivePlan(threadId, ctx); return plan ? this.decode(plan) : null; } diff --git a/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts b/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts index bd505af2594..80e29f62253 100644 --- a/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts +++ b/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts @@ -174,6 +174,7 @@ interface RuntimeReconstructionOptions extends ReconstructAgentRuntimeParams { credentialIntegrations: AgentIntegrationConfig[]; subAgentDelegation: SubAgentDelegationConfig; allowBackgroundTasks?: boolean; + allowPlanTools?: boolean; /** Tools the access filter already dropped; reported together with build-time stubs. */ unavailableTools?: UnavailableTool[]; /** Set by the in-app preview chat only — see `BuildFromJsonOptions.previewChat`. */ @@ -301,6 +302,7 @@ export class AgentRuntimeReconstructionService { supportsHitl, previewChat, allowBackgroundTasks = true, + allowPlanTools = true, attributionUserId, }: { /** Pass false when the caller cannot resume a suspended run (workflow executions). */ @@ -309,6 +311,7 @@ export class AgentRuntimeReconstructionService { previewChat?: boolean; /** Disable background jobs for task-triggered runtimes. */ allowBackgroundTasks?: boolean; + allowPlanTools?: boolean; attributionUserId?: string; } = {}, ): Promise { @@ -359,6 +362,7 @@ export class AgentRuntimeReconstructionService { sandboxPrincipalHash, unavailableTools, allowBackgroundTasks, + allowPlanTools, previewChat, }); return { @@ -1050,7 +1054,13 @@ export class AgentRuntimeReconstructionService { rootSessionCapUsd: params.rootSessionCapUsd, budgetForwarded: params.budgetForwarded, }); - this.attachWriteTodosTool(agent, agentId); + if (params.allowPlanTools !== false && Container.get(AgentsConfig).planToolsEnabled) { + const { AgentPlanService } = await import('./agent-plan.service.js'); + const { createAgentPlanTools } = await import('./plans/agent-plan-tools.js'); + agent.tool(createAgentPlanTools(Container.get(AgentPlanService))); + } else { + this.attachWriteTodosTool(agent, agentId); + } agent.tool(createMarkSessionFailedTool()); if (!backgroundTasksEnabled) return; // Background tools attach only to the root agent, so its cap is the root cap. diff --git a/packages/cli/src/modules/agents/agent-workflow-execution.service.ts b/packages/cli/src/modules/agents/agent-workflow-execution.service.ts index 4288a42bf19..193d1bc7ebf 100644 --- a/packages/cli/src/modules/agents/agent-workflow-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-workflow-execution.service.ts @@ -259,7 +259,7 @@ export class AgentWorkflowExecutionService { sandboxPrincipalHash, // A workflow execution cannot resume a suspended run — it throws // instead (see `recorder.suspended` below). - { supportsHitl: false }, + { supportsHitl: false, allowPlanTools: false }, ); const applied = this.applyPerCallAgentExtras(reconstructed.agent, outputSchema, extraTools); if (!applied.ok) return applied; diff --git a/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts b/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts new file mode 100644 index 00000000000..7f569f2ea4b --- /dev/null +++ b/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts @@ -0,0 +1,313 @@ +import type { BuiltTool, ToolContext } from '@n8n/agents'; +import { randomUUID } from 'node:crypto'; +import { mock } from 'vitest-mock-extended'; + +import type { AgentPlanService } from '../../agent-plan.service'; +import { AgentPlanWriteConflictError } from '../../repositories/agent-plan.repository'; +import { createAgentPlanTools } from '../agent-plan-tools'; +import { + AgentPlanValidationError, + parseAgentPlan, + type AgentPlanDocument, + type AgentPlanTask, +} from '../agent-plan.schema'; +import { getAgentPlanReadiness } from '../agent-plan.validation'; + +const service = mock(); +const context: ToolContext = { persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }; +const task = (id = 'new:research') => ({ + id, + kind: 'task' as const, + title: 'Research', + description: 'Find the facts', + status: 'pending' as const, + dependsOn: [] as string[], +}); +const document = (...items: unknown[]) => ({ title: 'Plan', description: 'Goal', items }); +const stored = (data: AgentPlanDocument) => ({ + id: randomUUID(), + threadId: 'thread-1', + revision: 1, + formatVersion: 1, + data, + closedAt: null, + createdAt: new Date(), + updatedAt: new Date(), + readiness: getAgentPlanReadiness(data), +}); +const original = stored(parseAgentPlan(document(task(randomUUID())), 1)); +const write = { planId: original.id, expectedRevision: 1 }; +let tools: BuiltTool[]; + +async function call(name: string, input: unknown, ctx = context) { + const tool = tools.find((candidate) => candidate.name === name); + if (!tool?.handler) throw new Error(`Missing handler: ${name}`); + return await tool.handler(input, ctx); +} + +beforeEach(() => { + vi.resetAllMocks(); + tools = createAgentPlanTools(service); + service.createActivePlan.mockImplementation(async (input) => ({ + ...stored(parseAgentPlan(input.data, input.formatVersion)), + id: input.id, + threadId: input.threadId, + })); + service.findPlan.mockResolvedValue(original); + service.replacePlan.mockImplementation(async (input) => ({ + ...original, + data: parseAgentPlan(input.data, input.formatVersion), + revision: input.expectedRevision + 1, + })); +}); + +describe('Agent plan tools', () => { + it('generates permanent IDs and resolves forward, group, and child references', async () => { + const result = await call('create_plan', { + document: document( + { ...task('new:report'), dependsOn: ['new:phase'] }, + { + ...task('new:phase'), + kind: 'group', + tasks: [{ ...task('new:second'), dependsOn: ['new:first'] }, task('new:first')], + }, + ), + }); + const [input, ctx] = service.createActivePlan.mock.calls[0]; + const data = parseAgentPlan(input.data, 1); + const phase = data.items[1]; + if (phase.kind !== 'group') throw new Error('Expected a group'); + expect(data.items[0].dependsOn).toEqual([phase.id]); + expect(phase.tasks[0].dependsOn).toEqual([phase.tasks[1].id]); + expect( + new Set([ + input.id, + ...data.items.map((item) => item.id), + ...phase.tasks.map((item) => item.id), + ]).size, + ).toBe(5); + expect(input).toMatchObject({ threadId: 'thread-1', formatVersion: 1 }); + expect(ctx).toEqual({}); + expect(result).toMatchObject({ + planId: input.id, + revision: 1, + closed: false, + readiness: { ready: [phase.id, phase.tasks[1].id] }, + }); + for (const hidden of [ + 'startedAt', + 'endedAt', + 'createdAt', + 'updatedAt', + 'threadId', + 'formatVersion', + ]) { + expect(JSON.stringify(result)).not.toContain(`"${hidden}"`); + } + expect(JSON.stringify(result)).not.toContain('new:'); + }); + + it.each([ + ['duplicate aliases', document(task(), task())], + [ + 'aliases shared by a group and child', + document({ ...task(), kind: 'group', tasks: [task()] }), + ], + ['missing aliases', document({ ...task(), dependsOn: ['new:missing'] })], + ['new UUIDs', document(task(randomUUID()))], + ['invalid aliases', document(task('new:'))], + ['unknown fields', { ...document(task()), extra: true }], + ['caller timestamps', document({ ...task(), startedAt: null })], + [ + 'child timestamps', + document({ ...task(), kind: 'group', tasks: [{ ...task('new:child'), endedAt: null }] }), + ], + ])('rejects %s before creation', async (_label, proposed) => { + expect(await call('create_plan', { document: proposed })).toMatchObject({ + error: 'invalid_plan', + }); + expect(service.createActivePlan).not.toHaveBeenCalled(); + }); + + it('restores group and task timestamps while preserving canonical IDs', async () => { + const startedAt = '2026-09-24T10:00:00.000Z'; + const currentTask: AgentPlanTask = { + ...task(randomUUID()), + status: 'in_progress', + startedAt, + endedAt: null, + }; + const current = stored( + parseAgentPlan( + document({ + ...task(randomUUID()), + kind: 'group', + status: 'in_progress', + startedAt, + tasks: [currentTask], + }), + 1, + ), + ); + service.findPlan.mockResolvedValue(current); + const result = await call('update_plan', { + planId: current.id, + expectedRevision: 1, + document: document({ + ...task(current.data.items[0].id), + kind: 'group', + status: 'in_progress', + tasks: [ + { ...task(currentTask.id), status: 'done', resultSummary: 'Findings' }, + task('new:extra'), + ], + }), + }); + expect(result).toMatchObject({ revision: 2 }); + expect(service.findPlan).toHaveBeenCalledWith('thread-1', current.id, {}); + expect(service.replacePlan).toHaveBeenCalledWith( + { + planId: current.id, + threadId: 'thread-1', + expectedRevision: 1, + formatVersion: 1, + data: expect.objectContaining({ + items: [ + expect.objectContaining({ + startedAt, + endedAt: null, + tasks: [ + expect.objectContaining({ id: currentTask.id, startedAt, endedAt: null }), + expect.objectContaining({ startedAt: null, endedAt: null }), + ], + }), + ], + }), + }, + {}, + ); + }); + + it('preserves fallback references and resolves new dependent references', async () => { + const failed = { + ...original.data.items[0], + status: 'failed', + startedAt: '2026-09-24T10:00:00Z', + endedAt: '2026-09-24T11:00:00Z', + }; + service.findPlan.mockResolvedValue(stored(parseAgentPlan(document(failed), 1))); + const current = await service.findPlan('thread-1', original.id, {}); + if (!current) throw new Error('Expected a plan'); + await call('update_plan', { + planId: current.id, + expectedRevision: 1, + document: document( + { ...task(failed.id), status: 'failed' }, + { ...task('new:fallback'), fallbackFor: failed.id }, + { ...task('new:dependent'), dependsOn: ['new:fallback'] }, + ), + }); + const data = parseAgentPlan(service.replacePlan.mock.calls[0][0].data, 1); + expect(data.items[1]).toMatchObject({ fallbackFor: failed.id }); + expect(data.items[2].dependsOn).toEqual([data.items[1].id]); + }); + + it('resolves fallback aliases before service validation', async () => { + await call('create_plan', { + document: document(task(), { ...task('new:replacement'), fallbackFor: 'new:research' }), + }); + const data = parseAgentPlan(service.createActivePlan.mock.calls[0][0].data, 1); + expect(data.items[1]).toMatchObject({ fallbackFor: data.items[0].id }); + }); + + it.each([null, { ...original, revision: 2 }, { ...original, closedAt: new Date() }])( + 'rejects updates without the expected active revision', + async (current) => { + service.findPlan.mockResolvedValue(current); + expect(await call('update_plan', { ...write, document: document() })).toMatchObject({ + error: 'conflict', + message: expect.stringContaining('read_plan'), + }); + expect(service.replacePlan).not.toHaveBeenCalled(); + }, + ); + + it('returns service conflicts without retrying', async () => { + service.replacePlan.mockRejectedValue(new AgentPlanWriteConflictError()); + expect(await call('update_plan', { ...write, document: document() })).toMatchObject({ + error: 'conflict', + }); + expect(service.replacePlan).toHaveBeenCalledTimes(1); + }); + + it('returns validation errors but propagates unexpected errors', async () => { + service.createActivePlan.mockRejectedValueOnce( + new AgentPlanValidationError('New items must be pending'), + ); + expect(await call('create_plan', { document: document(task()) })).toEqual({ + error: 'invalid_plan', + message: 'New items must be pending', + }); + service.findActivePlan.mockRejectedValueOnce(new Error('Database unavailable')); + await expect(call('read_plan', {})).rejects.toThrow('Database unavailable'); + }); + + it('reads each invocation scope without caching plan state', async () => { + service.findActivePlan.mockResolvedValueOnce(original).mockResolvedValueOnce(null); + expect(await call('read_plan', {})).toMatchObject({ planId: original.id }); + expect( + await call( + 'read_plan', + {}, + { persistence: { threadId: 'thread-2', resourceId: 'resource-2' } }, + ), + ).toBeNull(); + expect(service.findActivePlan.mock.calls).toEqual([ + ['thread-1', {}], + ['thread-2', {}], + ]); + }); + + it('closes at the supplied revision and returns only the model-facing document', async () => { + service.closePlan.mockResolvedValue({ + ...original, + revision: 2, + closedAt: new Date(), + readiness: { ready: [], blocked: [] }, + }); + expect(await call('close_plan', write)).toMatchObject({ + planId: original.id, + revision: 2, + closed: true, + readiness: { ready: [], blocked: [] }, + }); + expect(service.closePlan).toHaveBeenCalledWith({ ...write, threadId: 'thread-1' }, {}); + }); + + it.each(['create_plan', 'read_plan', 'update_plan', 'close_plan'])( + 'rejects %s without thread context', + async (name) => { + expect(await call(name, {}, {})).toMatchObject({ error: 'unavailable' }); + for (const method of [ + service.findPlan, + service.findActivePlan, + service.createActivePlan, + service.replacePlan, + service.closePlan, + ]) { + expect(method).not.toHaveBeenCalled(); + } + }, + ); + + it('rejects thread and format overrides', async () => { + expect(await call('read_plan', { threadId: 'other-thread' })).toMatchObject({ + error: 'invalid_plan', + }); + expect(await call('create_plan', { document: document(), formatVersion: 2 })).toMatchObject({ + error: 'invalid_plan', + }); + expect(service.findActivePlan).not.toHaveBeenCalled(); + expect(service.createActivePlan).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cli/src/modules/agents/plans/__tests__/agent-plan.validation.test.ts b/packages/cli/src/modules/agents/plans/__tests__/agent-plan.validation.test.ts index 357b39a1c11..87681ad07a5 100644 --- a/packages/cli/src/modules/agents/plans/__tests__/agent-plan.validation.test.ts +++ b/packages/cli/src/modules/agents/plans/__tests__/agent-plan.validation.test.ts @@ -57,6 +57,31 @@ const setStatuses = (document: AgentPlanDocument, statuses: Record { + it('accepts optional presentation text without changing work or timestamps', () => { + const initial = prepare(plan(task())); + const completed = prepare(setStatuses(initial, { [initial.items[0].id]: 'done' }), initial); + const presentation = { label: 'Research complete', detail: 'Found three suitable options.' }; + const updated = prepare({ ...completed, presentation }, completed, ended); + expect(updated.presentation).toEqual(presentation); + expect(updated.items).toEqual(completed.items); + expect(initial.presentation).toBeUndefined(); + expect(prepare({ ...initial, presentation: { label: ' Researching ' } }).presentation).toEqual({ + label: 'Researching', + }); + }); + + it.each([ + null, + {}, + { label: ' ' }, + { label: 12 }, + { label: 'Researching', detail: ' ' }, + { label: 'Researching', detail: 12 }, + { label: 'Researching', status: 'done' }, + ])('rejects invalid presentation fields: %j', (presentation) => { + expect(() => prepare({ ...plan(), presentation })).toThrow(AgentPlanValidationError); + }); + it('parses both item types and initializes timestamps without changing input', () => { const data = plan(task({ resultSummary: 'A summary' }), group([task()])); const before = structuredClone(data); diff --git a/packages/cli/src/modules/agents/plans/agent-plan-tools.ts b/packages/cli/src/modules/agents/plans/agent-plan-tools.ts new file mode 100644 index 00000000000..18338faf56f --- /dev/null +++ b/packages/cli/src/modules/agents/plans/agent-plan-tools.ts @@ -0,0 +1,265 @@ +import type { BuiltTool } from '@n8n/agents'; +import { Tool } from '@n8n/agents/tool'; +import type { OperationContext } from '@n8n/db'; +import { randomUUID } from 'node:crypto'; +import { z } from 'zod'; + +import type { AgentPlanService, AgentPlanSnapshot } from '../agent-plan.service'; +import { AgentPlanWriteConflictError } from '../repositories/agent-plan.repository'; +import { + AGENT_PLAN_FORMAT_VERSION, + AgentPlanValidationError, + agentPlanDocumentSchema, + agentPlanGroupSchema, + agentPlanTaskSchema, + type AgentPlanDocument, + type AgentPlanItem, +} from './agent-plan.schema'; + +const itemId = z.union([ + z.string().uuid().toLowerCase(), + z + .string() + .regex(/^new:[A-Za-z0-9_-]+$/, 'Use a temporary ID such as new:research for new items.'), +]); +const itemFields = { id: itemId, dependsOn: z.array(itemId) }; +const taskInput = agentPlanTaskSchema + .omit({ startedAt: true, endedAt: true }) + .extend({ ...itemFields, fallbackFor: itemId.optional() }); +const groupInput = agentPlanGroupSchema + .omit({ startedAt: true, endedAt: true }) + .extend({ ...itemFields, tasks: z.array(taskInput) }); +const documentInput = agentPlanDocumentSchema.extend({ + items: z.array(z.discriminatedUnion('kind', [taskInput, groupInput])), +}); +const writeInput = z + .object({ + planId: z.string().uuid().toLowerCase(), + expectedRevision: z.number().int().positive(), + }) + .strict(); + +type AgentPlanDocumentInput = z.infer; + +function prepareDocument( + data: AgentPlanDocumentInput, + previous?: AgentPlanDocument, +): AgentPlanDocument { + const existing = new Map( + previous?.items + .flatMap((item) => (item.kind === 'group' ? [item, ...item.tasks] : [item])) + .map((item) => [item.id, item]), + ); + const aliases = new Map(); + const items = data.items.flatMap((item) => + item.kind === 'group' ? [item, ...item.tasks] : [item], + ); + + for (const item of items) { + if (item.id.startsWith('new:')) { + if (aliases.has(item.id)) { + throw new AgentPlanValidationError(`Duplicate temporary ID: ${item.id}`); + } + aliases.set(item.id, randomUUID()); + } else if (!existing.has(item.id)) { + throw new AgentPlanValidationError(`Use a new: temporary ID for the new item: ${item.id}`); + } + } + + const resolveId = (id: string): string => { + if (!id.startsWith('new:')) return id; + const resolved = aliases.get(id); + if (!resolved) throw new AgentPlanValidationError(`Unknown temporary ID: ${id}`); + return resolved; + }; + + const restoreItem = (item: AgentPlanDocumentInput['items'][number]) => ({ + ...item, + id: resolveId(item.id), + dependsOn: item.dependsOn.map(resolveId), + startedAt: existing.get(item.id)?.startedAt ?? null, + endedAt: existing.get(item.id)?.endedAt ?? null, + }); + + const restoreTask = (task: z.infer) => ({ + ...restoreItem(task), + kind: task.kind, + ...(task.fallbackFor !== undefined ? { fallbackFor: resolveId(task.fallbackFor) } : {}), + }); + + return { + ...data, + items: data.items.map((item) => + item.kind === 'group' + ? { ...restoreItem(item), kind: item.kind, tasks: item.tasks.map(restoreTask) } + : restoreTask(item), + ), + }; +} + +function presentPlan(plan: AgentPlanSnapshot | null) { + if (!plan) return null; + + const withoutTiming = (item: Item) => { + const { startedAt, endedAt, ...visible } = item; + return visible; + }; + + return { + planId: plan.id, + revision: plan.revision, + closed: plan.closedAt !== null, + document: { + ...plan.data, + items: plan.data.items.map((item) => + item.kind === 'group' + ? { ...withoutTiming(item), tasks: item.tasks.map(withoutTiming) } + : withoutTiming(item), + ), + }, + readiness: plan.readiness, + }; +} + +function scopedTool( + name: string, + description: string, + schema: Schema, + handler: ( + input: z.output, + threadId: string, + ctx: OperationContext, + ) => Promise, +) { + return new Tool(name) + .description(description) + .input(schema) + .handler(async (input, context) => { + const threadId = context.persistence?.threadId; + if (!threadId) { + return { + error: 'unavailable', + message: 'Plan tools need a persisted conversation thread.', + }; + } + + try { + return presentPlan(await handler(schema.parse(input), threadId, {})); + } catch (error) { + if (error instanceof AgentPlanWriteConflictError) { + return { + error: 'conflict', + message: + 'The plan write conflicts with the stored state. Call read_plan before another write.', + }; + } + + if (error instanceof AgentPlanValidationError || error instanceof z.ZodError) { + return { error: 'invalid_plan', message: error.message }; + } + + throw error; + } + }); +} + +export function createAgentPlanTools(service: AgentPlanService): BuiltTool[] { + return [ + scopedTool( + 'create_plan', + 'Create the active plan for this conversation. Use unique new: names for new tasks and groups. ' + + 'Use these names in dependency references. Set every task and group status to "pending", including work you will start immediately. ' + + 'After creation succeeds, use update_plan with the returned permanent IDs and revision to start ready work.', + z.object({ document: documentInput }).strict(), + async ({ document }, threadId, ctx) => + await service.createActivePlan( + { + id: randomUUID(), + threadId, + formatVersion: AGENT_PLAN_FORMAT_VERSION, + data: prepareDocument(document), + }, + ctx, + ), + ) + .systemInstruction( + 'Use planning for potentially long-running work, work with multiple steps, or work with complex dependencies. ' + + 'For this work, create or update the plan before research, other task tool calls, or sub-agent delegation. ' + + 'Discovery followed by a shortlist and parallel research requires a plan. A chat message describing the steps is not a plan. ' + + 'If the user asks for quick work, keep the plan compact rather than skipping it. ' + + 'Always save new tasks and groups with status "pending", in both create_plan and update_plan. ' + + 'Do not create an item as "in_progress" or "done", even when work starts immediately or already finished. ' + + 'After the write succeeds, use the returned permanent IDs and revision in a separate update_plan call to change statuses. ' + + 'Only start or complete work after its prerequisites are Done. ' + + 'If later tasks depend on research results, start with the known tasks and extend the plan after research. ' + + 'Do not invent placeholder tasks for work whose scope or targets are not yet known. ' + + 'Represent that stage as a Pending group with an empty tasks list and dependencies on the prerequisite work. ' + + 'For example, add "Research finalists" without tasks named "Finalist 1", "Finalist 2", or "Finalist 3". ' + + 'After accepting the shortlist results, add concrete tasks to that group before starting the research. ' + + 'Do not add final communication-only tasks, such as "Presented findings to user", to the plan. ' + + 'Include such a task only when it requires a sub-agent or a tool call beyond plan maintenance. ' + + 'If the current plan content and revision are not in context, call read_plan before updating the plan. ' + + 'For sub-agent work on a plan task, always use spawn_background_subagent, even for short or sequential tasks. ' + + 'Do not use delegate_subagent for plan work. ' + + 'This rule overrides the default foreground-delegation guidance for plan work. ' + + 'If spawn_background_subagent is unavailable, explain the limitation instead of delegating plan work in the foreground. ' + + 'Keep task and group statuses current. Accept results before marking work Done. ' + + 'Before announcing plan changes or task progress to the user, always update the plan with that information. ' + + 'Wait for the plan update to succeed before announcing the change. ' + + 'Use presentation.label for a short, factual activity label in the plan card. ' + + 'Name the current activity in a few words, such as "Checking vendor pricing", not a list of completed tasks. ' + + 'Use presentation.detail for one brief progress note, limitation, or change of approach, not a results report. ' + + 'Keep task and group titles short. Each title must explain the work without its description. ' + + 'Avoid repeated words or phrases across task and group titles at the same nesting level when clarity permits. ' + + 'Put shared context in the parent title instead of repeating it in every child title. ' + + 'Keep each title distinct and clear. Do not replace repeated words with synonyms just for variety. ' + + 'Update titles of pending or in-progress tasks to reflect the current activity. The card shows titles, not task descriptions. ' + + 'For an in-progress task, use a simple action verb with a short note in parentheses, such as "Research pricing (3 vendors)". ' + + 'When marking a task Done, use a past-tense title, such as "Researched pricing". ' + + 'Save the completed title in the same update as the Done status. Final tasks cannot change. ' + + 'Put constraints in descriptions and findings in resultSummary. ' + + 'Keep this text current through update_plan. Presentation text does not set task or group statuses. ' + + 'Before closing a plan, save a final presentation.detail summary of completed and unfinished work. ' + + 'Underlying runs do not set plan statuses.', + ) + .build(), + scopedTool( + 'read_plan', + 'Read the active plan and its current revision for this conversation. Returns null if there is no active plan.', + z.object({}).strict(), + async (_input, threadId, ctx) => await service.findActivePlan(threadId, ctx), + ).build(), + scopedTool( + 'update_plan', + 'Replace the complete active plan at the expected revision. Keep permanent IDs for existing items. ' + + 'Use new: names for new items and their references. Omit timestamps. ' + + 'Dependencies connect siblings only. Prerequisites must be done before work starts or completes. ' + + 'New tasks and groups must have status "pending". Change their statuses in a separate update after this write succeeds. ' + + 'Final items cannot change. Only pending items can be removed. ' + + 'fallbackFor references a failed or cancelled sibling task. Read the saved result for redirected dependencies.', + writeInput.extend({ document: documentInput }), + async ({ document, ...input }, threadId, ctx) => { + const current = await service.findPlan(threadId, input.planId, ctx); + if (!current || current.closedAt !== null || current.revision !== input.expectedRevision) { + throw new AgentPlanWriteConflictError(); + } + return await service.replacePlan( + { + ...input, + threadId, + formatVersion: AGENT_PLAN_FORMAT_VERSION, + data: prepareDocument(document, current.data), + }, + ctx, + ); + }, + ).build(), + scopedTool( + 'close_plan', + 'Close the active plan at the expected revision. Preserve its document and release the active-plan slot. ' + + 'This does not cancel work or imply success.', + writeInput, + async (input, threadId, ctx) => await service.closePlan({ ...input, threadId }, ctx), + ).build(), + ]; +} diff --git a/packages/cli/src/modules/agents/plans/agent-plan.schema.ts b/packages/cli/src/modules/agents/plans/agent-plan.schema.ts index f4ea5ef2e9f..3ab05dea483 100644 --- a/packages/cli/src/modules/agents/plans/agent-plan.schema.ts +++ b/packages/cli/src/modules/agents/plans/agent-plan.schema.ts @@ -29,6 +29,13 @@ export const agentPlanDocumentSchema = z .object({ title: z.string().min(1), description: z.string(), + presentation: z + .object({ + label: z.string().trim().min(1), + detail: z.string().trim().min(1).optional(), + }) + .strict() + .optional(), items: z.array(z.discriminatedUnion('kind', [agentPlanTaskSchema, agentPlanGroupSchema])), }) .strict(); diff --git a/packages/cli/test/integration/agent-plan.service.test.ts b/packages/cli/test/integration/agent-plan.service.test.ts index 9d8de70b5e8..f0812c8190f 100644 --- a/packages/cli/test/integration/agent-plan.service.test.ts +++ b/packages/cli/test/integration/agent-plan.service.test.ts @@ -8,6 +8,7 @@ import { randomUUID } from 'node:crypto'; import { AgentPlanService } from '@/modules/agents/agent-plan.service'; import { AgentExecutionThread } from '@/modules/agents/entities/agent-execution-thread.entity'; import { Agent } from '@/modules/agents/entities/agent.entity'; +import { createAgentPlanTools } from '@/modules/agents/plans/agent-plan-tools'; import { type AgentPlanDocument, type AgentPlanTask, @@ -96,14 +97,20 @@ describe('AgentPlanService', () => { it('restores typed plans and immutable revisions through a fresh service', async () => { const first = task({ title: '研究 🔎' }); const second = task({ dependsOn: [first.id] }); - const initial = await create(document(first, second)); + const initial = await create({ + ...document(first, second), + presentation: { label: 'Researching', detail: 'Checking the first source.' }, + }); const updated = await service.replacePlan( { threadId, planId: initial.id, expectedRevision: 1, formatVersion: 1, - data: document({ ...first, status: 'done', resultSummary: 'Ready' }, second), + data: { + ...document({ ...first, status: 'done', resultSummary: 'Ready' }, second), + presentation: { label: 'Research finished', detail: 'One source remains unchecked.' }, + }, }, {}, ); @@ -137,6 +144,13 @@ describe('AgentPlanService', () => { {}, ), ).rejects.toThrow(AgentPlanWriteConflictError); + const closed = await fresh.closePlan({ threadId, planId: initial.id, expectedRevision: 2 }, {}); + expect(closed.data).toEqual(updated.data); + expect(closed.data.items[1].status).toBe('pending'); + expect(await fresh.findRevision(threadId, initial.id, 3, {})).toMatchObject({ + data: updated.data, + closedAt: closed.closedAt, + }); }); it('stores fallback redirection in one revision and leaves no history after an invalid update', async () => { @@ -251,6 +265,117 @@ describe('AgentPlanService', () => { expect(await service.findActivePlan(threadId, {})).toEqual(next); }); + it('supports plan tools across turns and reconstructed tool instances', async () => { + const call = async (name: string, input: unknown) => { + const tool = createAgentPlanTools(new AgentPlanService(repository)).find( + (entry) => entry.name === name, + ); + if (!tool?.handler) throw new Error(`Missing tool: ${name}`); + return await tool.handler(input, { persistence: { threadId, resourceId: agentId } }); + }; + const proposed = { + title: 'Plan', + description: 'Goal', + items: [ + { + id: 'new:research', + kind: 'task', + title: 'Research', + description: 'Find facts', + status: 'pending', + dependsOn: [], + }, + ], + }; + expect(await call('read_plan', {})).toBeNull(); + const created = await call('create_plan', { document: proposed }); + const current = await service.findActivePlan(threadId, {}); + if (!current) throw new Error('Expected active plan'); + expect(await call('read_plan', {})).toEqual(created); + expect(await call('create_plan', { document: proposed })).toMatchObject({ error: 'conflict' }); + const write = { planId: current.id, expectedRevision: 1 }; + const running = { + ...proposed, + items: [{ ...proposed.items[0], id: current.data.items[0].id, status: 'in_progress' }], + }; + expect(await call('update_plan', { ...write, document: running })).toMatchObject({ + revision: 2, + }); + const started = await service.findActivePlan(threadId, {}); + expect(started?.data.items[0].startedAt).not.toBeNull(); + expect(await call('update_plan', { ...write, document: running })).toMatchObject({ + error: 'conflict', + }); + expect( + await call('update_plan', { + ...write, + expectedRevision: 2, + document: { ...running, items: [{ ...running.items[0], description: 'Different work' }] }, + }), + ).toMatchObject({ error: 'invalid_plan' }); + expect((await service.listHistory(threadId, current.id, {}, {})).items).toHaveLength(2); + const finished = { ...running, items: [{ ...running.items[0], status: 'done' }] }; + expect( + await call('update_plan', { ...write, expectedRevision: 2, document: finished }), + ).toMatchObject({ revision: 3 }); + const done = await service.findActivePlan(threadId, {}); + expect(done?.data.items[0].startedAt).toEqual(started?.data.items[0].startedAt); + expect(done?.data.items[0].endedAt).not.toBeNull(); + expect(await call('close_plan', { ...write, expectedRevision: 3 })).toMatchObject({ + revision: 4, + closed: true, + document: finished, + }); + expect(await call('close_plan', { ...write, expectedRevision: 3 })).toMatchObject({ + error: 'conflict', + }); + expect(await call('read_plan', {})).toBeNull(); + expect(await call('create_plan', { document: proposed })).toMatchObject({ + revision: 1, + closed: false, + }); + expect((await service.findActivePlan(threadId, {}))?.id).not.toBe(current.id); + expect((await service.listHistory(threadId, current.id, {}, {})).items).toHaveLength(4); + }); + + it('scopes shared plan tools to each thread on every call', async () => { + const current = await create(); + const otherThreadId = `test-${agentId}:${randomUUID()}`; + await dataSource.getRepository(AgentExecutionThread).save({ + id: otherThreadId, + agentId, + agentName: 'Plan model agent', + projectId, + sessionNumber: 2, + }); + const tools = createAgentPlanTools(service); + const call = async (name: string, input: unknown, scope: string) => { + const tool = tools.find((entry) => entry.name === name); + if (!tool?.handler) throw new Error(`Missing tool: ${name}`); + return await tool.handler(input, { persistence: { threadId: scope, resourceId: agentId } }); + }; + const write = { planId: current.id, expectedRevision: 1 }; + expect(await call('read_plan', {}, threadId)).toMatchObject({ planId: current.id }); + expect(await call('read_plan', {}, otherThreadId)).toBeNull(); + expect(await call('close_plan', write, otherThreadId)).toMatchObject({ error: 'conflict' }); + expect( + await call( + 'update_plan', + { ...write, document: { title: 'Other', description: '', items: [] } }, + otherThreadId, + ), + ).toMatchObject({ error: 'conflict' }); + expect( + await call( + 'create_plan', + { document: { title: 'Other', description: '', items: [] } }, + otherThreadId, + ), + ).toMatchObject({ revision: 1 }); + expect(await service.findActivePlan(threadId, {})).toEqual(current); + expect((await service.listHistory(threadId, current.id, {}, {})).items).toHaveLength(1); + }); + it('rejects unsupported stored formats without changing the storage contract', async () => { const unsupported = await repository.createActivePlan( {