feat(core): Add parent-Agent plan tools (no-changelog) (#39542)

This commit is contained in:
Eugene 2026-10-05 11:59:50 +00:00 committed by GitHub
parent 92ae1b336f
commit c4c31a529c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 886 additions and 7 deletions

View File

@ -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);
});
});

View File

@ -34,6 +34,9 @@ class AgentsModuleArray extends CommaSeparatedStringArray<AgentsModuleName> {
@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;

View File

@ -724,6 +724,7 @@ describe('GlobalConfig', () => {
communityPackages: '',
},
agents: {
planToolsEnabled: false,
checkpointTtlSeconds: 345600,
tracingEnabled: true,
tracingRecordInputs: true,

View File

@ -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({

View File

@ -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<AgentWakeService>();
beforeEach(() => {
vi.clearAllMocks();
builtAgent.hasCheckpointStorage.mockReturnValue(true);
Container.set(AgentPlanService, mock<AgentPlanService>());
Container.set(AgentBackgroundJobService, mock<AgentBackgroundJobService>());
Container.set(SubAgentBackgroundRunner, mock<SubAgentBackgroundRunner>());
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<CredentialProvider>(),
'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<CredentialProvider>(),
'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<CredentialProvider>(),
'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<CredentialProvider>(),
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',

View File

@ -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<AgentPlanRecord, 'data'> & {
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<AgentPlanSnapshot | null> {
const plan = await this.repository.findActivePlan(threadId, ctx);
return plan ? this.decode(plan) : null;
}

View File

@ -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<ReconstructedAgentRuntime & { userToolAccessSnapshot?: UserToolAccessSnapshot }> {
@ -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.

View File

@ -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;

View File

@ -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<AgentPlanService>();
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();
});
});

View File

@ -57,6 +57,31 @@ const setStatuses = (document: AgentPlanDocument, statuses: Record<string, Agent
};
describe('Agent plan schema', () => {
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);

View File

@ -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<typeof documentInput>;
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<string, string>();
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<typeof taskInput>) => ({
...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 extends AgentPlanItem>(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<Schema extends z.ZodType>(
name: string,
description: string,
schema: Schema,
handler: (
input: z.output<Schema>,
threadId: string,
ctx: OperationContext,
) => Promise<AgentPlanSnapshot | null>,
) {
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(),
];
}

View File

@ -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();

View File

@ -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(
{