feat(HAI-016): add shared AgentSemaphore for concurrent agent limiting

- Create AgentSemaphore class in engine/src/concurrency.ts with configurable maxConcurrent
- Integrate semaphore into TriageProcessor and TaskExecutor to gate agent spawning
- Wire semaphore into merge path and dashboard engine components
- Add comprehensive tests for concurrency, executor, and triage integration
- Update JSDoc and types for maxConcurrent configuration option

packages/cli/src/commands/dashboard.ts

@@ -1,7 +1,7 @@
 import { exec } from "node:child_process";
 import { TaskStore } from "@hai/core";
 import { createServer } from "@hai/dashboard";
-import { TriageProcessor, TaskExecutor, Scheduler, aiMergeTask } from "@hai/engine";
+import { TriageProcessor, TaskExecutor, Scheduler, AgentSemaphore, aiMergeTask } from "@hai/engine";
 
 function openBrowser(url: string): void {
   const cmd =
@@ -17,13 +17,32 @@ export async function runDashboard(port: number, opts: { engine?: boolean; open?
   await store.init();
   await store.watch();
 
-  // AI-powered merge handler (used by the web UI for manual merges)
-  const onMerge = (taskId: string) =>
+  // ── Shared concurrency semaphore ──────────────────────────────────
+  //
+  // Gates all agentic activities (triage, execution, merge) behind a
+  // single slot limit so they collectively respect settings.maxConcurrent.
+  // Created eagerly so the merge queue can reference it; the engine block
+  // below passes it to triage/executor/scheduler as well.
+  //
+  // The limit is read from a cached value that is refreshed from the store
+  // on each scheduler poll cycle (see engine block below). This avoids
+  // async I/O in the synchronous getter while still picking up live changes.
+  //
+  const initialSettings = await store.getSettings();
+  let cachedMaxConcurrent = initialSettings.maxConcurrent;
+  const semaphore = new AgentSemaphore(() => cachedMaxConcurrent);
+
+  // AI-powered merge handler (used by the web UI for manual merges).
+  // Wrapped with the shared semaphore so merges count toward the global
+  // concurrency limit alongside triage and execution agents.
+  const rawMerge = (taskId: string) =>
     aiMergeTask(store, cwd, taskId, {
       onAgentText: (delta) => process.stdout.write(delta),
       onAgentTool: (name) => console.log(`[merger] tool: ${name}`),
     });
 
+  const onMerge = (taskId: string) => semaphore.run(() => rawMerge(taskId));
+
   // ── Serialized auto-merge queue ─────────────────────────────────────
   //
   // Three paths feed into this queue:
@@ -111,12 +130,14 @@ export async function runDashboard(port: number, opts: { engine?: boolean; open?
   // Optionally start the AI engine
   if (opts.engine) {
     const triage = new TriageProcessor(store, cwd, {
+      semaphore,
       onSpecifyStart: (t) => console.log(`[engine] Specifying ${t.id}...`),
       onSpecifyComplete: (t) => console.log(`[engine] ✓ ${t.id} → todo`),
       onSpecifyError: (t, e) => console.log(`[engine] ✗ ${t.id}: ${e.message}`),
     });
 
     const executor = new TaskExecutor(store, cwd, {
+      semaphore,
       onStart: (t, p) => console.log(`[engine] Executing ${t.id} in ${p}`),
       onComplete: (t) => console.log(`[engine] ✓ ${t.id} → in-review`),
       onError: (t, e) => console.log(`[engine] ✗ ${t.id}: ${e.message}`),
@@ -125,6 +146,7 @@ export async function runDashboard(port: number, opts: { engine?: boolean; open?
     const settings = await store.getSettings();
 
     const scheduler = new Scheduler(store, {
+      semaphore,
       maxConcurrent: settings.maxConcurrent,
       maxWorktrees: settings.maxWorktrees,
       onSchedule: (t) => console.log(`[engine] Scheduled ${t.id}`),
@@ -154,6 +176,8 @@ export async function runDashboard(port: number, opts: { engine?: boolean; open?
     const mergeRetryInterval = setInterval(async () => {
       try {
         const currentSettings = await store.getSettings();
+        // Refresh the cached limit so the semaphore picks up live changes
+        cachedMaxConcurrent = currentSettings.maxConcurrent;
         if (!currentSettings.autoMerge) return;
         const tasks = await store.listTasks();
         for (const t of tasks) {

packages/core/src/types.ts

@@ -43,6 +43,8 @@ export interface TaskCreateInput {
 }
 
 export interface Settings {
+  /** Maximum number of concurrent AI agents across all activity types
+   *  (triage specification, task execution, and merge operations). */
   maxConcurrent: number;
   maxWorktrees: number;
   pollIntervalMs: number;

packages/engine/package.json

@@ -7,7 +7,8 @@
   },
   "scripts": {
     "build": "tsc",
-    "typecheck": "tsc --noEmit"
+    "typecheck": "tsc --noEmit",
+    "test": "vitest run"
   },
   "dependencies": {
     "@hai/core": "workspace:*",
@@ -15,6 +16,7 @@
     "@mariozechner/pi-ai": "^0.62.0"
   },
   "devDependencies": {
-    "typescript": "^5.7.0"
+    "typescript": "^5.7.0",
+    "vitest": "^4.1.1"
   }
 }

packages/engine/src/concurrency.test.ts

@@ -0,0 +1,246 @@
+import { describe, it, expect, vi } from "vitest";
+import { AgentSemaphore } from "./concurrency.js";
+
+describe("AgentSemaphore", () => {
+  it("allows immediate acquire when under limit", async () => {
+    const sem = new AgentSemaphore(2);
+    await sem.acquire();
+    expect(sem.activeCount).toBe(1);
+    expect(sem.availableCount).toBe(1);
+    sem.release();
+    expect(sem.activeCount).toBe(0);
+    expect(sem.availableCount).toBe(2);
+  });
+
+  it("queues waiters when at capacity and unblocks FIFO", async () => {
+    const sem = new AgentSemaphore(1);
+    await sem.acquire(); // slot taken
+
+    const order: number[] = [];
+
+    const p1 = sem.acquire().then(() => order.push(1));
+    const p2 = sem.acquire().then(() => order.push(2));
+
+    // Both should be waiting
+    expect(sem.activeCount).toBe(1);
+
+    // Release — first waiter should be unblocked
+    sem.release();
+    await p1;
+    expect(order).toEqual([1]);
+    expect(sem.activeCount).toBe(1);
+
+    // Release again — second waiter
+    sem.release();
+    await p2;
+    expect(order).toEqual([1, 2]);
+    expect(sem.activeCount).toBe(1);
+
+    sem.release();
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("run() releases on success", async () => {
+    const sem = new AgentSemaphore(1);
+    const result = await sem.run(async () => {
+      expect(sem.activeCount).toBe(1);
+      return 42;
+    });
+    expect(result).toBe(42);
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("run() releases on error", async () => {
+    const sem = new AgentSemaphore(1);
+    await expect(
+      sem.run(async () => {
+        throw new Error("boom");
+      }),
+    ).rejects.toThrow("boom");
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("respects dynamic limit changes on next acquire", async () => {
+    let limit = 2;
+    const sem = new AgentSemaphore(() => limit);
+
+    await sem.acquire();
+    await sem.acquire();
+    expect(sem.activeCount).toBe(2);
+    expect(sem.availableCount).toBe(0);
+
+    // Increase the limit — next acquire should succeed immediately
+    limit = 3;
+    expect(sem.availableCount).toBe(1);
+    await sem.acquire();
+    expect(sem.activeCount).toBe(3);
+
+    sem.release();
+    sem.release();
+    sem.release();
+  });
+
+  it("blocks new acquires when limit is reduced below activeCount", async () => {
+    let limit = 3;
+    const sem = new AgentSemaphore(() => limit);
+
+    await sem.acquire();
+    await sem.acquire();
+    expect(sem.activeCount).toBe(2);
+
+    // Reduce limit below current active count
+    limit = 1;
+    expect(sem.availableCount).toBe(0);
+
+    let acquired = false;
+    const p = sem.acquire().then(() => {
+      acquired = true;
+    });
+
+    // Should not have acquired yet
+    await Promise.resolve(); // tick
+    expect(acquired).toBe(false);
+
+    // Release one slot — active goes from 2 to 1, still >= limit (1), so still blocked
+    sem.release();
+    await Promise.resolve();
+    expect(acquired).toBe(false);
+
+    // Release again — active drops to 0, which is < limit (1), so waiter unblocks
+    sem.release();
+    await p;
+    expect(acquired).toBe(true);
+    expect(sem.activeCount).toBe(1);
+
+    sem.release();
+  });
+
+  it("activeCount and availableCount are accurate under load", async () => {
+    const sem = new AgentSemaphore(3);
+    expect(sem.activeCount).toBe(0);
+    expect(sem.availableCount).toBe(3);
+    expect(sem.limit).toBe(3);
+
+    await sem.acquire();
+    expect(sem.activeCount).toBe(1);
+    expect(sem.availableCount).toBe(2);
+
+    await sem.acquire();
+    await sem.acquire();
+    expect(sem.activeCount).toBe(3);
+    expect(sem.availableCount).toBe(0);
+
+    sem.release();
+    expect(sem.activeCount).toBe(2);
+    expect(sem.availableCount).toBe(1);
+
+    sem.release();
+    sem.release();
+    expect(sem.activeCount).toBe(0);
+    expect(sem.availableCount).toBe(3);
+  });
+
+  it("run() gates concurrent calls", async () => {
+    const sem = new AgentSemaphore(2);
+    let concurrent = 0;
+    let maxConcurrent = 0;
+
+    const task = () =>
+      sem.run(async () => {
+        concurrent++;
+        maxConcurrent = Math.max(maxConcurrent, concurrent);
+        // Yield to allow other tasks to attempt to run
+        await new Promise((r) => setTimeout(r, 10));
+        concurrent--;
+      });
+
+    await Promise.all([task(), task(), task(), task(), task()]);
+    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: simulates merge-like usage with semaphore.run()", async () => {
+    const sem = new AgentSemaphore(1);
+    let concurrent = 0;
+    let maxConcurrent = 0;
+
+    // Simulate serialized merge queue where each merge also goes through semaphore
+    const rawMerge = async () => {
+      concurrent++;
+      maxConcurrent = Math.max(maxConcurrent, concurrent);
+      await new Promise((r) => setTimeout(r, 10));
+      concurrent--;
+    };
+    const onMerge = () => sem.run(rawMerge);
+
+    await Promise.all([onMerge(), onMerge(), onMerge()]);
+    expect(maxConcurrent).toBe(1);
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("integration: shared semaphore limits triage + execution + merge together", async () => {
+    const sem = new AgentSemaphore(2);
+    let concurrent = 0;
+    let maxConcurrent = 0;
+
+    const simulateAgent = () =>
+      sem.run(async () => {
+        concurrent++;
+        maxConcurrent = Math.max(maxConcurrent, concurrent);
+        await new Promise((r) => setTimeout(r, 10));
+        concurrent--;
+      });
+
+    // Simulate mixed activity: 2 triage + 2 execution + 2 merge = 6 total
+    await Promise.all([
+      simulateAgent(), // triage
+      simulateAgent(), // triage
+      simulateAgent(), // execution
+      simulateAgent(), // execution
+      simulateAgent(), // merge
+      simulateAgent(), // merge
+    ]);
+
+    // Should never exceed 2 concurrent despite 6 tasks
+    expect(maxConcurrent).toBe(2);
+    expect(sem.activeCount).toBe(0);
+  });
+
+  it("integration: semaphore is optional (no-op when absent)", async () => {
+    const opts: { semaphore?: AgentSemaphore } = {};
+    let ran = false;
+
+    const agentWork = async () => {
+      ran = true;
+    };
+
+    if (opts.semaphore) {
+      await opts.semaphore.run(agentWork);
+    } else {
+      await agentWork();
+    }
+
+    expect(ran).toBe(true);
+  });
+});

packages/engine/src/concurrency.ts

@@ -0,0 +1,98 @@
+/**
+ * A concurrency semaphore that gates all agentic activities (triage specification,
+ * task execution, and merge operations) behind a shared slot limit.
+ *
+ * The semaphore ensures that the total number of concurrently running AI agents
+ * never exceeds `maxConcurrent`, regardless of which subsystem spawned them.
+ *
+ * The limit is read dynamically at `acquire()` time via a getter callback, so
+ * live changes to `settings.maxConcurrent` take effect on the next acquire
+ * without restarting the engine. Reducing the limit below the current
+ * `activeCount` does not evict running agents — it simply blocks new acquires
+ * until enough releases bring the active count below the new limit.
+ *
+ * @example
+ * ```ts
+ * const sem = new AgentSemaphore(() => store.getSettings().then(s => s.maxConcurrent));
+ * await sem.run(async () => {
+ *   // at most maxConcurrent agents run this block concurrently
+ * });
+ * ```
+ */
+export class AgentSemaphore {
+  private _active = 0;
+  private _waiters: Array<() => void> = [];
+  private _getLimit: () => number;
+
+  /**
+   * @param limit - Either a static number or a getter that returns the current
+   *   `maxConcurrent` value. When a getter is provided the limit is re-read on
+   *   every `acquire()` call, allowing live setting changes.
+   */
+  constructor(limit: number | (() => number)) {
+    this._getLimit = typeof limit === "function" ? limit : () => limit;
+  }
+
+  /** Number of slots currently held by running agents. */
+  get activeCount(): number {
+    return this._active;
+  }
+
+  /** Number of slots available for immediate acquisition. May be 0 or negative
+   *  if the limit was reduced below the current active count. */
+  get availableCount(): number {
+    return Math.max(0, this._getLimit() - this._active);
+  }
+
+  /** Current concurrency limit. */
+  get limit(): number {
+    return this._getLimit();
+  }
+
+  /**
+   * Acquire a slot. Resolves immediately if a slot is available, otherwise
+   * queues the caller and resolves in FIFO order when a slot is released.
+   */
+  acquire(): Promise<void> {
+    if (this._active < this._getLimit()) {
+      this._active++;
+      return Promise.resolve();
+    }
+    return new Promise<void>((resolve) => {
+      this._waiters.push(() => {
+        this._active++;
+        resolve();
+      });
+    });
+  }
+
+  /**
+   * Release a previously acquired slot and unblock the next waiting caller
+   * (if any).
+   */
+  release(): void {
+    this._active--;
+    this._drain();
+  }
+
+  /**
+   * Convenience wrapper: acquires a slot, runs `fn`, and releases the slot
+   * when `fn` settles (whether it resolves or rejects).
+   */
+  async run<T>(fn: () => Promise<T>): Promise<T> {
+    await this.acquire();
+    try {
+      return await fn();
+    } finally {
+      this.release();
+    }
+  }
+
+  /** Unblock waiters while slots are available. */
+  private _drain(): void {
+    while (this._waiters.length > 0 && this._active < this._getLimit()) {
+      const next = this._waiters.shift()!;
+      next();
+    }
+  }
+}

packages/engine/src/executor.test.ts

@@ -0,0 +1,167 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { AgentSemaphore } from "./concurrency.js";
+
+// Mock external dependencies
+vi.mock("./pi.js", () => ({
+  createHaiAgent: vi.fn(),
+}));
+vi.mock("./reviewer.js", () => ({
+  reviewStep: vi.fn(),
+}));
+
+// Mock node modules used by executor
+vi.mock("node:child_process", () => ({
+  execSync: vi.fn(),
+}));
+vi.mock("node:fs", () => ({
+  existsSync: vi.fn().mockReturnValue(true),
+}));
+
+import { TaskExecutor } from "./executor.js";
+import { createHaiAgent } from "./pi.js";
+
+const mockedCreateHaiAgent = vi.mocked(createHaiAgent);
+
+function createMockStore() {
+  const listeners = new Map<string, Function[]>();
+  return {
+    on: vi.fn((event: string, fn: Function) => {
+      const existing = listeners.get(event) || [];
+      existing.push(fn);
+      listeners.set(event, existing);
+    }),
+    emit: vi.fn(),
+    listTasks: vi.fn().mockResolvedValue([]),
+    getTask: vi.fn().mockResolvedValue({
+      id: "HAI-001",
+      title: "Test",
+      description: "Test task",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    }),
+    updateTask: vi.fn().mockResolvedValue({}),
+    moveTask: vi.fn().mockResolvedValue({}),
+    logEntry: vi.fn().mockResolvedValue(undefined),
+    parseStepsFromPrompt: vi.fn().mockResolvedValue([]),
+    updateStep: vi.fn().mockResolvedValue({}),
+  } as any;
+}
+
+describe("TaskExecutor with semaphore", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("acquires semaphore before creating agent and releases after", async () => {
+    const sem = new AgentSemaphore(2);
+    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 executor = new TaskExecutor(store, "/tmp/test", { semaphore: sem });
+
+    await executor.execute({
+      id: "HAI-001",
+      title: "Test",
+      description: "Test",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    expect(acquireSpy).toHaveBeenCalledOnce();
+    expect(releaseSpy).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 executor = new TaskExecutor(store, "/tmp/test", {
+      semaphore: sem,
+      onError,
+    });
+
+    await executor.execute({
+      id: "HAI-001",
+      title: "Test",
+      description: "Test",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    expect(sem.activeCount).toBe(0);
+    expect(onError).toHaveBeenCalled();
+  });
+
+  it("concurrent executions 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 executor = new TaskExecutor(store, "/tmp/test", { semaphore: sem });
+
+    const task = (id: string) => ({
+      id,
+      title: "Test",
+      description: "Test",
+      column: "in-progress" as const,
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    await Promise.all([
+      executor.execute(task("HAI-001")),
+      executor.execute(task("HAI-002")),
+      executor.execute(task("HAI-003")),
+    ]);
+
+    expect(maxConcurrent).toBe(1);
+    expect(sem.activeCount).toBe(0);
+  });
+});

packages/engine/src/executor.ts

@@ -6,6 +6,7 @@ import { Type } from "@mariozechner/pi-ai";
 import { createHaiAgent } from "./pi.js";
 import { reviewStep } from "./reviewer.js";
 import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
+import type { AgentSemaphore } from "./concurrency.js";
 
 const STEP_STATUSES: StepStatus[] = ["pending", "in-progress", "done", "skipped"];
 
@@ -79,6 +80,7 @@ echo "done" > .DONE
 \`\`\``;
 
 export interface TaskExecutorOptions {
+  semaphore?: AgentSemaphore;
   onStart?: (task: Task, worktreePath: string) => void;
   onComplete?: (task: Task) => void;
   onError?: (task: Task, error: Error) => void;
@@ -152,33 +154,41 @@ export class TaskExecutor {
         this.createReviewStepTool(task.id, worktreePath, detail.prompt),
       ];
 
-      const { session } = await createHaiAgent({
-        cwd: worktreePath,
-        systemPrompt: EXECUTOR_SYSTEM_PROMPT,
-        tools: "coding",
-        customTools,
-        onText: (delta) => this.options.onAgentText?.(task.id, delta),
-        onToolStart: (name) => this.options.onAgentTool?.(task.id, name),
-      });
+      const agentWork = async () => {
+        const { session } = await createHaiAgent({
+          cwd: worktreePath,
+          systemPrompt: EXECUTOR_SYSTEM_PROMPT,
+          tools: "coding",
+          customTools,
+          onText: (delta) => this.options.onAgentText?.(task.id, delta),
+          onToolStart: (name) => this.options.onAgentTool?.(task.id, name),
+        });
 
-      try {
-        const agentPrompt = buildExecutionPrompt(detail);
-        await session.prompt(agentPrompt);
+        try {
+          const agentPrompt = buildExecutionPrompt(detail);
+          await session.prompt(agentPrompt);
 
-        const doneCwd = join(worktreePath, ".DONE");
-        if (existsSync(doneCwd)) {
-          await this.store.logEntry(task.id, "Execution complete — .DONE created");
-          await this.store.moveTask(task.id, "in-review");
-          console.log(`[executor] ✓ ${task.id} completed → in-review`);
-          this.options.onComplete?.(task);
-        } else {
-          await this.store.logEntry(task.id, "Agent finished without .DONE — moved to in-review for inspection");
-          await this.store.moveTask(task.id, "in-review");
-          console.log(`[executor] ⚠ ${task.id} agent finished without .DONE → in-review`);
-          this.options.onComplete?.(task);
+          const doneCwd = join(worktreePath, ".DONE");
+          if (existsSync(doneCwd)) {
+            await this.store.logEntry(task.id, "Execution complete — .DONE created");
+            await this.store.moveTask(task.id, "in-review");
+            console.log(`[executor] ✓ ${task.id} completed → in-review`);
+            this.options.onComplete?.(task);
+          } else {
+            await this.store.logEntry(task.id, "Agent finished without .DONE — moved to in-review for inspection");
+            await this.store.moveTask(task.id, "in-review");
+            console.log(`[executor] ⚠ ${task.id} agent finished without .DONE → in-review`);
+            this.options.onComplete?.(task);
+          }
+        } finally {
+          session.dispose();
         }
-      } finally {
-        session.dispose();
+      };
+
+      if (this.options.semaphore) {
+        await this.options.semaphore.run(agentWork);
+      } else {
+        await agentWork();
       }
     } catch (err: any) {
       console.error(`[executor] ✗ ${task.id} execution failed:`, err.message);

packages/engine/src/index.ts

@@ -1,3 +1,4 @@
+export { AgentSemaphore } from "./concurrency.js";
 export { TriageProcessor, type TriageProcessorOptions } from "./triage.js";
 export { TaskExecutor, type TaskExecutorOptions } from "./executor.js";
 export { Scheduler, type SchedulerOptions } from "./scheduler.js";

packages/engine/src/scheduler.ts

@@ -1,4 +1,5 @@
 import { resolveDependencyOrder, type TaskStore, type Task } from "@hai/core";
+import type { AgentSemaphore } from "./concurrency.js";
 
 export interface SchedulerOptions {
   /** Max concurrent in-progress tasks. Default: 2 */
@@ -7,6 +8,13 @@ export interface SchedulerOptions {
   maxWorktrees?: number;
   /** Milliseconds between scheduling polls. Default: 15000 */
   pollIntervalMs?: number;
+  /**
+   * Shared concurrency semaphore. When provided, the scheduler uses
+   * `semaphore.availableCount` to avoid scheduling more tasks than the
+   * global concurrency limit allows (accounting for triage and merge
+   * agents that also hold slots).
+   */
+  semaphore?: AgentSemaphore;
   /** Called when scheduler starts a task */
   onSchedule?: (task: Task) => void;
   /** Called when a task is blocked by deps */
@@ -106,9 +114,19 @@ export class Scheduler {
       }
 
       const inProgress = tasks.filter((t) => t.column === "in-progress");
+
+      // When a semaphore is provided, factor in its available slots so we
+      // don't schedule more tasks than the global limit allows. Triage and
+      // merge agents also hold semaphore slots, so availableCount may be
+      // lower than what maxConcurrent - inProgress.length would suggest.
+      const semaphoreAvailable = this.options.semaphore
+        ? this.options.semaphore.availableCount
+        : Infinity;
+
       const available = Math.min(
         maxConcurrent - inProgress.length,
         maxWorktrees - activeWorktrees,
+        semaphoreAvailable,
       );
       if (available <= 0) return;
 

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(() => {});