From 120df8d226d44a62dcb906be9636d56763bb0e2c Mon Sep 17 00:00:00 2001
From: daryllundy <daryl.lundy@gmail.com>
Date: Thu, 18 Jun 2026 16:55:23 -0700
Subject: [PATCH] Make A2A task store per-server-instance and bounded

Move the task Map from module scope into each createDirectorA2AServer call
so server instances no longer share one global registry (ListTasks on one
instance previously returned another instance's tasks). Add a MAX_TASKS=1000
cap with oldest-first eviction so a long-running server's store cannot grow
unbounded. Thread the store explicitly through handleJsonRpc and
createAndRunTask, matching how dependencies are already passed. Add an
isolation test verifying two servers do not share tasks.

Found and fixed with assistance from Claude Code (Anthropic).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 src/a2a/server.test.ts | 41 +++++++++++++++++++++++++++++++++++++++++
 src/a2a/server.ts      | 21 +++++++++++++++++----
 2 files changed, 58 insertions(+), 4 deletions(-)

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<string, A2ATask>();
+type TaskStore = Map<string, A2ATask>;
+
+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<unknown> {
   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<A2ATask> {
+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<A2ATask> {
   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);