feat(HAI-016): complete Step 2 — integrate semaphore into TriageProcessor

packages/engine/src/concurrency.test.ts

@@ -1,4 +1,4 @@
-import { describe, it, expect } from "vitest";
+import { describe, it, expect, vi } from "vitest";
 import { AgentSemaphore } from "./concurrency.js";
 
 describe("AgentSemaphore", () => {
@@ -158,4 +158,42 @@ describe("AgentSemaphore", () => {
     expect(maxConcurrent).toBe(2);
     expect(sem.activeCount).toBe(0);
   });
+
+  it("integration: simulates triage-like usage with semaphore.run()", async () => {
+    const sem = new AgentSemaphore(1);
+    let concurrent = 0;
+    let maxConcurrent = 0;
+
+    // Simulate two specifyTask-like calls that would normally run in parallel
+    const specifyTask = async () => {
+      const agentWork = async () => {
+        concurrent++;
+        maxConcurrent = Math.max(maxConcurrent, concurrent);
+        await new Promise((r) => setTimeout(r, 10));
+        concurrent--;
+      };
+      await sem.run(agentWork);
+    };
+
+    await Promise.all([specifyTask(), specifyTask(), specifyTask()]);
+    expect(maxConcurrent).toBe(1);
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("integration: semaphore is optional (no-op when absent)", async () => {
+    const semaphore: AgentSemaphore | undefined = undefined;
+    let ran = false;
+
+    const agentWork = async () => {
+      ran = true;
+    };
+
+    if (semaphore) {
+      await semaphore.run(agentWork);
+    } else {
+      await agentWork();
+    }
+
+    expect(ran).toBe(true);
+  });
 });

packages/engine/src/triage.test.ts

@@ -0,0 +1,148 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { AgentSemaphore } from "./concurrency.js";
+
+// Mock createHaiAgent before importing TriageProcessor
+vi.mock("./pi.js", () => ({
+  createHaiAgent: vi.fn(),
+}));
+
+import { TriageProcessor } from "./triage.js";
+import { createHaiAgent } from "./pi.js";
+
+const mockedCreateHaiAgent = vi.mocked(createHaiAgent);
+
+function createMockStore(tasks: any[] = []) {
+  return {
+    listTasks: vi.fn().mockResolvedValue(tasks),
+    getTask: vi.fn().mockResolvedValue({
+      id: "HAI-001",
+      title: "Test",
+      description: "Test task",
+      column: "triage",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      prompt: "# test",
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    }),
+    updateTask: vi.fn().mockResolvedValue({}),
+    moveTask: vi.fn().mockResolvedValue({}),
+  } as any;
+}
+
+describe("TriageProcessor with semaphore", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("acquires semaphore before creating agent and releases after", async () => {
+    const sem = new AgentSemaphore(1);
+    const store = createMockStore();
+    const acquireSpy = vi.spyOn(sem, "acquire");
+    const releaseSpy = vi.spyOn(sem, "release");
+
+    mockedCreateHaiAgent.mockResolvedValue({
+      session: {
+        prompt: vi.fn().mockResolvedValue(undefined),
+        dispose: vi.fn(),
+      },
+    } as any);
+
+    const triage = new TriageProcessor(store, "/tmp/test", { semaphore: sem });
+
+    await triage.specifyTask({
+      id: "HAI-001",
+      title: "Test",
+      description: "Test",
+      column: "triage",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    // Semaphore was used via run() which calls acquire + release
+    expect(acquireSpy).toHaveBeenCalledOnce();
+    expect(releaseSpy).toHaveBeenCalledOnce();
+    expect(mockedCreateHaiAgent).toHaveBeenCalledOnce();
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("releases semaphore on agent error", async () => {
+    const sem = new AgentSemaphore(1);
+    const store = createMockStore();
+
+    mockedCreateHaiAgent.mockRejectedValue(new Error("agent failed"));
+
+    const onError = vi.fn();
+    const triage = new TriageProcessor(store, "/tmp/test", {
+      semaphore: sem,
+      onSpecifyError: onError,
+    });
+
+    await triage.specifyTask({
+      id: "HAI-001",
+      title: "Test",
+      description: "Test",
+      column: "triage",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    expect(sem.activeCount).toBe(0);
+    expect(onError).toHaveBeenCalled();
+  });
+
+  it("concurrent specifyTask calls respect semaphore limit", async () => {
+    const sem = new AgentSemaphore(1);
+    const store = createMockStore();
+    let concurrent = 0;
+    let maxConcurrent = 0;
+
+    mockedCreateHaiAgent.mockImplementation(async () => {
+      concurrent++;
+      maxConcurrent = Math.max(maxConcurrent, concurrent);
+      return {
+        session: {
+          prompt: vi.fn().mockImplementation(async () => {
+            await new Promise((r) => setTimeout(r, 10));
+            concurrent--;
+          }),
+          dispose: vi.fn(),
+        },
+      } as any;
+    });
+
+    const triage = new TriageProcessor(store, "/tmp/test", { semaphore: sem });
+
+    const task = (id: string) => ({
+      id,
+      title: "Test",
+      description: "Test",
+      column: "triage" as const,
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    await Promise.all([
+      triage.specifyTask(task("HAI-001")),
+      triage.specifyTask(task("HAI-002")),
+      triage.specifyTask(task("HAI-003")),
+    ]);
+
+    expect(maxConcurrent).toBe(1);
+    expect(sem.activeCount).toBe(0);
+  });
+});

packages/engine/src/triage.ts

@@ -1,5 +1,6 @@
 import type { TaskStore, Task, TaskDetail } from "@hai/core";
 import { createHaiAgent } from "./pi.js";
+import type { AgentSemaphore } from "./concurrency.js";
 
 const TRIAGE_SYSTEM_PROMPT = `You are a task specification agent for "hai", an AI-orchestrated task board.
 
@@ -133,6 +134,7 @@ Write the PROMPT.md directly using the write tool. Nothing else.`;
 
 export interface TriageProcessorOptions {
   pollIntervalMs?: number;
+  semaphore?: AgentSemaphore;
   onSpecifyStart?: (task: Task) => void;
   onSpecifyComplete?: (task: Task) => void;
   onSpecifyError?: (task: Task, error: Error) => void;
@@ -198,26 +200,34 @@ export class TriageProcessor {
       const detail = await this.store.getTask(task.id);
       const promptPath = `.hai/tasks/${task.id}/PROMPT.md`;
 
-      const { session } = await createHaiAgent({
-        cwd: this.rootDir,
-        systemPrompt: TRIAGE_SYSTEM_PROMPT,
-        tools: "coding",
-        onText: (delta) => this.options.onAgentText?.(task.id, delta),
-        onToolStart: (name) =>
-          console.log(`[triage] ${task.id} tool: ${name}`),
-      });
+      const agentWork = async () => {
+        const { session } = await createHaiAgent({
+          cwd: this.rootDir,
+          systemPrompt: TRIAGE_SYSTEM_PROMPT,
+          tools: "coding",
+          onText: (delta) => this.options.onAgentText?.(task.id, delta),
+          onToolStart: (name) =>
+            console.log(`[triage] ${task.id} tool: ${name}`),
+        });
 
-      try {
-        const agentPrompt = buildSpecificationPrompt(detail, promptPath);
-        await session.prompt(agentPrompt);
+        try {
+          const agentPrompt = buildSpecificationPrompt(detail, promptPath);
+          await session.prompt(agentPrompt);
 
-        // Move to todo
-        await this.store.updateTask(task.id, { status: null });
-        await this.store.moveTask(task.id, "todo");
-        console.log(`[triage] ✓ ${task.id} specified and moved to todo`);
-        this.options.onSpecifyComplete?.(task);
-      } finally {
-        session.dispose();
+          // Move to todo
+          await this.store.updateTask(task.id, { status: null });
+          await this.store.moveTask(task.id, "todo");
+          console.log(`[triage] ✓ ${task.id} specified and moved to todo`);
+          this.options.onSpecifyComplete?.(task);
+        } finally {
+          session.dispose();
+        }
+      };
+
+      if (this.options.semaphore) {
+        await this.options.semaphore.run(agentWork);
+      } else {
+        await agentWork();
       }
     } catch (err: any) {
       await this.store.updateTask(task.id, { status: null }).catch(() => {});