From 86513816a06fbdf87dcec750ae4fccf0977af911 Mon Sep 17 00:00:00 2001 From: Marius Date: Thu, 9 Apr 2026 17:33:47 +0200 Subject: [PATCH] feat(agent-manager): add shared semaphore to cap concurrent git/gh processes (#8660) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(agent-manager): add shared semaphore to cap concurrent git/gh processes The agent manager's polling loops (GitStatsPoller, PRStatusPoller, diff watcher) could spawn unbounded concurrent git/gh child processes — one per worktree per tick. With many worktrees on Windows this causes process storms. Introduce a Semaphore(3) shared across GitOps, PRStatusPoller, and GitStatsPoller so at most 3 git/gh child processes run at any time. HTTP diffSummary calls are also gated since they trigger server-side git spawning. * fix(agent-manager): avoid nested semaphore acquisition in GitStatsPoller Gate only the HTTP diffSummary call through the semaphore, not the entire per-worktree lambda. The aheadBehind call goes through GitOps.raw() which already acquires the same semaphore — wrapping both would deadlock when all slots are held. --- .../src/agent-manager/AgentManagerProvider.ts | 6 +- .../kilo-vscode/src/agent-manager/GitOps.ts | 89 ++++++++------- .../src/agent-manager/GitStatsPoller.ts | 20 +++- .../src/agent-manager/PRStatusPoller.ts | 30 ++++-- .../src/agent-manager/pr-status-bridge.ts | 4 + .../src/agent-manager/semaphore.ts | 41 +++++++ .../kilo-vscode/tests/unit/git-ops.test.ts | 38 ++++++- .../tests/unit/git-stats-poller.test.ts | 52 +++++++++ .../kilo-vscode/tests/unit/semaphore.test.ts | 102 ++++++++++++++++++ 9 files changed, 327 insertions(+), 55 deletions(-) create mode 100644 packages/kilo-vscode/src/agent-manager/semaphore.ts create mode 100644 packages/kilo-vscode/tests/unit/semaphore.test.ts diff --git a/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts b/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts index 3089205232..1c6f285a53 100644 --- a/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts +++ b/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts @@ -25,6 +25,7 @@ import { continueInWorktree } from "./continue-in-worktree" import { shouldStopDiffPolling } from "./delete-worktree" import { buildKeybindingMap } from "./format-keybinding" import { resolveVersionModels, buildInitialMessages, type CreatedVersion } from "./multi-version" +import { Semaphore } from "./semaphore" import { PLATFORM } from "./constants" import type { AgentManagerOutMessage, AgentManagerInMessage } from "./types" import { hashFileDiffs, resolveLocalDiffTarget } from "../review-utils" @@ -76,11 +77,13 @@ export class AgentManagerProvider implements Disposable { (msg) => this.outputChannel.appendLine(`[SessionTerminal] ${msg}`), createTerminalHost(), ) - this.gitOps = new GitOps({ log: (...args) => this.log(...args) }) + const semaphore = new Semaphore(3) + this.gitOps = new GitOps({ log: (...args) => this.log(...args), semaphore }) this.statsPoller = new GitStatsPoller({ getWorktrees: () => this.state?.getWorktrees() ?? [], getWorkspaceRoot: () => this.getRoot(), getClient: () => this.connectionService.getClient(), + semaphore, onStats: (stats) => { const msg = { type: "agentManager.worktreeStats" as const, stats } this.cachedWorktreeStats = msg @@ -105,6 +108,7 @@ export class AgentManagerProvider implements Disposable { hasPersistedPR: (id: string) => !!this.state?.getWorktree(id)?.prNumber, openExternal: (u) => this.host.openExternal(u), log: (...a) => this.log(...a), + semaphore, }) } diff --git a/packages/kilo-vscode/src/agent-manager/GitOps.ts b/packages/kilo-vscode/src/agent-manager/GitOps.ts index 7f47ae46e5..3f36edb587 100644 --- a/packages/kilo-vscode/src/agent-manager/GitOps.ts +++ b/packages/kilo-vscode/src/agent-manager/GitOps.ts @@ -4,11 +4,14 @@ import * as fs from "fs/promises" import { spawn } from "../util/process" import simpleGit from "simple-git" import { parseWorktreeList, normalizePath } from "./git-import" +import type { Semaphore } from "./semaphore" interface GitOpsOptions { log: (...args: unknown[]) => void /** Override git command execution for testing. */ runGit?: (args: string[], cwd: string) => Promise + /** Shared concurrency gate for child process spawning. */ + semaphore?: Semaphore } export interface ApplyConflict { @@ -63,6 +66,7 @@ export class GitOps { private readonly log: (...args: unknown[]) => void private readonly runGit: (args: string[], cwd: string) => Promise private readonly controller = new AbortController() + private readonly semaphore: Semaphore | undefined get disposed(): boolean { return this.controller.signal.aborted @@ -70,6 +74,7 @@ export class GitOps { constructor(options: GitOpsOptions) { this.log = options.log + this.semaphore = options.semaphore this.runGit = options.runGit ?? ((args, cwd) => @@ -87,20 +92,22 @@ export class GitOps { private raw(args: string[], cwd: string): Promise { const signal = this.controller.signal if (signal.aborted) return Promise.reject(new Error("GitOps disposed")) - return new Promise((resolve, reject) => { - const onAbort = () => reject(new Error("GitOps disposed")) - signal.addEventListener("abort", onAbort, { once: true }) - this.runGit(args, cwd).then( - (value) => { - signal.removeEventListener("abort", onAbort) - resolve(value) - }, - (err) => { - signal.removeEventListener("abort", onAbort) - reject(err) - }, - ) - }) + const invoke = () => + new Promise((resolve, reject) => { + const onAbort = () => reject(new Error("GitOps disposed")) + signal.addEventListener("abort", onAbort, { once: true }) + this.runGit(args, cwd).then( + (value) => { + signal.removeEventListener("abort", onAbort) + resolve(value) + }, + (err) => { + signal.removeEventListener("abort", onAbort) + reject(err) + }, + ) + }) + return this.semaphore ? this.semaphore.run(invoke) : invoke() } /** Return the name of the currently checked-out branch, or `"HEAD"` if detached. */ @@ -413,37 +420,39 @@ export class GitOps { if (this.controller.signal.aborted) { return Promise.resolve({ code: 1, stdout: "", stderr: "GitOps disposed" }) } - return new Promise((resolve) => { - const child = spawn("git", args, { - cwd, - env: options?.env, - signal: this.controller.signal, - stdio: ["pipe", "pipe", "pipe"], - }) + const invoke = () => + new Promise((resolve) => { + const child = spawn("git", args, { + cwd, + env: options?.env, + signal: this.controller.signal, + stdio: ["pipe", "pipe", "pipe"], + }) - if (options?.stdin !== undefined) { - if (!child.stdin) { - resolve({ code: 1, stdout: "", stderr: "stdin not available for git process" }) - return + if (options?.stdin !== undefined) { + if (!child.stdin) { + resolve({ code: 1, stdout: "", stderr: "stdin not available for git process" }) + return + } + child.stdin.end(options.stdin) } - child.stdin.end(options.stdin) - } - const out: Buffer[] = [] - const err: Buffer[] = [] - child.stdout?.on("data", (chunk: Buffer) => out.push(chunk)) - child.stderr?.on("data", (chunk: Buffer) => err.push(chunk)) + const out: Buffer[] = [] + const err: Buffer[] = [] + child.stdout?.on("data", (chunk: Buffer) => out.push(chunk)) + child.stderr?.on("data", (chunk: Buffer) => err.push(chunk)) - child.on("error", (error) => { - resolve({ code: 1, stdout: "", stderr: error.message }) - }) - child.on("close", (code) => { - resolve({ - code: code ?? 1, - stdout: Buffer.concat(out).toString("utf8"), - stderr: Buffer.concat(err).toString("utf8"), + child.on("error", (error) => { + resolve({ code: 1, stdout: "", stderr: error.message }) + }) + child.on("close", (code) => { + resolve({ + code: code ?? 1, + stdout: Buffer.concat(out).toString("utf8"), + stderr: Buffer.concat(err).toString("utf8"), + }) }) }) - }) + return this.semaphore ? this.semaphore.run(invoke) : invoke() } } diff --git a/packages/kilo-vscode/src/agent-manager/GitStatsPoller.ts b/packages/kilo-vscode/src/agent-manager/GitStatsPoller.ts index 301261d137..2cde458d43 100644 --- a/packages/kilo-vscode/src/agent-manager/GitStatsPoller.ts +++ b/packages/kilo-vscode/src/agent-manager/GitStatsPoller.ts @@ -3,6 +3,7 @@ import * as path from "path" import type { KiloClient, FileDiff } from "@kilocode/sdk/v2/client" import { remoteRef, type Worktree } from "./WorktreeStateManager" import type { GitOps } from "./GitOps" +import type { Semaphore } from "./semaphore" import { normalizePath } from "./git-import" export interface WorktreeStats { @@ -45,6 +46,8 @@ interface GitStatsPollerOptions { onWorktreePresence?: (result: WorktreePresenceResult) => void log: (...args: unknown[]) => void intervalMs?: number + /** Shared concurrency gate for child process spawning. */ + semaphore?: Semaphore } export class GitStatsPoller { @@ -154,15 +157,20 @@ export class GitStatsPoller { return } + // Gate the HTTP diffSummary call through the semaphore but NOT the + // aheadBehind call — that goes through GitOps.raw() which already + // acquires the same semaphore. Wrapping both would deadlock. + const gate = this.options.semaphore + const diff = (dir: string, base: string) => { + const invoke = () => client.worktree.diffSummary({ directory: dir, base }, { throwOnError: true }) + return gate ? gate.run(invoke) : invoke() + } const stats = ( await Promise.all( active.map(async (wt) => { try { const base = remoteRef(wt) - const [{ data: diffs }, ab] = await Promise.all([ - client.worktree.diffSummary({ directory: wt.path, base }, { throwOnError: true }), - this.git.aheadBehind(wt.path, base), - ]) + const [{ data: diffs }, ab] = await Promise.all([diff(wt.path, base), this.git.aheadBehind(wt.path, base)]) const files = diffs.length const additions = diffs.reduce((sum: number, diff: FileDiff) => sum + diff.additions, 0) const deletions = diffs.reduce((sum: number, diff: FileDiff) => sum + diff.deletions, 0) @@ -260,8 +268,10 @@ export class GitStatsPoller { try { if (base && client) { this.options.log(`Local stats: using HTTP client with base=${base}`) + const gate = this.options.semaphore + const invoke = () => client.worktree.diffSummary({ directory: root, base }, { throwOnError: true }) const [{ data: diffs }, ab] = await Promise.all([ - client.worktree.diffSummary({ directory: root, base }, { throwOnError: true }), + gate ? gate.run(invoke) : invoke(), this.git.aheadBehind(root, base), ]) files = diffs.length diff --git a/packages/kilo-vscode/src/agent-manager/PRStatusPoller.ts b/packages/kilo-vscode/src/agent-manager/PRStatusPoller.ts index 6ec85c11ea..f655616b96 100644 --- a/packages/kilo-vscode/src/agent-manager/PRStatusPoller.ts +++ b/packages/kilo-vscode/src/agent-manager/PRStatusPoller.ts @@ -1,7 +1,9 @@ +import type { ExecFileOptionsWithStringEncoding } from "child_process" import type { Worktree } from "./WorktreeStateManager" import type { PRStatus, PRCheck, PRComment, CheckStatus, AggregateCheckStatus, PRState, ReviewDecision } from "./types" import { execWithShellEnv } from "./shell-env" import { classifyPRError } from "./git-import" +import type { Semaphore } from "./semaphore" interface PRStatusPollerOptions { getWorktrees: () => Worktree[] @@ -9,6 +11,8 @@ interface PRStatusPollerOptions { onStatus: (worktreeId: string, pr: PRStatus | null, error?: "gh_missing" | "gh_auth" | "fetch_failed") => void log: (...args: unknown[]) => void intervalMs?: number + /** Shared concurrency gate for child process spawning. */ + semaphore?: Semaphore } const GH_PROBE_TTL = 300_000 // 5 minutes — gh installation state rarely changes at runtime @@ -33,9 +37,21 @@ export class PRStatusPoller { private prCache = new Map() private lastFullSync = 0 // timestamp of last full (all-worktree) sync private readonly intervalMs: number + private readonly semaphore: Semaphore | undefined constructor(private readonly options: PRStatusPollerOptions) { this.intervalMs = options.intervalMs ?? 15_000 + this.semaphore = options.semaphore + } + + /** Run a command through the shared concurrency gate (when configured). */ + private shell( + cmd: string, + args: string[], + options?: Omit, + ): Promise<{ stdout: string; stderr: string }> { + const invoke = () => execWithShellEnv(cmd, args, options) + return this.semaphore ? this.semaphore.run(invoke) : invoke() } setEnabled(enabled: boolean): void { @@ -140,7 +156,7 @@ export class PRStatusPoller { return this.ghAvailable } try { - await execWithShellEnv("gh", ["--version"], { timeout: 5_000 }) + await this.shell("gh", ["--version"], { timeout: 5_000 }) this.ghAvailable = true } catch { this.ghAvailable = false @@ -279,7 +295,7 @@ export class PRStatusPoller { if (branch) args.push(branch) args.push("--json", PRStatusPoller.PR_JSON_FIELDS) - const { stdout } = await execWithShellEnv("gh", args, { cwd, timeout: 15_000 }) + const { stdout } = await this.shell("gh", args, { cwd, timeout: 15_000 }) return parsePRResult(stdout) } catch (err) { const msg = err instanceof Error ? err.message : String(err) @@ -291,11 +307,11 @@ export class PRStatusPoller { /** Search for PRs containing the current HEAD SHA. Finds PRs when branch name/tracking ref don't match. */ private async ghPRListBySHA(cwd: string): Promise { try { - const { stdout: sha } = await execWithShellEnv("git", ["rev-parse", "HEAD"], { cwd, timeout: 5_000 }) + const { stdout: sha } = await this.shell("git", ["rev-parse", "HEAD"], { cwd, timeout: 5_000 }) const head = sha.trim() if (!head) return null - const { stdout } = await execWithShellEnv( + const { stdout } = await this.shell( "gh", [ "pr", @@ -337,7 +353,7 @@ export class PRStatusPoller { items: PRCheck[] }> { try { - const { stdout } = await execWithShellEnv( + const { stdout } = await this.shell( "gh", ["pr", "checks", String(prNumber), "--json", "name,state,link,startedAt,completedAt"], { cwd, timeout: 15_000 }, @@ -375,7 +391,7 @@ export class PRStatusPoller { if (this.cachedRepo && this.cachedRepo.cwd === cwd) { return this.cachedRepo } - const { stdout } = await execWithShellEnv("gh", ["repo", "view", "--json", "owner,name"], { + const { stdout } = await this.shell("gh", ["repo", "view", "--json", "owner,name"], { cwd, timeout: 10_000, }) @@ -415,7 +431,7 @@ export class PRStatusPoller { } }` - const { stdout } = await execWithShellEnv( + const { stdout } = await this.shell( "gh", [ "api", diff --git a/packages/kilo-vscode/src/agent-manager/pr-status-bridge.ts b/packages/kilo-vscode/src/agent-manager/pr-status-bridge.ts index 15d5d754d7..03146dc49a 100644 --- a/packages/kilo-vscode/src/agent-manager/pr-status-bridge.ts +++ b/packages/kilo-vscode/src/agent-manager/pr-status-bridge.ts @@ -7,6 +7,7 @@ import type { Worktree } from "./WorktreeStateManager" import type { AgentManagerOutMessage, PRStatus } from "./types" import type { Disposable } from "./host" +import type { Semaphore } from "./semaphore" import { PRStatusPoller } from "./PRStatusPoller" interface PRBridgeHost { @@ -17,6 +18,7 @@ interface PRBridgeHost { hasPersistedPR(id: string): boolean openExternal(url: string): void log(...args: unknown[]): void + semaphore?: Semaphore } /** Minimal panel surface needed by the bridge (subset of PanelContext). */ @@ -43,6 +45,7 @@ export class PRStatusBridge { hasPersistedPR: (id: string) => boolean openExternal: (url: string) => void log: (...args: unknown[]) => void + semaphore?: Semaphore }): PRStatusBridge { return new PRStatusBridge(opts) } @@ -85,6 +88,7 @@ function bridgePollerOpts(bridge: PRStatusBridge, host: PRBridgeHost) { return { getWorktrees: () => host.getWorktrees(), getWorkspaceRoot: () => host.getWorkspaceRoot(), + semaphore: host.semaphore, onStatus: (id: string, pr: PRStatus | null, err?: "gh_missing" | "gh_auth" | "fetch_failed") => { if (err) { // Don't forward errors to the webview when we have prior PR data diff --git a/packages/kilo-vscode/src/agent-manager/semaphore.ts b/packages/kilo-vscode/src/agent-manager/semaphore.ts new file mode 100644 index 0000000000..37f3ab9cdd --- /dev/null +++ b/packages/kilo-vscode/src/agent-manager/semaphore.ts @@ -0,0 +1,41 @@ +/** + * Bounded-concurrency gate for git/gh child processes. + * + * Shared across GitOps and PRStatusPoller so that all polling loops + * (GitStatsPoller, PRStatusPoller, diff watcher) compete for the same + * slots. Prevents process storms when many worktrees are active. + */ +export class Semaphore { + private running = 0 + private readonly pending: (() => void)[] = [] + + constructor(private readonly limit: number) {} + + async run(fn: () => Promise): Promise { + await this.acquire() + try { + return await fn() + } finally { + this.release() + } + } + + private acquire(): Promise { + if (this.running < this.limit) { + this.running++ + return Promise.resolve() + } + return new Promise((resolve) => { + this.pending.push(() => { + this.running++ + resolve() + }) + }) + } + + private release(): void { + this.running-- + const next = this.pending.shift() + if (next) next() + } +} diff --git a/packages/kilo-vscode/tests/unit/git-ops.test.ts b/packages/kilo-vscode/tests/unit/git-ops.test.ts index 64f0b78c85..bd56efa827 100644 --- a/packages/kilo-vscode/tests/unit/git-ops.test.ts +++ b/packages/kilo-vscode/tests/unit/git-ops.test.ts @@ -3,9 +3,10 @@ import * as fs from "fs/promises" import * as os from "os" import * as nodePath from "path" import { GitOps } from "../../src/agent-manager/GitOps" +import { Semaphore } from "../../src/agent-manager/semaphore" -function ops(handler: (args: string[], cwd: string) => Promise): GitOps { - return new GitOps({ log: () => undefined, runGit: handler }) +function ops(handler: (args: string[], cwd: string) => Promise, semaphore?: Semaphore): GitOps { + return new GitOps({ log: () => undefined, runGit: handler, semaphore }) } function sleep(ms: number): Promise { @@ -556,4 +557,37 @@ describe("GitOps", () => { expect(git.disposed).toBe(true) }) }) + + describe("semaphore integration", () => { + it("limits concurrent raw() calls", async () => { + let running = 0 + let peak = 0 + const sem = new Semaphore(2) + const git = ops(async () => { + running++ + peak = Math.max(peak, running) + await sleep(10) + running-- + return "ok" + }, sem) + + await Promise.all(Array.from({ length: 6 }, () => git.currentBranch("/repo"))) + expect(peak).toBe(2) + }) + + it("works without a semaphore (no gating)", async () => { + let running = 0 + let peak = 0 + const git = ops(async () => { + running++ + peak = Math.max(peak, running) + await sleep(10) + running-- + return "ok" + }) + + await Promise.all(Array.from({ length: 4 }, () => git.currentBranch("/repo"))) + expect(peak).toBe(4) + }) + }) }) diff --git a/packages/kilo-vscode/tests/unit/git-stats-poller.test.ts b/packages/kilo-vscode/tests/unit/git-stats-poller.test.ts index 53176ffccb..2df698e3d0 100644 --- a/packages/kilo-vscode/tests/unit/git-stats-poller.test.ts +++ b/packages/kilo-vscode/tests/unit/git-stats-poller.test.ts @@ -5,6 +5,7 @@ import * as path from "path" import type { KiloClient } from "@kilocode/sdk/v2/client" import { GitStatsPoller, type WorktreePresenceResult } from "../../src/agent-manager/GitStatsPoller" import { GitOps } from "../../src/agent-manager/GitOps" +import { Semaphore } from "../../src/agent-manager/semaphore" import type { Worktree } from "../../src/agent-manager/WorktreeStateManager" function sleep(ms: number): Promise { @@ -430,4 +431,55 @@ describe("GitStatsPoller", () => { const fetches = commands.filter((cmd) => cmd[0] === "fetch") expect(fetches.length).toBe(0) }) + + it("limits concurrent diffSummary calls when semaphore is provided", async () => { + let running = 0 + let peak = 0 + let ticks = 0 + const sem = new Semaphore(2) + + const client = { + worktree: { + diffSummary: async () => { + running++ + peak = Math.max(peak, running) + await sleep(20) + running-- + return { data: diff(1, 0) } + }, + }, + } as unknown as KiloClient + + // Wire the SAME semaphore into GitOps to prove there's no deadlock — + // aheadBehind acquires the semaphore independently, not nested inside + // the diffSummary gate. + const wts = Array.from({ length: 5 }, (_, i) => worktree(String(i))) + const poller = new GitStatsPoller({ + getWorktrees: () => wts, + getWorkspaceRoot: () => undefined, + getClient: () => client, + onStats: () => { + ticks++ + }, + onLocalStats: () => undefined, + log: () => undefined, + intervalMs: 5, + semaphore: sem, + git: new GitOps({ + log: () => undefined, + semaphore: sem, + runGit: async (args) => { + if (args[0] === "rev-list" && args[1] === "--left-right") return "0\t0" + return "" + }, + }), + }) + + poller.setEnabled(true) + await waitFor(() => ticks >= 1) + poller.stop() + + // Only diffSummary calls are tracked — they should be bounded. + expect(peak).toBeLessThanOrEqual(2) + }) }) diff --git a/packages/kilo-vscode/tests/unit/semaphore.test.ts b/packages/kilo-vscode/tests/unit/semaphore.test.ts new file mode 100644 index 0000000000..697ba146b2 --- /dev/null +++ b/packages/kilo-vscode/tests/unit/semaphore.test.ts @@ -0,0 +1,102 @@ +import { describe, it, expect } from "bun:test" +import { Semaphore } from "../../src/agent-manager/semaphore" + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + +describe("Semaphore", () => { + it("runs tasks up to the concurrency limit", async () => { + const sem = new Semaphore(2) + let running = 0 + let peak = 0 + + const task = () => + sem.run(async () => { + running++ + peak = Math.max(peak, running) + await delay(50) + running-- + }) + + await Promise.all([task(), task(), task(), task(), task()]) + expect(peak).toBe(2) + expect(running).toBe(0) + }) + + it("returns the value produced by the function", async () => { + const sem = new Semaphore(1) + const result = await sem.run(async () => 42) + expect(result).toBe(42) + }) + + it("propagates rejections without blocking the queue", async () => { + const sem = new Semaphore(1) + const order: string[] = [] + + const failing = sem.run(async () => { + order.push("fail-start") + throw new Error("boom") + }) + + const passing = sem.run(async () => { + order.push("pass-start") + return "ok" + }) + + await expect(failing).rejects.toThrow("boom") + expect(await passing).toBe("ok") + expect(order).toEqual(["fail-start", "pass-start"]) + }) + + it("processes queued tasks in FIFO order", async () => { + const sem = new Semaphore(1) + const order: number[] = [] + + // First task holds the slot while 2 and 3 queue + const t1 = sem.run(async () => { + order.push(1) + await delay(50) + }) + const t2 = sem.run(async () => { + order.push(2) + }) + const t3 = sem.run(async () => { + order.push(3) + }) + + await Promise.all([t1, t2, t3]) + expect(order).toEqual([1, 2, 3]) + }) + + it("allows full concurrency when limit exceeds task count", async () => { + const sem = new Semaphore(10) + let running = 0 + let peak = 0 + + const task = () => + sem.run(async () => { + running++ + peak = Math.max(peak, running) + await delay(30) + running-- + }) + + await Promise.all([task(), task(), task()]) + expect(peak).toBe(3) + }) + + it("releases the slot on synchronous throw", async () => { + const sem = new Semaphore(1) + + await expect( + sem.run(() => { + throw new Error("sync") + }), + ).rejects.toThrow("sync") + + // Slot is free — next task should run immediately + const result = await sem.run(async () => "recovered") + expect(result).toBe("recovered") + }) +})