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 <noreply@anthropic.com>
This commit is contained in:
parent
4661892820
commit
04450bb6d9
5 changed files with 180 additions and 1 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
46
src/runtime/concurrency.ts
Normal file
46
src/runtime/concurrency.ts
Normal file
|
|
@ -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++;
|
||||
}
|
||||
}
|
||||
}
|
||||
88
tests/unit/semaphore.test.ts
Normal file
88
tests/unit/semaphore.test.ts
Normal file
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue