feat(HAI-016): complete Step 1 — create shared AgentSemaphore

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,161 @@
+import { describe, it, expect } 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);
+  });
+});

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/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";

pnpm-lock.yaml

@@ -104,6 +104,9 @@ importers:
       typescript:
         specifier: ^5.7.0
         version: 5.9.3
+      vitest:
+        specifier: ^4.1.1
+        version: 4.1.1(@types/node@25.5.0)(vite@6.4.1(@types/node@25.5.0)(tsx@4.21.0)(yaml@2.8.3))
 
 packages: