Merge pull request 'feat(DIS-104): global semaphore for max_concurrent_agents cap' (#44) from phase-1/concurrency-semaphore into main

Reviewed-on: #44

disclaw.yaml

@@ -11,3 +11,7 @@ claude_command: "claude"
 
 # Timeout in seconds for claude CLI processes (default: 120)
 claude_timeout_seconds: 120
+
+# Maximum number of Claude CLI processes that may run concurrently (default: 4)
+# Increase if your machine has more resources; lower to reduce memory pressure.
+# max_concurrent_agents: 4

src/agent/runner.ts

@@ -5,6 +5,37 @@ import { loadAgentIdentity } from "./identity";
 import { loadDisclawConfig } from "../config/loader";
 import { resolveClaude } from "../runtime/resolve-claude";
 import { sanitizedEnv } from "../runtime/env";
+import { Semaphore } from "../runtime/concurrency";
+
+// ---------------------------------------------------------------------------
+// Module-level semaphore — lazy-initialised on first runAgent() call so the
+// config is guaranteed to be loaded before we fix the concurrency cap.
+// ---------------------------------------------------------------------------
+let _semaphore: Semaphore | null = null;
+
+function getSemaphore(): Semaphore {
+  if (!_semaphore) {
+    let max = 4;
+    try {
+      const rootDir = path.resolve(__dirname, "../..");
+      const config = loadDisclawConfig(rootDir);
+      max = config.max_concurrent_agents;
+    } catch {
+      // Config not available — fall back to default
+    }
+    _semaphore = new Semaphore(max);
+  }
+  return _semaphore;
+}
+
+/**
+ * Exposed only for testing — resets the module-level semaphore so unit tests
+ * can inject a fresh instance with a known concurrency cap.
+ * @internal
+ */
+export function _resetSemaphore(max?: number): void {
+  _semaphore = max !== undefined ? new Semaphore(max) : null;
+}
 
 export interface RunAgentOptions {
   workspacePath: string;
@@ -162,7 +193,12 @@ export async function runAgent(options: RunAgentOptions): Promise<string> {
     // Config not available, use empty blocklist
   }
 
-  return new Promise<string>((resolve, reject) => {
+  // Acquire a concurrency slot before spawning; always release in finally.
+  const semaphore = getSemaphore();
+  await semaphore.acquire();
+
+  try {
+  return await new Promise<string>((resolve, reject) => {
     // NOTE: --bare is intentionally NOT used here. --bare disables OAuth/keychain
     // auth and requires ANTHROPIC_API_KEY, which breaks Claude Pro/Max subscriptions.
     // CLAUDE.md isolation is handled structurally: workspaces live under
@@ -266,4 +302,7 @@ export async function runAgent(options: RunAgentOptions): Promise<string> {
       child.stdin.end();
     }
   });
+  } finally {
+    semaphore.release();
+  }
 }

src/config/loader.ts

@@ -10,6 +10,7 @@ export interface DisclawConfig {
   claude_command: string;
   claude_timeout_seconds: number;
   env_blocklist: string[];
+  max_concurrent_agents: number;
 }
 
 export interface EnvConfig {
@@ -86,6 +87,7 @@ export function loadDisclawConfig(rootDir: string): DisclawConfig {
     claude_command: parsed.claude_command ?? "claude",
     claude_timeout_seconds: parsed.claude_timeout_seconds ?? 120,
     env_blocklist: parsed.env_blocklist ?? [],
+    max_concurrent_agents: parsed.max_concurrent_agents ?? 4,
   };
 
   return config;

src/runtime/concurrency.ts

@@ -0,0 +1,46 @@
+/**
+ * A simple Promise-based counting semaphore.
+ *
+ * Used to cap the number of concurrently running Claude CLI processes.
+ * Callers acquire a slot before spawning and release it when done (via finally).
+ */
+export class Semaphore {
+  private slots: number;
+  private queue: Array<() => void> = [];
+
+  constructor(max: number) {
+    if (max < 1) {
+      throw new RangeError("Semaphore max must be at least 1");
+    }
+    this.slots = max;
+  }
+
+  /**
+   * Acquire a slot. Resolves immediately when a slot is free;
+   * otherwise queues the caller until a slot becomes available.
+   */
+  acquire(): Promise<void> {
+    if (this.slots > 0) {
+      this.slots--;
+      return Promise.resolve();
+    }
+
+    return new Promise<void>((resolve) => {
+      this.queue.push(resolve);
+    });
+  }
+
+  /**
+   * Release a previously acquired slot.
+   * If callers are queued, the next one is unblocked immediately.
+   */
+  release(): void {
+    const next = this.queue.shift();
+    if (next) {
+      // Hand the slot directly to the next waiter — slot count stays the same.
+      next();
+    } else {
+      this.slots++;
+    }
+  }
+}

tests/unit/semaphore.test.ts

@@ -0,0 +1,88 @@
+import { describe, it, expect } from "vitest";
+import { Semaphore } from "../../src/runtime/concurrency";
+
+describe("Semaphore", () => {
+  it("(a) acquire resolves immediately when slots are free", async () => {
+    const sem = new Semaphore(2);
+
+    // Both acquires should resolve without needing any release
+    await expect(sem.acquire()).resolves.toBeUndefined();
+    await expect(sem.acquire()).resolves.toBeUndefined();
+  });
+
+  it("(b) third acquire queues and waits until a slot is released", async () => {
+    const sem = new Semaphore(2);
+
+    // Fill both slots
+    await sem.acquire();
+    await sem.acquire();
+
+    let thirdResolved = false;
+    const third = sem.acquire().then(() => {
+      thirdResolved = true;
+    });
+
+    // Allow microtasks to flush — third should still be waiting
+    await Promise.resolve();
+    expect(thirdResolved).toBe(false);
+
+    // Release one slot — third should now resolve
+    sem.release();
+    await third;
+    expect(thirdResolved).toBe(true);
+  });
+
+  it("(c) release unblocks a queued waiter in FIFO order", async () => {
+    const sem = new Semaphore(1);
+
+    await sem.acquire(); // slot now occupied
+
+    const order: number[] = [];
+
+    const p1 = sem.acquire().then(() => order.push(1));
+    const p2 = sem.acquire().then(() => order.push(2));
+    const p3 = sem.acquire().then(() => order.push(3));
+
+    // Release three times — each wakes one waiter in turn
+    sem.release();
+    await p1;
+    sem.release();
+    await p2;
+    sem.release();
+    await p3;
+
+    expect(order).toEqual([1, 2, 3]);
+  });
+
+  it("(d) error inside critical section still releases the slot (via finally)", async () => {
+    const sem = new Semaphore(1);
+
+    // Simulate the try/finally pattern used in runner.ts
+    const runWithSemaphore = async (task: () => Promise<void>): Promise<void> => {
+      await sem.acquire();
+      try {
+        await task();
+      } finally {
+        sem.release();
+      }
+    };
+
+    // This task throws — the slot must still be freed
+    await expect(
+      runWithSemaphore(async () => {
+        throw new Error("task failed");
+      })
+    ).rejects.toThrow("task failed");
+
+    // If release was called, the next acquire should resolve immediately
+    let resolved = false;
+    await sem.acquire().then(() => {
+      resolved = true;
+    });
+    expect(resolved).toBe(true);
+  });
+
+  it("(e) constructor rejects max < 1", () => {
+    expect(() => new Semaphore(0)).toThrow(RangeError);
+  });
+});