smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(task): scoped provider concurrency cap to each LLM turn

Browse Source 
The per-provider semaphore (e.g. `providers.ollama-cloud.maxConcurrency`) was acquired before `SessionManager.open` and released only after `driveSessionToYield` returned, so it bracketed the whole subagent lifecycle. Any spawn tree wider than `maxConcurrency` deadlocked: parents held every slot while waiting for children that were queued on the same cap — symptoms matched zero LLM requests and tokens=0/requests=0 cancellations.

Moved the bracket into a `StreamFn` wrapper. The wrapper acquires the slot just before each provider HTTP request and releases it the moment the response stream produces 'done'/'error', so a parent's slot is free between turns and child subagents can acquire while their parent's tool calls run. Wraps both the main agent and the advisor (both consume `settingsAwareStreamFn`).

Fixes #3749
This commit is contained in:
roboomp
2026-06-28 20:36:40 +00:00
parent c178ceaea5
commit febbc26f28
6 changed files with 350 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/coding-agent/CHANGELOG.md
+1
View File
@@ -5,6 +5,7 @@
### Fixed
- Fixed recoverable context-overflow compaction keeping the failed assistant error turn in visible session history after scheduling the retry. ([#3747](https://github.com/can1357/oh-my-pi/issues/3747))
- Fixed the per-provider concurrency cap (e.g. `providers.ollama-cloud.maxConcurrency`) bracketing the whole subagent lifecycle, which deadlocked any spawn tree wider than the cap because parents held every slot while waiting for children queued on the same cap. The semaphore now wraps each provider HTTP request, so a parent's slot frees between turns and child subagents can acquire while their parent's tool calls are running. ([#3749](https://github.com/can1357/oh-my-pi/issues/3749))
## [16.2.4] - 2026-06-28
packages/coding-agent/src/sdk.ts
+8 -1
View File
@@ -130,6 +130,7 @@ import {
loadProjectContextFiles as loadContextFilesInternal,
} from "./system-prompt";
import { AgentOutputManager } from "./task/output-manager";
import { wrapStreamFnWithProviderConcurrency } from "./task/provider-concurrency";
import {
AUTO_THINKING,
type ConfiguredThinkingLevel,
@@ -2539,7 +2540,13 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {}
// the advisor (via AgentSessionConfig.streamFn). Keeps OpenRouter
// sticky-routing variants, antigravity endpoint routing, in-flight caps,
// and the loop guard consistent across every agent the session drives.
const settingsAwareStreamFn = createSettingsAwareStreamFn(settings);
// Wrapped in a per-provider concurrency limiter so each LLM HTTP request
// — not the whole subagent lifecycle — holds the slot, preventing the
// nested-spawn deadlock from issue #3749.
const settingsAwareStreamFn = wrapStreamFnWithProviderConcurrency(
settings,
createSettingsAwareStreamFn(settings),
);
agent = new Agent({
initialState: {
systemPrompt,
packages/coding-agent/src/task/executor.ts
+4 -62
View File
@@ -56,7 +56,6 @@ import { ToolAbortError } from "../tools/tool-errors";
import type { EventBus } from "../utils/event-bus";
import { buildNamedToolChoice } from "../utils/tool-choice";
import type { WorkspaceTree } from "../workspace-tree";
import { Semaphore } from "./parallel";
import { subprocessToolRegistry } from "./subprocess-tool-registry";
import {
type AgentDefinition,
@@ -199,51 +198,6 @@ function installSubagentRetryFallbackChain(args: {
return role;
}
const PROVIDER_MAX_CONCURRENCY_SETTINGS: Record<string, SettingPath> = {
"ollama-cloud": "providers.ollama-cloud.maxConcurrency",
};
interface ProviderSemaphoreEntry {
limit: number;
semaphore: Semaphore;
}
const providerSemaphores = new Map<string, ProviderSemaphoreEntry>();
/**
* Resolve the configured concurrency ceiling for a provider, or `undefined`
* when the provider has no cap concept at all. A configured value `<= 0` means
* "unlimited" and maps to `Infinity` — still a tracked ceiling, so every run
* holds a slot and a later finite resize counts work started while unlimited.
*/
function getProviderConcurrencyLimit(settings: Settings, provider: string): number | undefined {
const settingPath = PROVIDER_MAX_CONCURRENCY_SETTINGS[provider];
if (!settingPath) return undefined;
const raw = settings.get(settingPath);
const limit = Number.isFinite(raw) ? Math.trunc(raw) : 0;
return limit > 0 ? limit : Number.POSITIVE_INFINITY;
}
function getProviderSemaphore(settings: Settings, provider: string): Semaphore | undefined {
const limit = getProviderConcurrencyLimit(settings, provider);
if (limit === undefined) return undefined;
// Always hand out (and acquire on) the single shared limiter, even when
// unlimited (Infinity). Resizing it in place — rather than replacing it —
// keeps every in-flight slot counted, so a runtime or mixed limit change can
// never push concurrency past the cap (issue #3464 review feedback).
const existing = providerSemaphores.get(provider);
if (existing) {
if (existing.limit !== limit) {
existing.limit = limit;
existing.semaphore.resize(limit);
}
return existing.semaphore;
}
const semaphore = new Semaphore(limit);
providerSemaphores.set(provider, { limit, semaphore });
return semaphore;
}
function renderIrcPeerRoster(selfId: string): string {
const peers = AgentRegistry.global()
.list()
@@ -2028,8 +1982,6 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
let sessionOpenedAt: number | undefined;
let sessionCreatedAt: number | undefined;
let readyAt: number | undefined;
let providerSemaphore: Semaphore | undefined;
let providerSemaphoreAcquired = false;
try {
checkAbort();
@@ -2098,13 +2050,6 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
? resolvedThinkingLevel
: (thinkingLevel ?? resolvedThinkingLevel);
resolvedAt = performance.now();
if (model) {
providerSemaphore = getProviderSemaphore(settings, model.provider);
if (providerSemaphore) {
await providerSemaphore.acquire(abortSignal);
providerSemaphoreAcquired = true;
}
}
const effectiveCwd = worktree ?? cwd;
const sessionManager = sessionFile
@@ -2395,10 +2340,6 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
}
if (exitCode === 0) exitCode = 1;
}
if (providerSemaphoreAcquired) {
providerSemaphore?.release();
providerSemaphoreAcquired = false;
}
sessionAbortController.abort();
try {
await untilAborted(AbortSignal.timeout(5000), () => monitor.waitForActiveSessionAbort());
@@ -2429,9 +2370,10 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
}
// Launch-latency breakdown (subagent invocation → first chat dispatch).
// Phase deltas are performance.now() spans; the semaphore brackets use the
// Date.now epochs captured by the spawn site (invokedAt before acquire,
// acquiredAt after) so queue wait and pre-run setup are reported apart.
// Phase deltas are performance.now() spans; the task-tool concurrency
// brackets use the Date.now epochs captured by the spawn site
// (invokedAt before acquire, acquiredAt after) so queue wait and
// pre-run setup are reported apart.
const span = (from: number | undefined, to: number | undefined): number | undefined =>
from !== undefined && to !== undefined ? Math.round(to - from) : undefined;
const queueMs =
packages/coding-agent/src/task/provider-concurrency.ts
+100
View File
@@ -0,0 +1,100 @@
/**
* Per-provider LLM concurrency cap, applied around each provider HTTP request.
*
* The semaphore brackets only the streaming request itself, not the whole
* agent lifetime: a parent subagent releases its slot the moment its LLM
* stream finishes producing, so children spawned during tool execution can
* acquire slots for their own turns. Holding the slot across the parent's
* full conversation deadlocks any spawn tree whose width exceeds
* `maxConcurrency` because the parents wait for children that wait for
* slots the parents are holding (issue
* [#3749](https://github.com/can1357/oh-my-pi/issues/3749)).
*/
import type { StreamFn } from "@oh-my-pi/pi-agent-core";
import type { Settings } from "../config/settings";
import type { SettingPath } from "../config/settings-schema";
import { Semaphore } from "./parallel";
const PROVIDER_MAX_CONCURRENCY_SETTINGS: Record<string, SettingPath> = {
"ollama-cloud": "providers.ollama-cloud.maxConcurrency",
};
interface ProviderSemaphoreEntry {
limit: number;
semaphore: Semaphore;
}
const providerSemaphores = new Map<string, ProviderSemaphoreEntry>();
/**
* Resolve the configured concurrency ceiling for a provider, or `undefined`
* when the provider has no cap concept at all. A configured value `<= 0` means
* "unlimited" and maps to `Infinity` — still a tracked ceiling, so every run
* holds a slot and a later finite resize counts work started while unlimited.
*/
export function getProviderConcurrencyLimit(settings: Settings, provider: string): number | undefined {
const settingPath = PROVIDER_MAX_CONCURRENCY_SETTINGS[provider];
if (!settingPath) return undefined;
const raw = settings.get(settingPath);
const limit = Number.isFinite(raw) ? Math.trunc(raw) : 0;
return limit > 0 ? limit : Number.POSITIVE_INFINITY;
}
/**
* Hand out the single shared limiter for `provider` (creating one lazily) and
* resize it in place when the configured limit changes. Replacing the
* semaphore would orphan in-flight slots on the old instance and let a
* runtime or mixed limit value exceed the cap (issue #3464 review feedback).
*/
export function getProviderSemaphore(settings: Settings, provider: string): Semaphore | undefined {
const limit = getProviderConcurrencyLimit(settings, provider);
if (limit === undefined) return undefined;
const existing = providerSemaphores.get(provider);
if (existing) {
if (existing.limit !== limit) {
existing.limit = limit;
existing.semaphore.resize(limit);
}
return existing.semaphore;
}
const semaphore = new Semaphore(limit);
providerSemaphores.set(provider, { limit, semaphore });
return semaphore;
}
/**
* Wrap a {@link StreamFn} so every LLM HTTP request acquires the provider's
* concurrency slot before the request goes out and releases it when the
* stream finishes producing (success, error, or abort). Providers without a
* configured cap pass straight through.
*
* The acquire bracket is intentionally narrow (one slot per LLM call), so
* spawn trees deeper than `maxConcurrency` no longer deadlock on themselves —
* see the module-level comment for the failure mode this fixes.
*/
export function wrapStreamFnWithProviderConcurrency(settings: Settings, base: StreamFn): StreamFn {
return async (model, context, options) => {
const semaphore = getProviderSemaphore(settings, model.provider);
if (!semaphore) return base(model, context, options);
await semaphore.acquire(options?.signal);
let released = false;
const release = () => {
if (released) return;
released = true;
semaphore.release();
};
try {
const stream = await base(model, context, options);
// EventStream.result() settles when the producer pushes 'done'/'error'
// or calls fail() — i.e. once the provider has finished producing.
// Releasing here keeps the slot held for the network request and
// nothing else.
stream.result().then(release, release);
return stream;
} catch (err) {
release();
throw err;
}
};
}
packages/coding-agent/test/issue-3464-ollama-cloud-task-backoff.test.ts
+69 -119
View File
@@ -1,31 +1,19 @@
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "bun:test";
import * as path from "node:path";
import { Agent } from "@oh-my-pi/pi-agent-core";
import type { Model } from "@oh-my-pi/pi-ai";
import { Agent, type StreamFn } from "@oh-my-pi/pi-agent-core";
import type { AssistantMessage, Model } from "@oh-my-pi/pi-ai";
import { createMockModel } from "@oh-my-pi/pi-ai/providers/mock";
import { AssistantMessageEventStream } from "@oh-my-pi/pi-ai/utils/event-stream";
import { type GeneratedProvider, getBundledModel } from "@oh-my-pi/pi-catalog/models";
import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry";
import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings";
import type { LoadExtensionsResult } from "@oh-my-pi/pi-coding-agent/extensibility/extensions/types";
import type { CreateAgentSessionResult } from "@oh-my-pi/pi-coding-agent/sdk";
import * as sdkModule from "@oh-my-pi/pi-coding-agent/sdk";
import {
AgentSession,
type AgentSessionEvent,
type PromptOptions,
} from "@oh-my-pi/pi-coding-agent/session/agent-session";
import { AgentSession } from "@oh-my-pi/pi-coding-agent/session/agent-session";
import { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage";
import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager";
import { runSubprocess } from "@oh-my-pi/pi-coding-agent/task/executor";
import { Semaphore } from "@oh-my-pi/pi-coding-agent/task/parallel";
import type { AgentDefinition } from "@oh-my-pi/pi-coding-agent/task/types";
import { EventBus } from "@oh-my-pi/pi-coding-agent/utils/event-bus";
import { wrapStreamFnWithProviderConcurrency } from "@oh-my-pi/pi-coding-agent/task/provider-concurrency";
import { TempDir } from "@oh-my-pi/pi-utils";
type MockPromptSession = AgentSession & {
emit(event: AgentSessionEvent): void;
};
interface Deferred {
promise: Promise<void>;
resolve: () => void;
@@ -36,64 +24,12 @@ function deferred(): Deferred {
return { promise, resolve };
}
function createSessionResult(session: AgentSession): CreateAgentSessionResult {
return {
session,
extensionsResult: { extensions: [], errors: [], runtime: {} as unknown } as LoadExtensionsResult,
setToolUIContext: () => {},
eventBus: new EventBus(),
};
}
function createGateSession(onPrompt: () => Promise<void>): MockPromptSession {
const listeners: Array<(event: AgentSessionEvent) => void> = [];
const session = {
agent: { state: { systemPrompt: ["test"] } },
state: { messages: [] },
extensionRunner: undefined,
sessionManager: { appendSessionInit: () => {} },
getActiveToolNames: () => ["yield"],
setActiveToolsByName: async () => {},
subscribe: (listener: (event: AgentSessionEvent) => void) => {
listeners.push(listener);
return () => {};
},
prompt: async (_text: string, _options?: PromptOptions) => {
await onPrompt();
for (const listener of listeners) {
listener({
type: "tool_execution_end",
toolCallId: "tool-yield",
toolName: "yield",
result: { content: [{ type: "text", text: "Result submitted." }], details: { status: "success" } },
isError: false,
});
}
},
waitForIdle: async () => {},
getLastAssistantMessage: () => undefined,
abort: async () => {},
dispose: async () => {},
emit: (event: AgentSessionEvent) => {
for (const listener of listeners) listener(event);
},
};
return session as unknown as MockPromptSession;
}
function requireModel(provider: GeneratedProvider, id: string): Model {
const model = getBundledModel(provider, id);
if (!model) throw new Error(`Expected bundled model ${provider}/${id}`);
return model;
}
const taskAgent: AgentDefinition = {
name: "task",
description: "General task agent",
systemPrompt: "test",
source: "bundled",
};
describe("issue #3464: ollama-cloud task backoff", () => {
let tempDir: TempDir;
let authStorage: AuthStorage;
@@ -161,62 +97,76 @@ describe("issue #3464: ollama-cloud task backoff", () => {
expect(session.model?.id).toBe(fallback.id);
});
it("bounds concurrent subagent runs by the resolved ollama-cloud provider limit", async () => {
it("bounds concurrent ollama-cloud LLM streams by the configured maxConcurrency", async () => {
const cloudModel = requireModel("ollama-cloud", "gpt-oss:120b");
const started: string[] = [];
const gates = new Map<string, Deferred>();
const firstStarted = deferred();
const secondStarted = deferred();
vi.spyOn(sdkModule, "createAgentSession").mockImplementation(async options => {
const id = options?.agentId ?? "unknown";
const gate = deferred();
gates.set(id, gate);
return createSessionResult(
createGateSession(async () => {
started.push(id);
if (id === "CloudOne") firstStarted.resolve();
if (id === "CloudTwo") secondStarted.resolve();
await gate.promise;
}),
);
});
const settings = Settings.isolated({
"providers.ollama-cloud.maxConcurrency": 1,
"providers.ollama-cloud.maxConcurrency": 2,
});
const first = runSubprocess({
cwd: "/tmp",
agent: taskAgent,
task: "first",
index: 0,
id: "CloudOne",
modelOverride: `${cloudModel.provider}/${cloudModel.id}`,
settings,
modelRegistry,
enableLsp: false,
});
const second = runSubprocess({
cwd: "/tmp",
agent: taskAgent,
task: "second",
index: 1,
id: "CloudTwo",
modelOverride: `${cloudModel.provider}/${cloudModel.id}`,
settings,
modelRegistry,
enableLsp: false,
});
let inFlight = 0;
let peakInFlight = 0;
let invocations = 0;
const gates: Deferred[] = [];
const base: StreamFn = model => {
const gate = deferred();
gates.push(gate);
invocations++;
inFlight++;
peakInFlight = Math.max(peakInFlight, inFlight);
const stream = new AssistantMessageEventStream();
void gate.promise.then(() => {
inFlight--;
const message: AssistantMessage = {
role: "assistant",
content: [],
api: model.api,
provider: model.provider,
model: model.id,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now(),
};
stream.push({ type: "done", reason: "stop", message });
stream.end();
});
return stream;
};
const wrapped = wrapStreamFnWithProviderConcurrency(settings, base);
await firstStarted.promise;
expect(started).toEqual(["CloudOne"]);
expect(gates.has("CloudTwo")).toBe(false);
const waitForInvocations = async (target: number): Promise<void> => {
for (let i = 0; i < 1000 && invocations < target; i++) {
await Promise.resolve();
}
expect(invocations).toBe(target);
};
gates.get("CloudOne")?.resolve();
await first;
await secondStarted.promise;
expect(started).toEqual(["CloudOne", "CloudTwo"]);
gates.get("CloudTwo")?.resolve();
await second;
const calls = Array.from({ length: 4 }, async () => wrapped(cloudModel, { messages: [] }, {}));
// Two slots admit two calls; the next two queue behind them.
await waitForInvocations(2);
expect(inFlight).toBe(2);
// Release the first slot; exactly one queued waiter is admitted.
gates[0]!.resolve();
await waitForInvocations(3);
expect(inFlight).toBe(2);
expect(peakInFlight).toBe(2);
gates[1]!.resolve();
await waitForInvocations(4);
expect(inFlight).toBe(2);
gates[2]!.resolve();
gates[3]!.resolve();
await Promise.all(calls);
expect(inFlight).toBe(0);
expect(peakInFlight).toBe(2);
});
it("frees a queued slot when its acquire waiter is aborted", async () => {
packages/coding-agent/test/issue-3749-provider-semaphore-deadlock.test.ts
+168
View File
@@ -0,0 +1,168 @@
/**
* Regression for [#3749](https://github.com/can1357/oh-my-pi/issues/3749):
* the per-provider concurrency cap used to bracket the whole subagent
* lifecycle (acquired before session creation, released only after the
* subagent yielded), so any spawn tree wider than `maxConcurrency`
* deadlocked — parents held every slot while they waited for children
* that were queued on the same cap. The fix moves the bracket to each
* LLM HTTP request; this file exercises the new contract.
*/
import { describe, expect, it } from "bun:test";
import type { StreamFn } from "@oh-my-pi/pi-agent-core";
import type { AssistantMessage, Model } from "@oh-my-pi/pi-ai";
import { AssistantMessageEventStream } from "@oh-my-pi/pi-ai/utils/event-stream";
import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings";
import { wrapStreamFnWithProviderConcurrency } from "@oh-my-pi/pi-coding-agent/task/provider-concurrency";
interface Deferred {
promise: Promise<void>;
resolve: () => void;
}
function deferred(): Deferred {
const { promise, resolve } = Promise.withResolvers<void>();
return { promise, resolve };
}
function requireModel(provider: string, id: string): Model {
const model = getBundledModel(provider as Parameters<typeof getBundledModel>[0], id);
if (!model) throw new Error(`Expected bundled model ${provider}/${id}`);
return model;
}
/**
* Build a base StreamFn that gates each invocation through an externally
* resolved Deferred, so the test can interleave parent and child turns
* deterministically without leaning on wall-clock timers.
*/
function makeGatedStream(): {
stream: StreamFn;
gates: Deferred[];
invocations: () => number;
inFlight: () => number;
peakInFlight: () => number;
} {
let inFlight = 0;
let peakInFlight = 0;
let invocations = 0;
const gates: Deferred[] = [];
const stream: StreamFn = model => {
const gate = deferred();
gates.push(gate);
invocations++;
inFlight++;
peakInFlight = Math.max(peakInFlight, inFlight);
const events = new AssistantMessageEventStream();
void gate.promise.then(() => {
inFlight--;
const message: AssistantMessage = {
role: "assistant",
content: [],
api: model.api,
provider: model.provider,
model: model.id,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now(),
};
events.push({ type: "done", reason: "stop", message });
events.end();
});
return events;
};
return {
stream,
gates,
invocations: () => invocations,
inFlight: () => inFlight,
peakInFlight: () => peakInFlight,
};
}
/**
* Drain microtasks until `check()` is true. Uses `Promise.resolve()` (a
* microtask hop, not a real-time delay), so the wait is bounded by the
* number of pending continuations and never burns wall-clock seconds.
*/
async function waitFor(check: () => boolean, label: string): Promise<void> {
for (let i = 0; i < 1000 && !check(); i++) {
await Promise.resolve();
}
if (!check()) {
throw new Error(`Timed out waiting for: ${label}`);
}
}
describe("issue #3749: provider semaphore deadlock", () => {
it("releases the slot between LLM turns so a child can acquire while the parent is mid-conversation", async () => {
const model = requireModel("ollama-cloud", "gpt-oss:120b");
const settings = Settings.isolated({ "providers.ollama-cloud.maxConcurrency": 1 });
const { stream, gates, invocations, inFlight, peakInFlight } = makeGatedStream();
const wrapped = wrapStreamFnWithProviderConcurrency(settings, stream);
// Parent's first turn acquires the only slot.
const parentTurn1 = wrapped(model, { messages: [] }, {});
await waitFor(() => invocations() === 1, "parent turn 1 invoked");
expect(inFlight()).toBe(1);
// Child tries to acquire while parent's first turn is in flight.
// Under the old lifetime-scoped bracket the child would queue and
// the parent (waiting for the child) would deadlock. The wrapper
// bounds only the HTTP request, so the child waits one parent turn,
// not the parent's whole lifetime.
const childTurn = wrapped(model, { messages: [] }, {});
await waitFor(() => gates.length === 1, "child queued");
expect(invocations()).toBe(1);
// Parent's first LLM stream completes → slot frees → child acquires.
gates[0]!.resolve();
await parentTurn1;
await waitFor(() => invocations() === 2, "child admitted after parent turn");
expect(inFlight()).toBe(1);
// Child completes. Parent's second turn can now start.
gates[1]!.resolve();
await childTurn;
const parentTurn2 = wrapped(model, { messages: [] }, {});
await waitFor(() => invocations() === 3, "parent turn 2 invoked");
gates[2]!.resolve();
await parentTurn2;
expect(peakInFlight()).toBe(1);
});
it("admits a deeper spawn tree than maxConcurrency without deadlocking", async () => {
const model = requireModel("ollama-cloud", "gpt-oss:120b");
const settings = Settings.isolated({ "providers.ollama-cloud.maxConcurrency": 2 });
const { stream, gates, invocations, peakInFlight } = makeGatedStream();
const wrapped = wrapStreamFnWithProviderConcurrency(settings, stream);
// 3 "parents" + 6 "children" all sharing a cap of 2. The old bracket
// would freeze after the first two parents acquired both slots.
const parents = [0, 1, 2].map(async () => wrapped(model, { messages: [] }, {}));
const children: Promise<unknown>[] = [];
for (let i = 0; i < 3; i++) {
children.push(Promise.resolve(wrapped(model, { messages: [] }, {})));
children.push(Promise.resolve(wrapped(model, { messages: [] }, {})));
}
// Drain by resolving gates in submission order as they appear.
for (let i = 0; i < 9; i++) {
await waitFor(() => gates.length > i, `gate ${i} created`);
gates[i]!.resolve();
}
await Promise.all([...parents, ...children]);
expect(invocations()).toBe(9);
expect(peakInFlight()).toBe(2);
});
});