fix(agent): make project lane scheduling fair

Author: lee-to

docs/architecture.md

@@ -102,9 +102,9 @@ Short-lived in-memory caches exist only for dedupe/throttling repeated identical
 
 ## Agent Pipeline
 
-The coordinator (`packages/agent/src/coordinator.ts`) uses a dual-trigger model: it polls via `node-cron` every 30 seconds as a fallback and also reacts to real-time events from the API WebSocket (task creation, moves, and explicit `agent:wake` signals). Duplicate wakes are debounced. If the WebSocket is unavailable, the coordinator falls back to polling-only mode.
+The coordinator (`packages/agent/src/coordinator.ts`) uses a dual-trigger model: it polls via `node-cron` every 30 seconds as a fallback and also reacts to real-time events from the API WebSocket (task creation, moves, and explicit `agent:wake` signals). Duplicate wakes are debounced, and both trigger sources share a single-flight poll loop; a trigger received during an active cycle requests one coalesced follow-up cycle. If the WebSocket is unavailable, the coordinator falls back to polling-only mode.
 
-The coordinator supports **parallel task execution** (experimental, per-project). It first selects up to `COORDINATOR_MAX_CONCURRENT_PROJECTS` (default 4) independent project lanes and runs those lanes concurrently, while `COORDINATOR_MAX_CONCURRENT_TASKS` (default 12) remains a global safety ceiling across all lanes. Within one lane, pipeline stages still drain sequentially to preserve project-local ordering. When a project has "Parallel Execution" enabled in settings, up to `COORDINATOR_MAX_CONCURRENT_TASKS_PER_PROJECT` (default 3) tasks per stage run concurrently via `Promise.allSettled`; non-parallel projects always process 1 task at a time. Tasks are atomically claimed (`lockedBy`/`lockedUntil` columns) with lock duration tied to the stage timeout; heartbeats renew the lock periodically. Stale claims (expired TTL or dead heartbeat) are auto-released. On shutdown, active locks are released immediately.
+The coordinator supports **parallel task execution** (experimental, per-project). It first selects up to `COORDINATOR_MAX_CONCURRENT_PROJECTS` (default 4) independent project lanes and runs those lanes concurrently, while `COORDINATOR_MAX_CONCURRENT_TASKS` (default 12) remains a global safety ceiling across all lanes. A FIFO permit governor distributes global capacity across runnable lanes and waits for released slots instead of dropping later selected lanes. Within one lane, pipeline stages still drain sequentially to preserve project-local ordering. When a project has "Parallel Execution" enabled in settings, up to `COORDINATOR_MAX_CONCURRENT_TASKS_PER_PROJECT` (default 3) tasks per stage run concurrently via `Promise.allSettled`; non-parallel projects always process 1 task at a time. Tasks are atomically claimed (`lockedBy`/`lockedUntil` columns) with lock duration tied to the stage timeout; heartbeats renew the lock periodically. Stale claims (expired TTL or dead heartbeat) are auto-released. On shutdown, active locks are released immediately.
 
 It delegates workflow stages to `.claude/agents/` definitions, but actual execution transport/model/session behavior is adapter-owned through `@aif/runtime`:
 

docs/configuration.md

@@ -423,6 +423,11 @@ COORDINATOR_MAX_CONCURRENT_TASKS_PER_PROJECT=3
 COORDINATOR_MAX_CONCURRENT_PROJECTS=4
 ```
 
+> **Upgrading from the previous coordinator limit:** `COORDINATOR_MAX_CONCURRENT_TASKS`
+> used to control per-project/stage concurrency. It is now the global ceiling across all
+> projects. Existing deployments should set all three values above explicitly; leaving an old
+> value such as `COORDINATOR_MAX_CONCURRENT_TASKS=3` limits total coordinator concurrency to 3.
+
 2. Enable per-project in the web UI: open project settings and toggle **Parallel Execution**.
 
 3. For projects that also use `git.create_branches=true`, opt into task worktrees:
@@ -437,7 +442,8 @@ AIF_TASK_WORKTREES_ENABLED=true
   - **Parallel off** (default): 1 task per project at a time — identical to serial behavior
   - **Parallel on**: up to `COORDINATOR_MAX_CONCURRENT_TASKS_PER_PROJECT` tasks per project per stage
 - The coordinator processes up to `COORDINATOR_MAX_CONCURRENT_PROJECTS` independent project lanes concurrently in a poll cycle. Stage ordering is preserved inside a project lane, so a project's planner still drains before that same project's reviewer, but a slow planner in project A no longer blocks a reviewer in project B.
-- Across all lanes, `COORDINATOR_MAX_CONCURRENT_TASKS` remains the global safety ceiling for active coordinator tasks.
+- Across all lanes, `COORDINATOR_MAX_CONCURRENT_TASKS` remains the global safety ceiling for active coordinator tasks. Runnable lanes receive slots through a fair FIFO governor, so an older busy project cannot consume the whole first wave when other selected lanes are ready.
+- Cron ticks and WebSocket wakes share one single-flight poll loop. A wake received during an active cycle is coalesced into one follow-up cycle, preserving project-local stage order across trigger sources.
 - With `AIF_TASK_WORKTREES_ENABLED=false` (default), any branch-isolated project (`git.create_branches=true`) remains serial. The API also rejects parallel auto-queue for that combination.
 - With `AIF_TASK_WORKTREES_ENABLED=true`, full-mode planning for parallel branch-isolated projects creates a sibling git worktree for each task, persists its absolute path in `tasks.worktree_path`, and runs all downstream stages from that path. Legacy branch-bound tasks that have `branchName` but no `worktreePath` still force serial execution until they drain.
 - Tasks within a stage run concurrently via `Promise.allSettled` — a failure in one task does not block others

packages/agent/CHECKLIST.md

@@ -11,7 +11,7 @@ Run through this list whenever you touch anything under `packages/agent/`.
 - [ ] If you added a new subagent, add tests that verify it resolves the correct agent definition and handles the runtime capability fallback.
 - [ ] If you touched the coordinator polling logic, verify the state machine transitions in `@aif/shared/stateMachine.ts` still line up.
 - [ ] Polling intervals are configured in milliseconds. Do not convert values above 59 seconds into a cron step expression.
-- [ ] If you touched the poll scheduler, verify periodic ticks do not overlap while a previous poll callback is still running.
+- [ ] If you touched the poll scheduler, verify periodic ticks and event-driven wakes share a single-flight coordinator loop; triggers received during an active cycle may request only one coalesced follow-up cycle.
 - [ ] If you touched first-activity watchdog logic, verify streamed runtime events (not only tool/subagent hooks) count as activity for tool-less workflows.
 - [ ] `npm run lint`
 - [ ] `npm test`

packages/agent/src/__tests__/coordinator.test.ts

@@ -108,6 +108,24 @@ describe("coordinator", () => {
     getStageSemaphore().reset();
   });
 
+  it("should remove inactive project-stage semaphore keys", async () => {
+    const semaphore = getStageSemaphore();
+
+    await semaphore.acquire("project-1:planner", 2, 2);
+    await semaphore.acquire("project-1:planner", 2, 2);
+    expect(semaphore.totalActive()).toBe(2);
+    expect(semaphore.trackedKeyCount()).toBe(1);
+
+    semaphore.release("project-1:planner");
+    expect(semaphore.totalActive()).toBe(1);
+    expect(semaphore.trackedKeyCount()).toBe(1);
+
+    semaphore.release("project-1:planner");
+    semaphore.release("missing:planner");
+    expect(semaphore.totalActive()).toBe(0);
+    expect(semaphore.trackedKeyCount()).toBe(0);
+  });
+
   function insertRuntimeProfile(input: {
     id: string;
     projectId?: string | null;
@@ -1688,9 +1706,44 @@ describe("coordinator", () => {
         }
       }
 
-      await pollAndProcess();
+      const startedProjectIds: string[] = [];
+      const releasePlanners: Array<() => void> = [];
+      let activePlanners = 0;
+      let peakActivePlanners = 0;
+
+      vi.mocked(runPlanner).mockImplementation((taskId) => {
+        const projectId = taskId.startsWith("global-cap-task-1-")
+          ? "global-cap-project-1"
+          : "global-cap-project-2";
+        startedProjectIds.push(projectId);
+        activePlanners += 1;
+        peakActivePlanners = Math.max(peakActivePlanners, activePlanners);
+
+        if (startedProjectIds.length > 2) {
+          activePlanners -= 1;
+          return Promise.resolve();
+        }
+
+        return new Promise<void>((resolve) => {
+          releasePlanners.push(() => {
+            activePlanners -= 1;
+            resolve();
+          });
+        });
+      });
+
+      const pollPromise = pollAndProcess();
+      try {
+        await vi.waitFor(() => expect(releasePlanners).toHaveLength(2));
+        expect(new Set(startedProjectIds)).toEqual(
+          new Set(["global-cap-project-1", "global-cap-project-2"]),
+        );
+        expect(peakActivePlanners).toBeLessThanOrEqual(2);
+      } finally {
+        for (const release of releasePlanners) release();
+        await pollPromise;
+      }
 
-      expect(runPlanner).toHaveBeenCalledTimes(2);
       expect(getStageSemaphore().totalActive()).toBe(0);
     } finally {
       Object.assign(coordinatorEnv, previousLimits);
@@ -1743,6 +1796,57 @@ describe("coordinator", () => {
     expect(runReviewer).toHaveBeenCalledWith("ready-review-task", "/tmp/review");
   });
 
+  it("should serialize overlapping poll cycles to preserve stage order within a project", async () => {
+    const db = testDb.current;
+    db.insert(projects)
+      .values({
+        id: "overlap-project",
+        name: "Overlap",
+        rootPath: "/tmp/overlap",
+        parallelEnabled: true,
+      })
+      .run();
+    db.insert(tasks)
+      .values({
+        id: "overlap-planning-task",
+        projectId: "overlap-project",
+        title: "Slow planning",
+        status: "planning",
+      })
+      .run();
+    db.insert(tasks)
+      .values({
+        id: "overlap-review-task",
+        projectId: "overlap-project",
+        title: "Ready review",
+        status: "review",
+      })
+      .run();
+
+    let resolvePlanner: (() => void) | undefined;
+    vi.mocked(runPlanner).mockImplementationOnce(
+      () =>
+        new Promise<void>((resolve) => {
+          resolvePlanner = resolve;
+        }),
+    );
+
+    const firstPoll = pollAndProcess();
+    await vi.waitFor(() => expect(resolvePlanner).toBeTypeOf("function"));
+
+    const secondPoll = pollAndProcess();
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    const reviewerStartedBeforePlannerFinished = vi
+      .mocked(runReviewer)
+      .mock.calls.some(([taskId]) => taskId === "overlap-review-task");
+
+    resolvePlanner?.();
+    await Promise.all([firstPoll, secondPoll]);
+
+    expect(reviewerStartedBeforePlannerFinished).toBe(false);
+    expect(runReviewer).toHaveBeenCalledWith("overlap-review-task", "/tmp/overlap");
+  });
+
   it("should start one task in each independent project lane beyond the per-project task cap", async () => {
     const db = testDb.current;
     for (let i = 1; i <= 4; i++) {
@@ -1759,10 +1863,26 @@ describe("coordinator", () => {
         .run();
     }
 
-    await pollAndProcess();
+    const startedTaskIds: string[] = [];
+    const releasePlanners: Array<() => void> = [];
+    vi.mocked(runPlanner).mockImplementation(
+      (taskId) =>
+        new Promise<void>((resolve) => {
+          startedTaskIds.push(taskId);
+          releasePlanners.push(resolve);
+        }),
+    );
 
-    for (let i = 1; i <= 4; i++) {
-      expect(runPlanner).toHaveBeenCalledWith(`lane-task-${i}`, `/tmp/lane-${i}`);
+    const pollPromise = pollAndProcess();
+    try {
+      await vi.waitFor(() => expect(releasePlanners).toHaveLength(4));
+
+      for (let i = 1; i <= 4; i++) {
+        expect(startedTaskIds).toContain(`lane-task-${i}`);
+      }
+    } finally {
+      for (const release of releasePlanners) release();
+      await pollPromise;
     }
   });
 });

packages/agent/src/coordinator.ts

@@ -127,33 +127,82 @@ const PIPELINE: StatusTransition[] = [
 
 class StageSemaphore {
   private counts = new Map<string, number>();
+  private activeCount = 0;
+  private waiters: Array<{
+    key: string;
+    keyMax: number;
+    globalMax: number;
+    resolve: () => void;
+  }> = [];
+
+  private canAcquire(key: string, keyMax: number, globalMax: number): boolean {
+    const current = this.counts.get(key) ?? 0;
+    return current < keyMax && this.activeCount < globalMax;
+  }
 
   tryAcquire(key: string, keyMax: number, globalMax: number): boolean {
-    const current = this.counts.get(key) ?? 0;
-    if (current >= keyMax || this.totalActive() >= globalMax) return false;
-    this.counts.set(key, current + 1);
+    if (!this.canAcquire(key, keyMax, globalMax)) return false;
+    this.counts.set(key, (this.counts.get(key) ?? 0) + 1);
+    this.activeCount += 1;
     return true;
   }
 
+  acquire(key: string, keyMax: number, globalMax: number): Promise<void> {
+    if (this.tryAcquire(key, keyMax, globalMax)) {
+      return Promise.resolve();
+    }
+
+    return new Promise<void>((resolve) => {
+      this.waiters.push({ key, keyMax, globalMax, resolve });
+    });
+  }
+
   release(key: string): void {
     const current = this.counts.get(key) ?? 0;
-    this.counts.set(key, Math.max(0, current - 1));
-  }
+    if (current <= 0) return;
 
-  available(key: string, keyMax: number, globalMax: number): number {
-    const keyAvailable = keyMax - (this.counts.get(key) ?? 0);
-    const globalAvailable = globalMax - this.totalActive();
-    return Math.max(0, Math.min(keyAvailable, globalAvailable));
+    if (current === 1) {
+      this.counts.delete(key);
+    } else {
+      this.counts.set(key, current - 1);
+    }
+    this.activeCount -= 1;
+    this.drainWaiters();
   }
 
   totalActive(): number {
-    let total = 0;
-    for (const count of this.counts.values()) total += count;
-    return total;
+    return this.activeCount;
+  }
+
+  trackedKeyCount(): number {
+    return this.counts.size;
   }
 
   reset(): void {
+    if (this.waiters.length > 0) {
+      throw new Error("Cannot reset stage semaphore while acquisitions are queued");
+    }
     this.counts.clear();
+    this.activeCount = 0;
+  }
+
+  private drainWaiters(): void {
+    let granted = true;
+    while (granted) {
+      granted = false;
+      const waiterIndex = this.waiters.findIndex((waiter) =>
+        this.canAcquire(waiter.key, waiter.keyMax, waiter.globalMax),
+      );
+      if (waiterIndex < 0) return;
+
+      const [waiter] = this.waiters.splice(waiterIndex, 1);
+      if (!waiter) return;
+
+      this.counts.set(waiter.key, (this.counts.get(waiter.key) ?? 0) + 1);
+      this.activeCount += 1;
+      waiter.resolve();
+      granted = true;
+    }
   }
 }
 
@@ -815,7 +864,10 @@ export function processAutoQueueAdvance(): number {
 
 // ── Poll cycle ───────────────────────────────────────────────
 
-export async function pollAndProcess(): Promise<void> {
+let activePollPromise: Promise<void> | null = null;
+let followUpPollRequested = false;
+
+async function runPollCycle(): Promise<void> {
   log.debug("Starting poll cycle");
 
   // Release stale locks BEFORE watchdog — otherwise watchdog moves task to blocked_external
@@ -891,13 +943,8 @@ export async function pollAndProcess(): Promise<void> {
       const parallel = concurrency.parallel;
       const projectMax = concurrency.max;
       const stageKey = `${projectId}:${stage.label}`;
-      const available = stageSemaphore.available(stageKey, projectMax, globalMaxTasks);
-      if (available <= 0) {
-        log.debug({ stage: stage.label, projectId }, "Project stage at capacity, skipping");
-        continue;
-      }
 
-      const candidateWindow = Math.min(Math.max(available * 5, available), 50);
+      const candidateWindow = Math.min(Math.max(projectMax * 5, projectMax), 50);
       const candidates = findCoordinatorTaskCandidatesForProject(
         projectId,
         stage.label,
@@ -915,7 +962,6 @@ export async function pollAndProcess(): Promise<void> {
           projectId,
           candidateCount: candidates.length,
           candidateWindow,
-          available,
           projectMax,
           globalMaxTasks,
         },
@@ -967,17 +1013,14 @@ export async function pollAndProcess(): Promise<void> {
           continue;
         }
 
+        await stageSemaphore.acquire(stageKey, projectMax, globalMaxTasks);
+
         if (!claimTask(task.id, COORDINATOR_ID, CLAIM_LOCK_DURATION_MS)) {
+          stageSemaphore.release(stageKey);
           log.debug({ taskId: task.id, stage: stage.label }, "Task claim failed (already claimed)");
           continue;
         }
 
-        if (!stageSemaphore.tryAcquire(stageKey, projectMax, globalMaxTasks)) {
-          releaseTaskClaim(task.id);
-          log.debug({ stage: stage.label, projectId }, "Project stage semaphore full after claim");
-          break;
-        }
-
         log.debug(
           { stage: stage.label, taskId: task.id, candidateStatus: task.status, parallel },
           "Task claimed for processing",
@@ -1013,3 +1056,23 @@ export async function pollAndProcess(): Promise<void> {
 
   log.debug("Poll cycle complete");
 }
+
+export function pollAndProcess(): Promise<void> {
+  if (activePollPromise) {
+    followUpPollRequested = true;
+    log.debug("Poll cycle already active; queued one follow-up cycle");
+    return activePollPromise;
+  }
+
+  async function drainPollRequests(): Promise<void> {
+    do {
+      followUpPollRequested = false;
+      await runPollCycle();
+    } while (followUpPollRequested);
+  }
+
+  activePollPromise = drainPollRequests().finally(() => {
+    activePollPromise = null;
+  });
+  return activePollPromise;
+}