diff --git a/packages/agent-core/src/agent/swarm/concurrency.ts b/packages/agent-core/src/agent/swarm/concurrency.ts new file mode 100644 index 000000000..0cde561ed --- /dev/null +++ b/packages/agent-core/src/agent/swarm/concurrency.ts @@ -0,0 +1,21 @@ +export async function mapWithConcurrency( + items: readonly T[], + limit: number, + fn: (item: T, index: number) => Promise, +): Promise { + const max = Math.max(1, Math.floor(limit)); + let cursor = 0; + + async function worker(): Promise { + while (cursor < items.length) { + const index = cursor; + cursor += 1; + const item = items[index]; + if (item === undefined) continue; + await fn(item, index); + } + } + + const count = Math.min(max, items.length); + await Promise.all(Array.from({ length: count }, () => worker())); +} diff --git a/packages/agent-core/src/agent/swarm/parse.ts b/packages/agent-core/src/agent/swarm/parse.ts new file mode 100644 index 000000000..82c903fff --- /dev/null +++ b/packages/agent-core/src/agent/swarm/parse.ts @@ -0,0 +1,52 @@ +import type { SwarmPlan, Subtask } from './types'; + +export function extractJsonObject(text: string): string | null { + const fence = /```(?:json)?\s*([\s\S]*?)```/.exec(text); + const candidate = fence?.[1] ?? text; + const start = candidate.indexOf('{'); + const end = candidate.lastIndexOf('}'); + if (start === -1 || end === -1 || end < start) return null; + return candidate.slice(start, end + 1); +} + +export function parsePlan(rootTask: string, text: string): SwarmPlan | 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 subtasksRaw = (parsed as { subtasks?: unknown }).subtasks; + if (!Array.isArray(subtasksRaw) || subtasksRaw.length === 0) return null; + + const subtasks: Subtask[] = []; + for (let i = 0; i < subtasksRaw.length; i += 1) { + const raw = subtasksRaw[i]; + if (typeof raw !== 'object' || raw === null) return null; + const o = raw as Record; + if ( + typeof o['role'] !== 'string' || + typeof o['systemPrompt'] !== 'string' || + typeof o['prompt'] !== 'string' + ) { + return null; + } + const toolAllowlist = Array.isArray(o['toolAllowlist']) + ? o['toolAllowlist'].filter((t): t is string => typeof t === 'string') + : undefined; + subtasks.push({ + id: typeof o['id'] === 'string' && o['id'].length > 0 ? o['id'] : `task-${String(i + 1)}`, + role: o['role'], + systemPrompt: o['systemPrompt'], + prompt: o['prompt'], + toolAllowlist, + status: 'pending', + }); + } + return { rootTask, subtasks }; +} diff --git a/packages/agent-core/src/agent/swarm/types.ts b/packages/agent-core/src/agent/swarm/types.ts new file mode 100644 index 000000000..e15c36681 --- /dev/null +++ b/packages/agent-core/src/agent/swarm/types.ts @@ -0,0 +1,32 @@ +export interface Subtask { + id: string; + role: string; + systemPrompt: string; + prompt: string; + toolAllowlist?: string[] | undefined; + status: 'pending' | 'running' | 'done' | 'failed'; + result?: string | undefined; + error?: string | undefined; +} + +export interface SwarmPlan { + rootTask: string; + subtasks: Subtask[]; +} + +/** What the coordinator needs to run one subagent to completion. */ +export type SpawnSubagentFn = (args: { + profileName: string; + systemPrompt: string; + tools: string[]; + prompt: string; + description: string; + signal: AbortSignal; +}) => Promise<{ result: string }>; + +export interface SwarmCoordinatorDeps { + spawnSubagent: SpawnSubagentFn; + signal: AbortSignal; + onProgress?: ((text: string) => void) | undefined; + maxConcurrency?: number | undefined; +} diff --git a/packages/agent-core/test/swarm/concurrency.test.ts b/packages/agent-core/test/swarm/concurrency.test.ts new file mode 100644 index 000000000..bd7866107 --- /dev/null +++ b/packages/agent-core/test/swarm/concurrency.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from 'vitest'; + +import { mapWithConcurrency } from '../../src/agent/swarm/concurrency'; + +describe('mapWithConcurrency', () => { + it('processes every item', async () => { + const seen: number[] = []; + await mapWithConcurrency([1, 2, 3, 4], 2, async (n) => { + seen.push(n); + }); + expect(seen.sort((a, b) => a - b)).toEqual([1, 2, 3, 4]); + }); + + it('never exceeds the concurrency limit', async () => { + let active = 0; + let peak = 0; + await mapWithConcurrency([1, 2, 3, 4, 5, 6], 2, async () => { + active += 1; + peak = Math.max(peak, active); + await new Promise((r) => setTimeout(r, 5)); + active -= 1; + }); + expect(peak).toBeLessThanOrEqual(2); + }); + + it('treats a limit below 1 as 1', async () => { + const seen: number[] = []; + await mapWithConcurrency([1, 2], 0, async (n) => { + seen.push(n); + }); + expect(seen.sort((a, b) => a - b)).toEqual([1, 2]); + }); +}); diff --git a/packages/agent-core/test/swarm/parse.test.ts b/packages/agent-core/test/swarm/parse.test.ts new file mode 100644 index 000000000..b2146527c --- /dev/null +++ b/packages/agent-core/test/swarm/parse.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest'; + +import { extractJsonObject, parsePlan } from '../../src/agent/swarm/parse'; + +describe('extractJsonObject', () => { + it('extracts a fenced json block', () => { + expect(extractJsonObject('blah\n```json\n{"a":1}\n```\ntail')).toBe('{"a":1}'); + }); + it('extracts a bare object from surrounding prose', () => { + expect(extractJsonObject('here you go: {"a":1} done')).toBe('{"a":1}'); + }); + it('returns null when no object is present', () => { + expect(extractJsonObject('no json here')).toBeNull(); + }); +}); + +describe('parsePlan', () => { + const good = JSON.stringify({ + subtasks: [ + { role: 'Researcher', systemPrompt: 'be a researcher', prompt: 'research X' }, + { id: 'b', role: 'Writer', systemPrompt: 'be a writer', prompt: 'write Y', toolAllowlist: ['Read'] }, + ], + }); + + it('parses a valid plan and fills default ids', () => { + const plan = parsePlan('root', '```json\n' + good + '\n```'); + expect(plan).not.toBeNull(); + expect(plan?.rootTask).toBe('root'); + expect(plan?.subtasks).toHaveLength(2); + expect(plan?.subtasks[0]?.id).toBe('task-1'); + expect(plan?.subtasks[0]?.status).toBe('pending'); + expect(plan?.subtasks[1]?.id).toBe('b'); + expect(plan?.subtasks[1]?.toolAllowlist).toEqual(['Read']); + }); + + it('returns null for empty subtasks', () => { + expect(parsePlan('root', '{"subtasks":[]}')).toBeNull(); + }); + + it('returns null when a subtask misses required fields', () => { + expect(parsePlan('root', '{"subtasks":[{"role":"R"}]}')).toBeNull(); + }); + + it('returns null for non-json garbage', () => { + expect(parsePlan('root', 'totally not json')).toBeNull(); + }); +});