Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
3.0 kB · 106 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107/** * Tests for the async Semaphore used to bound subagent kernel-boot concurrency. */
import { describe, expect, it } from "vitest";import { Semaphore } from "../src/utils/semaphore.js";
function deferred<T>(): { promise: Promise<T>; resolve: (value: T) => void } { let resolve!: (value: T) => void; const promise = new Promise<T>((r) => { resolve = r; }); return { promise, resolve };}
describe("Semaphore", () => { it("rejects non-positive permit counts", () => { expect(() => new Semaphore(0)).toThrow(); expect(() => new Semaphore(-1)).toThrow(); expect(() => new Semaphore(1.5)).toThrow(); });
it("never exceeds the permit count under heavy fan-out", async () => { const limit = 3; const total = 50; const sem = new Semaphore(limit); let active = 0; let peak = 0; let completed = 0;
await Promise.all( Array.from({ length: total }, () => sem.run(async () => { active += 1; peak = Math.max(peak, active); await new Promise((r) => setTimeout(r, 1)); active -= 1; completed += 1; }), ), );
expect(peak).toBeLessThanOrEqual(limit); expect(completed).toBe(total); });
it("releases the permit when the task throws", async () => { const sem = new Semaphore(1); await expect(sem.run(async () => Promise.reject(new Error("boom")))).rejects.toThrow("boom"); // If the permit leaked, this second acquire would hang forever. await expect(sem.run(async () => "ok")).resolves.toBe("ok"); });
it("hands permits to waiters in FIFO order", async () => { const sem = new Semaphore(1); const order: number[] = []; const gate = deferred<void>();
// Holder occupies the single permit until we release `gate`. const holder = sem.run(async () => { await gate.promise; });
// Queue three waiters; they should run in the order they queued. const waiters = [0, 1, 2].map((i) => sem.run(async () => { order.push(i); }), );
expect(sem.queueLength).toBe(3); gate.resolve(); await Promise.all([holder, ...waiters]); expect(order).toEqual([0, 1, 2]); });
it("a queued waiter aborts out of the queue without consuming a permit", async () => { const sem = new Semaphore(1); const gate = deferred<void>(); const holder = sem.run(async () => { await gate.promise; });
const ac = new AbortController(); const queued = sem.run(async () => "ran", ac.signal); expect(sem.queueLength).toBe(1);
ac.abort(); await expect(queued).rejects.toBeTruthy(); expect(sem.queueLength).toBe(0);
// The aborted waiter must not have stolen the permit: once the holder // releases, a fresh acquire runs immediately. gate.resolve(); await holder; await expect(sem.run(async () => "ok")).resolves.toBe("ok"); });
it("rejects immediately if the signal is already aborted", async () => { const sem = new Semaphore(1); const ac = new AbortController(); ac.abort(); await expect(sem.run(async () => "ran", ac.signal)).rejects.toBeTruthy(); await expect(sem.run(async () => "ok")).resolves.toBe("ok"); });});