diff --git a/src/a2a/server.test.ts b/src/a2a/server.test.ts index 5d4f697..f42452d 100644 --- a/src/a2a/server.test.ts +++ b/src/a2a/server.test.ts @@ -111,6 +111,47 @@ describe("A2A Director server", () => { }); }); + it("isolates tasks between server instances", async () => { + const appA = createDirectorA2AServer({ + dependencies: { + draftStoryboard: async () => storyboard, + writeStoryboard: vi.fn(async () => undefined), + }, + }); + const appB = createDirectorA2AServer({ + dependencies: { + draftStoryboard: async () => storyboard, + writeStoryboard: vi.fn(async () => undefined), + }, + }); + + await withServer(appA, async (baseUrlA) => { + await withServer(appB, async (baseUrlB) => { + // Send a task to server A only + await sendJsonRpc(baseUrlA, "SendMessage", { + skill: "draft_storyboard", + payload: { title: "Sample App", appUrl: "https://example.test" }, + }); + + // Server A should have 1 task + const listA = await sendJsonRpc<{ readonly result: { readonly tasks: readonly unknown[] } }>( + baseUrlA, + "ListTasks", + {}, + ); + expect(listA.result.tasks).toHaveLength(1); + + // Server B should have 0 tasks + const listB = await sendJsonRpc<{ readonly result: { readonly tasks: readonly unknown[] } }>( + baseUrlB, + "ListTasks", + {}, + ); + expect(listB.result.tasks).toHaveLength(0); + }); + }); + }); + it("completes an approved premiere task and exposes artifact paths", async () => { const paths = makeProductionPaths(); const app = createDirectorA2AServer({ diff --git a/src/a2a/server.ts b/src/a2a/server.ts index 0c52c52..52e28ca 100644 --- a/src/a2a/server.ts +++ b/src/a2a/server.ts @@ -88,7 +88,9 @@ const defaultDirectorA2ADependencies: DirectorA2ADependencies = { getProductionPaths, }; -const tasks = new Map(); +type TaskStore = Map; + +const MAX_TASKS = 1000; export type A2AAdapterOptions = { readonly port?: number; @@ -101,6 +103,7 @@ export function createDirectorA2AServer(options: A2AAdapterOptions = {}): expres const port = options.port ?? Number(process.env.DIRECTOR_A2A_PORT ?? 4129); const baseUrl = options.baseUrl ?? process.env.DIRECTOR_A2A_BASE_URL ?? `http://localhost:${port}`; const dependencies: DirectorA2ADependencies = { ...defaultDirectorA2ADependencies, ...options.dependencies }; + const tasks: TaskStore = new Map(); app.use(express.json({ limit: "2mb" })); @@ -111,7 +114,7 @@ export function createDirectorA2AServer(options: A2AAdapterOptions = {}): expres app.post("/a2a", async (request, response) => { const body = request.body as { readonly id?: unknown; readonly method?: unknown; readonly params?: unknown }; try { - const result = await handleJsonRpc(String(body.method ?? ""), body.params, dependencies); + const result = await handleJsonRpc(String(body.method ?? ""), body.params, dependencies, tasks); response.json({ jsonrpc: "2.0", id: body.id ?? null, result }); } catch (error) { response.status(400).json({ @@ -143,10 +146,11 @@ async function handleJsonRpc( method: string, params: unknown, dependencies: DirectorA2ADependencies, + tasks: TaskStore, ): Promise { switch (method) { case "SendMessage": - return { task: await createAndRunTask(extractTaskInput(params), dependencies) }; + return { task: await createAndRunTask(extractTaskInput(params), dependencies, tasks) }; case "GetTask": { const taskId = extractTaskId(params); const task = tasks.get(taskId); @@ -168,7 +172,15 @@ async function handleJsonRpc( } } -async function createAndRunTask(input: DirectorTaskInput, dependencies: DirectorA2ADependencies): Promise { +function evictOldest(tasks: TaskStore, max: number): void { + while (tasks.size > max) { + const oldest = tasks.keys().next().value; + if (oldest === undefined) break; + tasks.delete(oldest); + } +} + +async function createAndRunTask(input: DirectorTaskInput, dependencies: DirectorA2ADependencies, tasks: TaskStore): Promise { const id = crypto.randomUUID(); const contextId = crypto.randomUUID(); const submitted = makeTask( @@ -180,6 +192,7 @@ async function createAndRunTask(input: DirectorTaskInput, dependencies: Director { skill: input.skill }, ); tasks.set(id, submitted); + evictOldest(tasks, MAX_TASKS); const working = updateTask(submitted, "TASK_STATE_WORKING", `Director is working on ${input.skill}.`, []); tasks.set(id, working);