feat(HAI-016): complete Step 5 — wire semaphore into dashboard engine components

packages/cli/src/commands/dashboard.ts

@@ -130,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}`),
@@ -144,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}`),
@@ -173,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/engine/src/concurrency.test.ts

@@ -228,15 +228,15 @@ describe("AgentSemaphore", () => {
   });
 
   it("integration: semaphore is optional (no-op when absent)", async () => {
-    const semaphore: AgentSemaphore | undefined = undefined;
+    const opts: { semaphore?: AgentSemaphore } = {};
     let ran = false;
 
     const agentWork = async () => {
       ran = true;
     };
 
-    if (semaphore) {
-      await semaphore.run(agentWork);
+    if (opts.semaphore) {
+      await opts.semaphore.run(agentWork);
     } else {
       await agentWork();
     }

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;