mirror of
https://github.com/MoonshotAI/kimi-code.git
synced 2026-07-10 01:39:25 +00:00
feat(agent-core): swarm coordinator failure-recovery loop (retry/regenerate/reassign/drop)
This commit is contained in:
Kaiyi
parent
60bc6beeee
commit
f2cc14889b
7 changed files with 550 additions and 15 deletions
|
|
@ -1,15 +1,25 @@
|
|||
import { mapWithConcurrency } from './concurrency';
|
||||
import { parsePlan } from './parse';
|
||||
import { parsePlan, parseReviseDecision } from './parse';
|
||||
import {
|
||||
ALLOWED_WORKER_TOOLS,
|
||||
DEFAULT_WORKER_TOOLS,
|
||||
PLANNER_SYSTEM_PROMPT,
|
||||
REVISER_SYSTEM_PROMPT,
|
||||
SYNTHESIZER_SYSTEM_PROMPT,
|
||||
renderPlannerPrompt,
|
||||
renderPlannerRetryPrompt,
|
||||
renderReviseSubtaskPrompt,
|
||||
renderSynthesizerPrompt,
|
||||
} from './prompts';
|
||||
import type { SwarmCoordinatorDeps, SwarmPlan, SwarmProgress } from './types';
|
||||
import {
|
||||
DEFAULT_MAX_ATTEMPTS,
|
||||
DEFAULT_MAX_WAVES,
|
||||
type ReviseDecision,
|
||||
type Subtask,
|
||||
type SwarmCoordinatorDeps,
|
||||
type SwarmPlan,
|
||||
type SwarmProgress,
|
||||
} from './types';
|
||||
|
||||
export class SwarmCoordinator {
|
||||
constructor(private readonly deps: SwarmCoordinatorDeps) {}
|
||||
|
|
@ -29,7 +39,7 @@ export class SwarmCoordinator {
|
|||
this.progress(`Planned ${String(plan.subtasks.length)} subtasks`);
|
||||
this.emit({ phase: 'planned', total: plan.subtasks.length });
|
||||
|
||||
await this.runWave(plan);
|
||||
await this.runWithRetries(plan);
|
||||
|
||||
this.emit({ phase: 'synthesizing' });
|
||||
this.progress('Synthesizing results…');
|
||||
|
|
@ -43,7 +53,8 @@ export class SwarmCoordinator {
|
|||
});
|
||||
const succeeded = plan.subtasks.filter((s) => s.status === 'done').length;
|
||||
const failed = plan.subtasks.filter((s) => s.status === 'failed').length;
|
||||
this.emit({ phase: 'done', succeeded, failed });
|
||||
const dropped = plan.subtasks.filter((s) => s.status === 'dropped').length;
|
||||
this.emit({ phase: 'done', succeeded, failed, dropped });
|
||||
return result.result;
|
||||
}
|
||||
|
||||
|
|
@ -73,11 +84,56 @@ export class SwarmCoordinator {
|
|||
throw new Error('Swarm planner failed to produce a valid plan after one retry');
|
||||
}
|
||||
|
||||
private async runWave(plan: SwarmPlan): Promise<void> {
|
||||
/**
|
||||
* Wave loop with bounded failure recovery. Each iteration runs the pending
|
||||
* subtasks; then, for every subtask still 'failed', either force-drops it
|
||||
* (attempts exhausted) or asks the reviser how to recover it and re-queues it
|
||||
* for the next wave. Terminates when no subtasks remain pending, or when the
|
||||
* {@link DEFAULT_MAX_WAVES} safety bound is hit.
|
||||
*/
|
||||
private async runWithRetries(plan: SwarmPlan): Promise<void> {
|
||||
const maxAttempts = this.deps.maxAttempts ?? DEFAULT_MAX_ATTEMPTS;
|
||||
const maxWaves = this.deps.maxWaves ?? DEFAULT_MAX_WAVES;
|
||||
|
||||
for (let wave = 0; wave < maxWaves; wave += 1) {
|
||||
const pending = plan.subtasks.filter((s) => s.status === 'pending');
|
||||
if (pending.length === 0) break;
|
||||
|
||||
await this.runWave(pending);
|
||||
|
||||
for (const st of plan.subtasks) {
|
||||
if (st.status !== 'failed') continue;
|
||||
if (st.attempts >= maxAttempts) {
|
||||
this.forceDrop(st, `attempts exhausted (${String(st.attempts)})`);
|
||||
continue;
|
||||
}
|
||||
const decision = await this.reviseSubtask(st);
|
||||
this.emit({
|
||||
phase: 'revising',
|
||||
subtaskId: st.id,
|
||||
decision: decision.kind,
|
||||
attempt: st.attempts,
|
||||
});
|
||||
this.applyDecision(st, decision);
|
||||
}
|
||||
}
|
||||
|
||||
// Safety net: anything still pending after the wave bound is dropped so the
|
||||
// loop is guaranteed to terminate and the subtask surfaces as a gap.
|
||||
for (const st of plan.subtasks) {
|
||||
if (st.status === 'pending' || st.status === 'failed') {
|
||||
this.forceDrop(st, 'recovery wave limit reached');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Run a SUBSET of subtasks (the pending ones passed in) concurrently. */
|
||||
private async runWave(subtasks: Subtask[]): Promise<void> {
|
||||
const limit = this.deps.maxConcurrency ?? 4;
|
||||
await mapWithConcurrency(plan.subtasks, limit, async (st) => {
|
||||
await mapWithConcurrency(subtasks, limit, async (st) => {
|
||||
this.deps.signal.throwIfAborted();
|
||||
st.status = 'running';
|
||||
st.attempts += 1;
|
||||
this.progress(`▸ ${st.role}: started`);
|
||||
try {
|
||||
const out = await this.deps.spawnSubagent({
|
||||
|
|
@ -94,6 +150,7 @@ export class SwarmCoordinator {
|
|||
st.status = 'done';
|
||||
this.progress(`✓ ${st.role}: done`);
|
||||
} catch (err) {
|
||||
// A genuine swarm-wide cancel must propagate (and must NOT be revised).
|
||||
if (this.deps.signal.aborted) throw err;
|
||||
st.status = 'failed';
|
||||
st.error = err instanceof Error ? err.message : String(err);
|
||||
|
|
@ -101,4 +158,56 @@ export class SwarmCoordinator {
|
|||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Ask a reviser subagent how to recover one failed subtask. On a malformed
|
||||
* response we conservatively drop (rather than burn an attempt on a confused
|
||||
* reviser).
|
||||
*/
|
||||
private async reviseSubtask(st: Subtask): Promise<ReviseDecision> {
|
||||
const out = await this.deps.spawnSubagent({
|
||||
profileName: 'swarm-reviser',
|
||||
systemPrompt: REVISER_SYSTEM_PROMPT,
|
||||
tools: [],
|
||||
prompt: renderReviseSubtaskPrompt(st, st.error),
|
||||
description: `Swarm reviser (${st.role})`,
|
||||
signal: this.deps.signal,
|
||||
});
|
||||
return (
|
||||
parseReviseDecision(out.result) ?? {
|
||||
kind: 'drop',
|
||||
reason: 'reviser produced no valid decision',
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
/** Apply a reviser decision in place, re-queueing the subtask unless dropped. */
|
||||
private applyDecision(st: Subtask, decision: ReviseDecision): void {
|
||||
switch (decision.kind) {
|
||||
case 'retry':
|
||||
st.status = 'pending';
|
||||
return;
|
||||
case 'regenerate':
|
||||
st.prompt = decision.prompt;
|
||||
st.status = 'pending';
|
||||
return;
|
||||
case 'reassign':
|
||||
st.role = decision.role;
|
||||
st.systemPrompt = decision.systemPrompt;
|
||||
st.toolAllowlist = decision.toolAllowlist;
|
||||
st.status = 'pending';
|
||||
return;
|
||||
case 'drop':
|
||||
this.forceDrop(st, decision.reason);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/** Mark a subtask dropped, record the reason, and emit a 'dropped' event. */
|
||||
private forceDrop(st: Subtask, reason: string): void {
|
||||
st.status = 'dropped';
|
||||
st.error = st.error === undefined ? `dropped: ${reason}` : `${st.error} (dropped: ${reason})`;
|
||||
this.progress(`x ${st.role}: dropped (${reason})`);
|
||||
this.emit({ phase: 'dropped', subtaskId: st.id, reason });
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import type { SwarmPlan, Subtask } from './types';
|
||||
import type { ReviseDecision, SwarmPlan, Subtask } from './types';
|
||||
|
||||
export function extractJsonObject(text: string): string | null {
|
||||
const fence = /```(?:json)?\s*([\s\S]*?)```/.exec(text);
|
||||
|
|
@ -46,7 +46,50 @@ export function parsePlan(rootTask: string, text: string): SwarmPlan | null {
|
|||
prompt: o['prompt'],
|
||||
toolAllowlist,
|
||||
status: 'pending',
|
||||
attempts: 0,
|
||||
});
|
||||
}
|
||||
return { rootTask, subtasks };
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a reviser subagent's decision about a single failed subtask. Returns
|
||||
* `null` on any malformed input (missing/invalid `kind` or required per-variant
|
||||
* fields) so the caller can apply a conservative fallback.
|
||||
*/
|
||||
export function parseReviseDecision(text: string): ReviseDecision | null {
|
||||
const json = extractJsonObject(text);
|
||||
if (json === null) return null;
|
||||
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(json);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (typeof parsed !== 'object' || parsed === null) return null;
|
||||
const o = parsed as Record<string, unknown>;
|
||||
|
||||
switch (o['kind']) {
|
||||
case 'retry':
|
||||
return { kind: 'retry' };
|
||||
case 'regenerate':
|
||||
if (typeof o['prompt'] !== 'string' || o['prompt'].length === 0) return null;
|
||||
return { kind: 'regenerate', prompt: o['prompt'] };
|
||||
case 'reassign': {
|
||||
if (typeof o['role'] !== 'string' || o['role'].length === 0) return null;
|
||||
if (typeof o['systemPrompt'] !== 'string' || o['systemPrompt'].length === 0) return null;
|
||||
const toolAllowlist = Array.isArray(o['toolAllowlist'])
|
||||
? o['toolAllowlist'].filter((t): t is string => typeof t === 'string')
|
||||
: undefined;
|
||||
return toolAllowlist === undefined
|
||||
? { kind: 'reassign', role: o['role'], systemPrompt: o['systemPrompt'] }
|
||||
: { kind: 'reassign', role: o['role'], systemPrompt: o['systemPrompt'], toolAllowlist };
|
||||
}
|
||||
case 'drop':
|
||||
if (typeof o['reason'] !== 'string' || o['reason'].length === 0) return null;
|
||||
return { kind: 'drop', reason: o['reason'] };
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import type { SwarmPlan } from './types';
|
||||
import type { Subtask, SwarmPlan } from './types';
|
||||
|
||||
/** Read-only default tool set for workers; planner may widen via toolAllowlist within the allowlist. */
|
||||
export const DEFAULT_WORKER_TOOLS: readonly string[] = ['Read', 'Grep', 'Glob', 'WebSearch', 'FetchURL'];
|
||||
|
|
@ -41,14 +41,50 @@ export function renderPlannerRetryPrompt(rootTask: string, previous: string): st
|
|||
export const SYNTHESIZER_SYSTEM_PROMPT = [
|
||||
'You are a swarm synthesizer. You are given the original task and the outputs of several worker subagents.',
|
||||
'Merge them into one coherent, complete answer for the user.',
|
||||
'If a subtask failed, note the gap explicitly instead of inventing its content.',
|
||||
'If a subtask failed or was dropped, surface the gap explicitly instead of inventing its content. Never pretend a dropped or failed subtask succeeded.',
|
||||
].join('\n');
|
||||
|
||||
export function renderSynthesizerPrompt(plan: SwarmPlan): string {
|
||||
const blocks = plan.subtasks.map((st) => {
|
||||
const body =
|
||||
st.status === 'done' ? (st.result ?? '') : `[FAILED: ${st.error ?? 'unknown error'}]`;
|
||||
let body: string;
|
||||
if (st.status === 'done') {
|
||||
body = st.result ?? '';
|
||||
} else if (st.status === 'dropped') {
|
||||
body = `[DROPPED: ${st.error ?? 'no reason given'}]`;
|
||||
} else {
|
||||
body = `[FAILED: ${st.error ?? 'unknown error'}]`;
|
||||
}
|
||||
return `### ${st.role} (${st.status})\n${body}`;
|
||||
});
|
||||
return [`Original task:\n${plan.rootTask}`, '', 'Worker outputs:', '', ...blocks].join('\n');
|
||||
}
|
||||
|
||||
export const REVISER_SYSTEM_PROMPT = [
|
||||
'You are a swarm reviser. You are given ONE subtask that failed (a real error or a detected stall/loop) along with its error.',
|
||||
'Decide how to recover it by choosing exactly one of:',
|
||||
'- retry: re-run the subtask unchanged (use only for transient/flaky errors).',
|
||||
'- regenerate: re-run with a more specific, better-scoped prompt you provide.',
|
||||
'- reassign: re-run under a different role with a new system prompt (and optionally a restricted toolAllowlist).',
|
||||
'- drop: abandon the subtask when it is impossible or not worth retrying; give a short reason.',
|
||||
'For stalled or looping errors, prefer regenerate (with a tighter, more concrete prompt) or reassign — a plain retry will usually stall again.',
|
||||
`Tools available to workers: ${ALLOWED_WORKER_TOOLS.join(', ')} (toolAllowlist may only restrict to a subset).`,
|
||||
'Output ONLY a JSON object, no prose, matching exactly one of:',
|
||||
'{"kind":"retry"}',
|
||||
'{"kind":"regenerate","prompt":"..."}',
|
||||
'{"kind":"reassign","role":"...","systemPrompt":"...","toolAllowlist":["Read"]}',
|
||||
'{"kind":"drop","reason":"..."}',
|
||||
].join('\n');
|
||||
|
||||
export function renderReviseSubtaskPrompt(subtask: Subtask, error: string | undefined): string {
|
||||
return [
|
||||
'A subtask failed. Decide how to recover it.',
|
||||
'',
|
||||
`Role: ${subtask.role}`,
|
||||
`System prompt: ${subtask.systemPrompt}`,
|
||||
`Prompt: ${subtask.prompt}`,
|
||||
`Attempts so far: ${String(subtask.attempts)}`,
|
||||
`Error: ${error ?? 'unknown error'}`,
|
||||
'',
|
||||
'Return ONLY the JSON decision object.',
|
||||
].join('\n');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,9 +4,11 @@ export interface Subtask {
|
|||
systemPrompt: string;
|
||||
prompt: string;
|
||||
toolAllowlist?: string[] | undefined;
|
||||
status: 'pending' | 'running' | 'done' | 'failed';
|
||||
status: 'pending' | 'running' | 'done' | 'failed' | 'dropped';
|
||||
result?: string | undefined;
|
||||
error?: string | undefined;
|
||||
/** Number of times this subtask has actually been executed by a worker. */
|
||||
attempts: number;
|
||||
}
|
||||
|
||||
export interface SwarmPlan {
|
||||
|
|
@ -24,10 +26,27 @@ export type SpawnSubagentFn = (args: {
|
|||
signal: AbortSignal;
|
||||
}) => Promise<{ result: string }>;
|
||||
|
||||
/**
|
||||
* Decision a reviser subagent makes about a single failed/stalled subtask.
|
||||
* Shape mirrors the JSON the reviser emits (see {@link parseReviseDecision}).
|
||||
*/
|
||||
export type ReviseDecision =
|
||||
| { kind: 'retry' }
|
||||
| { kind: 'regenerate'; prompt: string }
|
||||
| { kind: 'reassign'; role: string; systemPrompt: string; toolAllowlist?: string[] }
|
||||
| { kind: 'drop'; reason: string };
|
||||
|
||||
export type SwarmProgress =
|
||||
| { phase: 'planned'; total: number }
|
||||
| {
|
||||
phase: 'revising';
|
||||
subtaskId: string;
|
||||
decision: 'retry' | 'regenerate' | 'reassign' | 'drop';
|
||||
attempt: number;
|
||||
}
|
||||
| { phase: 'dropped'; subtaskId: string; reason: string }
|
||||
| { phase: 'synthesizing' }
|
||||
| { phase: 'done'; succeeded: number; failed: number };
|
||||
| { phase: 'done'; succeeded: number; failed: number; dropped: number };
|
||||
|
||||
export interface SwarmCoordinatorDeps {
|
||||
spawnSubagent: SpawnSubagentFn;
|
||||
|
|
@ -42,7 +61,25 @@ export interface SwarmCoordinatorDeps {
|
|||
* {@link DEFAULT_STALL_REPEAT_THRESHOLD}.
|
||||
*/
|
||||
stallRepeatThreshold?: number | undefined;
|
||||
/**
|
||||
* Maximum number of times a single subtask is executed before it is
|
||||
* force-dropped (counting the original run). Defaults to
|
||||
* {@link DEFAULT_MAX_ATTEMPTS}.
|
||||
*/
|
||||
maxAttempts?: number | undefined;
|
||||
/**
|
||||
* Safety bound on the number of wave iterations the recovery loop performs
|
||||
* before giving up, to guarantee termination. Defaults to
|
||||
* {@link DEFAULT_MAX_WAVES}.
|
||||
*/
|
||||
maxWaves?: number | undefined;
|
||||
}
|
||||
|
||||
/** Default repeat threshold for swarm worker stall detection. */
|
||||
export const DEFAULT_STALL_REPEAT_THRESHOLD = 10;
|
||||
|
||||
/** Default cap on per-subtask execution attempts before a force-drop. */
|
||||
export const DEFAULT_MAX_ATTEMPTS = 2;
|
||||
|
||||
/** Default safety cap on recovery-loop wave iterations. */
|
||||
export const DEFAULT_MAX_WAVES = 6;
|
||||
|
|
|
|||
|
|
@ -110,7 +110,11 @@ describe('SwarmCoordinator.run', () => {
|
|||
const payloads = (onProgressCustom as ReturnType<typeof vi.fn>).mock.calls.map((c) => c[0]);
|
||||
expect(payloads).toContainEqual({ phase: 'planned', total: 2 });
|
||||
expect(payloads).toContainEqual({ phase: 'synthesizing' });
|
||||
expect(payloads.some((p) => p.phase === 'done' && p.succeeded === 2 && p.failed === 0)).toBe(true);
|
||||
expect(
|
||||
payloads.some(
|
||||
(p) => p.phase === 'done' && p.succeeded === 2 && p.failed === 0 && p.dropped === 0,
|
||||
),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it('propagates abort instead of swallowing it (no synthesis after cancel)', async () => {
|
||||
|
|
@ -129,3 +133,239 @@ describe('SwarmCoordinator.run', () => {
|
|||
expect(profiles).not.toContain('swarm-synthesizer');
|
||||
});
|
||||
});
|
||||
|
||||
// One-subtask plan keeps wave behavior deterministic for recovery tests.
|
||||
const ONE_PLAN = JSON.stringify({
|
||||
subtasks: [{ id: 'task-1', role: 'Worker', systemPrompt: 'sp', prompt: 'p-original' }],
|
||||
});
|
||||
|
||||
describe('SwarmCoordinator failure recovery', () => {
|
||||
it('retry: a worker fails once, reviser says retry, re-run succeeds', async () => {
|
||||
let workerCalls = 0;
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' };
|
||||
if (args.profileName === 'swarm-reviser') return { result: '{"kind":"retry"}' };
|
||||
// swarm:Worker
|
||||
workerCalls += 1;
|
||||
if (workerCalls === 1) throw new Error('boom');
|
||||
return { result: 'worker-ok' };
|
||||
});
|
||||
const onProgressCustom = vi.fn();
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
onProgressCustom,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(workerCalls).toBe(2);
|
||||
const payloads = (onProgressCustom as ReturnType<typeof vi.fn>).mock.calls.map((c) => c[0]);
|
||||
expect(payloads).toContainEqual({
|
||||
phase: 'revising',
|
||||
subtaskId: 'task-1',
|
||||
decision: 'retry',
|
||||
attempt: 1,
|
||||
});
|
||||
expect(
|
||||
payloads.some(
|
||||
(p) => p.phase === 'done' && p.succeeded === 1 && p.failed === 0 && p.dropped === 0,
|
||||
),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it('regenerate: re-run uses the new prompt from the reviser', async () => {
|
||||
const workerPrompts: string[] = [];
|
||||
let workerCalls = 0;
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' };
|
||||
if (args.profileName === 'swarm-reviser')
|
||||
return { result: '{"kind":"regenerate","prompt":"NEW PROMPT"}' };
|
||||
workerCalls += 1;
|
||||
workerPrompts.push(args.prompt);
|
||||
if (workerCalls === 1) throw new Error('boom');
|
||||
return { result: 'worker-ok' };
|
||||
});
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(workerPrompts[0]).toBe('p-original');
|
||||
expect(workerPrompts[1]).toBe('NEW PROMPT');
|
||||
});
|
||||
|
||||
it('reassign: re-run uses the new role, systemPrompt, and tools', async () => {
|
||||
const seen: Array<{ profileName: string; systemPrompt: string; tools: string[] }> = [];
|
||||
let workerCalls = 0;
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' };
|
||||
if (args.profileName === 'swarm-reviser')
|
||||
return {
|
||||
result: '{"kind":"reassign","role":"R2","systemPrompt":"SP2","toolAllowlist":["Read"]}',
|
||||
};
|
||||
seen.push({
|
||||
profileName: args.profileName,
|
||||
systemPrompt: args.systemPrompt,
|
||||
tools: args.tools,
|
||||
});
|
||||
workerCalls += 1;
|
||||
if (workerCalls === 1) throw new Error('boom');
|
||||
return { result: 'worker-ok' };
|
||||
});
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(seen[0]?.profileName).toBe('swarm:Worker');
|
||||
expect(seen[1]?.profileName).toBe('swarm:R2');
|
||||
expect(seen[1]?.systemPrompt).toBe('SP2');
|
||||
expect(seen[1]?.tools).toEqual(['Read']);
|
||||
});
|
||||
|
||||
it('drop (LLM-chosen): a dropped subtask is not re-run and is surfaced as a gap', async () => {
|
||||
let workerCalls = 0;
|
||||
let synthesizerPrompt: string | undefined;
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') {
|
||||
synthesizerPrompt = args.prompt;
|
||||
return { result: 'SYNTH' };
|
||||
}
|
||||
if (args.profileName === 'swarm-reviser')
|
||||
return { result: '{"kind":"drop","reason":"impossible"}' };
|
||||
workerCalls += 1;
|
||||
throw new Error('boom');
|
||||
});
|
||||
const onProgressCustom = vi.fn();
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
onProgressCustom,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(workerCalls).toBe(1); // ran once, then dropped — never re-run
|
||||
const payloads = (onProgressCustom as ReturnType<typeof vi.fn>).mock.calls.map((c) => c[0]);
|
||||
expect(payloads).toContainEqual({
|
||||
phase: 'dropped',
|
||||
subtaskId: 'task-1',
|
||||
reason: 'impossible',
|
||||
});
|
||||
expect(
|
||||
payloads.some(
|
||||
(p) => p.phase === 'done' && p.succeeded === 0 && p.failed === 0 && p.dropped === 1,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(synthesizerPrompt).toMatch(/DROPPED/);
|
||||
expect(synthesizerPrompt).toContain('impossible');
|
||||
});
|
||||
|
||||
it('maxAttempts: a perpetually failing subtask runs exactly maxAttempts times then force-drops', async () => {
|
||||
let workerCalls = 0;
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' };
|
||||
if (args.profileName === 'swarm-reviser') return { result: '{"kind":"retry"}' };
|
||||
workerCalls += 1;
|
||||
throw new Error('always-boom');
|
||||
});
|
||||
const onProgressCustom = vi.fn();
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
maxAttempts: 2,
|
||||
onProgressCustom,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(workerCalls).toBe(2); // exactly maxAttempts runs
|
||||
const payloads = (onProgressCustom as ReturnType<typeof vi.fn>).mock.calls.map((c) => c[0]);
|
||||
expect(payloads.some((p) => p.phase === 'dropped' && p.subtaskId === 'task-1')).toBe(true);
|
||||
expect(
|
||||
payloads.some((p) => p.phase === 'done' && p.succeeded === 0 && p.dropped === 1),
|
||||
).toBe(true);
|
||||
// Reviser is consulted only after attempt 1 (attempt 2 hits the cap and force-drops).
|
||||
const reviserCalls = (spawn as ReturnType<typeof vi.fn>).mock.calls
|
||||
.map((c) => c[0])
|
||||
.filter((c) => c.profileName === 'swarm-reviser');
|
||||
expect(reviserCalls).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('reviser parse failure falls back to a conservative drop (does not burn attempts)', async () => {
|
||||
let workerCalls = 0;
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' };
|
||||
if (args.profileName === 'swarm-reviser') return { result: 'I am confused, no json here' };
|
||||
workerCalls += 1;
|
||||
throw new Error('boom');
|
||||
});
|
||||
const onProgressCustom = vi.fn();
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
onProgressCustom,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(workerCalls).toBe(1);
|
||||
const payloads = (onProgressCustom as ReturnType<typeof vi.fn>).mock.calls.map((c) => c[0]);
|
||||
expect(payloads).toContainEqual({
|
||||
phase: 'revising',
|
||||
subtaskId: 'task-1',
|
||||
decision: 'drop',
|
||||
attempt: 1,
|
||||
});
|
||||
expect(payloads.some((p) => p.phase === 'dropped' && p.subtaskId === 'task-1')).toBe(true);
|
||||
});
|
||||
|
||||
it('multi-wave: a revised subtask re-runs in a later wave and the loop terminates', async () => {
|
||||
// Two subtasks; both fail on wave 1, retry, both succeed on wave 2.
|
||||
const TWO_PLAN = JSON.stringify({
|
||||
subtasks: [
|
||||
{ id: 'task-1', role: 'A', systemPrompt: 'spa', prompt: 'pa' },
|
||||
{ id: 'task-2', role: 'B', systemPrompt: 'spb', prompt: 'pb' },
|
||||
],
|
||||
});
|
||||
const calls: Record<string, number> = {};
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: TWO_PLAN };
|
||||
if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' };
|
||||
if (args.profileName === 'swarm-reviser') return { result: '{"kind":"retry"}' };
|
||||
calls[args.profileName] = (calls[args.profileName] ?? 0) + 1;
|
||||
if (calls[args.profileName] === 1) throw new Error('boom');
|
||||
return { result: 'ok' };
|
||||
});
|
||||
const coordinator = new SwarmCoordinator({
|
||||
spawnSubagent: spawn,
|
||||
signal: new AbortController().signal,
|
||||
});
|
||||
const result = await coordinator.run('x');
|
||||
expect(result).toBe('SYNTH');
|
||||
expect(calls['swarm:A']).toBe(2);
|
||||
expect(calls['swarm:B']).toBe(2);
|
||||
});
|
||||
|
||||
it('does not revise on a genuine swarm-wide cancel (re-throws the abort)', async () => {
|
||||
const controller = new AbortController();
|
||||
const spawn = vi.fn(async (args) => {
|
||||
if (args.profileName === 'swarm-planner') return { result: ONE_PLAN };
|
||||
// Worker: a real swarm-wide cancel — abort the coordinator signal and throw.
|
||||
controller.abort();
|
||||
const e = new Error('aborted');
|
||||
e.name = 'AbortError';
|
||||
throw e;
|
||||
});
|
||||
const coordinator = new SwarmCoordinator({ spawnSubagent: spawn, signal: controller.signal });
|
||||
await expect(coordinator.run('x')).rejects.toThrow();
|
||||
const profiles = (spawn as ReturnType<typeof vi.fn>).mock.calls.map((c) => c[0].profileName);
|
||||
expect(profiles).not.toContain('swarm-reviser');
|
||||
expect(profiles).not.toContain('swarm-synthesizer');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { extractJsonObject, parsePlan } from '../../src/agent/swarm/parse';
|
||||
import { extractJsonObject, parsePlan, parseReviseDecision } from '../../src/agent/swarm/parse';
|
||||
|
||||
describe('extractJsonObject', () => {
|
||||
it('extracts a fenced json block', () => {
|
||||
|
|
@ -45,3 +45,63 @@ describe('parsePlan', () => {
|
|||
expect(parsePlan('root', 'totally not json')).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('parseReviseDecision', () => {
|
||||
it('parses a retry decision', () => {
|
||||
expect(parseReviseDecision('{"kind":"retry"}')).toEqual({ kind: 'retry' });
|
||||
});
|
||||
|
||||
it('parses a retry decision from a fenced block', () => {
|
||||
expect(parseReviseDecision('```json\n{"kind":"retry"}\n```')).toEqual({ kind: 'retry' });
|
||||
});
|
||||
|
||||
it('parses a regenerate decision with a new prompt', () => {
|
||||
expect(parseReviseDecision('{"kind":"regenerate","prompt":"NEW"}')).toEqual({
|
||||
kind: 'regenerate',
|
||||
prompt: 'NEW',
|
||||
});
|
||||
});
|
||||
|
||||
it('parses a reassign decision with role, systemPrompt, and toolAllowlist', () => {
|
||||
expect(
|
||||
parseReviseDecision(
|
||||
'{"kind":"reassign","role":"R2","systemPrompt":"SP2","toolAllowlist":["Read"]}',
|
||||
),
|
||||
).toEqual({ kind: 'reassign', role: 'R2', systemPrompt: 'SP2', toolAllowlist: ['Read'] });
|
||||
});
|
||||
|
||||
it('parses a reassign decision without a toolAllowlist', () => {
|
||||
expect(parseReviseDecision('{"kind":"reassign","role":"R2","systemPrompt":"SP2"}')).toEqual({
|
||||
kind: 'reassign',
|
||||
role: 'R2',
|
||||
systemPrompt: 'SP2',
|
||||
});
|
||||
});
|
||||
|
||||
it('parses a drop decision with a reason', () => {
|
||||
expect(parseReviseDecision('{"kind":"drop","reason":"impossible"}')).toEqual({
|
||||
kind: 'drop',
|
||||
reason: 'impossible',
|
||||
});
|
||||
});
|
||||
|
||||
it('returns null for an unknown kind', () => {
|
||||
expect(parseReviseDecision('{"kind":"explode"}')).toBeNull();
|
||||
});
|
||||
|
||||
it('returns null when a regenerate decision misses its prompt', () => {
|
||||
expect(parseReviseDecision('{"kind":"regenerate"}')).toBeNull();
|
||||
});
|
||||
|
||||
it('returns null when a reassign decision misses required fields', () => {
|
||||
expect(parseReviseDecision('{"kind":"reassign","role":"R2"}')).toBeNull();
|
||||
});
|
||||
|
||||
it('returns null when a drop decision misses its reason', () => {
|
||||
expect(parseReviseDecision('{"kind":"drop"}')).toBeNull();
|
||||
});
|
||||
|
||||
it('returns null for non-json garbage', () => {
|
||||
expect(parseReviseDecision('totally not json')).toBeNull();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -114,6 +114,16 @@ describe('SwarmTool', () => {
|
|||
completion: Promise.resolve({ result: 'SYNTH' }),
|
||||
};
|
||||
}
|
||||
if (profileName === 'swarm-reviser') {
|
||||
// The coordinator now consults a reviser for the stalled subtask; drop
|
||||
// it so the worker is not re-run and the stall surfaces as a gap.
|
||||
return {
|
||||
agentId: 'r',
|
||||
profileName,
|
||||
resumed: false,
|
||||
completion: Promise.resolve({ result: '{"kind":"drop","reason":"stall is unrecoverable"}' }),
|
||||
};
|
||||
}
|
||||
// Worker: drive the injected stall hook with a repeated tool call. The
|
||||
// hook's onStall aborts the per-worker signal; we mirror the real
|
||||
// subagent-host path by rejecting with the generic cancel message once
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue