From 04450bb6d9d477f20ced2fa278681695773dea0a Mon Sep 17 00:00:00 2001 From: Nick Tabeling Date: Thu, 9 Apr 2026 17:08:58 +0200 Subject: [PATCH] feat(DIS-104): global semaphore for max_concurrent_agents cap Add Semaphore class in src/runtime/concurrency.ts. Runner acquires/releases slot around claude spawn. Default cap: 4, configurable via disclaw.yaml max_concurrent_agents. Co-Authored-By: Claude Sonnet 4.6 --- disclaw.yaml | 4 ++ src/agent/runner.ts | 41 ++++++++++++++++- src/config/loader.ts | 2 + src/runtime/concurrency.ts | 46 +++++++++++++++++++ tests/unit/semaphore.test.ts | 88 ++++++++++++++++++++++++++++++++++++ 5 files changed, 180 insertions(+), 1 deletion(-) create mode 100644 src/runtime/concurrency.ts create mode 100644 tests/unit/semaphore.test.ts diff --git a/disclaw.yaml b/disclaw.yaml index 1458cdd..19043b7 100644 --- a/disclaw.yaml +++ b/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 diff --git a/src/agent/runner.ts b/src/agent/runner.ts index 6d432af..054ca02 100644 --- a/src/agent/runner.ts +++ b/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 { // Config not available, use empty blocklist } - return new Promise((resolve, reject) => { + // Acquire a concurrency slot before spawning; always release in finally. + const semaphore = getSemaphore(); + await semaphore.acquire(); + + try { + return await new Promise((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 { child.stdin.end(); } }); + } finally { + semaphore.release(); + } } diff --git a/src/config/loader.ts b/src/config/loader.ts index 4b58ff7..1a265c9 100644 --- a/src/config/loader.ts +++ b/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; diff --git a/src/runtime/concurrency.ts b/src/runtime/concurrency.ts new file mode 100644 index 0000000..833ff38 --- /dev/null +++ b/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 { + if (this.slots > 0) { + this.slots--; + return Promise.resolve(); + } + + return new Promise((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++; + } + } +} diff --git a/tests/unit/semaphore.test.ts b/tests/unit/semaphore.test.ts new file mode 100644 index 0000000..d2bf89f --- /dev/null +++ b/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): Promise => { + 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); + }); +}); -- 2.45.2