Compare commits
No commits in common. "86683297affb9446becf1dca8a85290abddd291a" and "0a4cd53b711e4b9ca2fba761c8effc65cc874b75" have entirely different histories.
86683297af
...
0a4cd53b71
5 changed files with 1 additions and 180 deletions
|
|
@ -11,7 +11,3 @@ claude_command: "claude"
|
||||||
|
|
||||||
# Timeout in seconds for claude CLI processes (default: 120)
|
# Timeout in seconds for claude CLI processes (default: 120)
|
||||||
claude_timeout_seconds: 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,37 +5,6 @@ import { loadAgentIdentity } from "./identity";
|
||||||
import { loadDisclawConfig } from "../config/loader";
|
import { loadDisclawConfig } from "../config/loader";
|
||||||
import { resolveClaude } from "../runtime/resolve-claude";
|
import { resolveClaude } from "../runtime/resolve-claude";
|
||||||
import { sanitizedEnv } from "../runtime/env";
|
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 {
|
export interface RunAgentOptions {
|
||||||
workspacePath: string;
|
workspacePath: string;
|
||||||
|
|
@ -193,12 +162,7 @@ export async function runAgent(options: RunAgentOptions): Promise<string> {
|
||||||
// Config not available, use empty blocklist
|
// Config not available, use empty blocklist
|
||||||
}
|
}
|
||||||
|
|
||||||
// Acquire a concurrency slot before spawning; always release in finally.
|
return new Promise<string>((resolve, reject) => {
|
||||||
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
|
// NOTE: --bare is intentionally NOT used here. --bare disables OAuth/keychain
|
||||||
// auth and requires ANTHROPIC_API_KEY, which breaks Claude Pro/Max subscriptions.
|
// auth and requires ANTHROPIC_API_KEY, which breaks Claude Pro/Max subscriptions.
|
||||||
// CLAUDE.md isolation is handled structurally: workspaces live under
|
// CLAUDE.md isolation is handled structurally: workspaces live under
|
||||||
|
|
@ -302,7 +266,4 @@ export async function runAgent(options: RunAgentOptions): Promise<string> {
|
||||||
child.stdin.end();
|
child.stdin.end();
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
} finally {
|
|
||||||
semaphore.release();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -10,7 +10,6 @@ export interface DisclawConfig {
|
||||||
claude_command: string;
|
claude_command: string;
|
||||||
claude_timeout_seconds: number;
|
claude_timeout_seconds: number;
|
||||||
env_blocklist: string[];
|
env_blocklist: string[];
|
||||||
max_concurrent_agents: number;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface EnvConfig {
|
export interface EnvConfig {
|
||||||
|
|
@ -87,7 +86,6 @@ export function loadDisclawConfig(rootDir: string): DisclawConfig {
|
||||||
claude_command: parsed.claude_command ?? "claude",
|
claude_command: parsed.claude_command ?? "claude",
|
||||||
claude_timeout_seconds: parsed.claude_timeout_seconds ?? 120,
|
claude_timeout_seconds: parsed.claude_timeout_seconds ?? 120,
|
||||||
env_blocklist: parsed.env_blocklist ?? [],
|
env_blocklist: parsed.env_blocklist ?? [],
|
||||||
max_concurrent_agents: parsed.max_concurrent_agents ?? 4,
|
|
||||||
};
|
};
|
||||||
|
|
||||||
return config;
|
return config;
|
||||||
|
|
|
||||||
|
|
@ -1,46 +0,0 @@
|
||||||
/**
|
|
||||||
* 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++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,88 +0,0 @@
|
||||||
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