feat: add optional chooseMany concurrency

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 - Added the MCP `chooseMany` tool for independent named questions with shared context and optional images.
+- Added optional `concurrency` to `chooseMany()` without changing its default parallelism. Queued questions keep a snapshot of their inputs and stop on failure or cancellation.
 
 ## 0.8.0 - 2026-10-10
 

README.md

@@ -166,7 +166,7 @@ const decisions = await chooseMany(choose, {
 console.log(decisions.route.choice, decisions.urgency.choice);
 ```
 
-Each question calls `choose()` concurrently and returns its own decision and, when available, token and request counts. Supplied images are sent with each call. If one call fails, the group rejects while calls already in progress continue.
+Each question calls `choose()` concurrently by default. Set `concurrency` to limit parallel calls. Each question returns its own decision, with usage when available. Images are sent with each call. On failure, queued questions stop. Running calls continue.
 
 ## Scoring modes
 

src/choose-many.ts

@@ -10,6 +10,8 @@ type Request<Q extends Questions> = {
   readonly questions: Q;
   readonly images?: readonly ImageInput[];
   readonly signal?: AbortSignal;
+  /** Maximum simultaneous choose calls in this group. Omit to start all questions concurrently. */
+  readonly concurrency?: number;
 };
 
 type Results<Q extends Questions> = Readonly<{
@@ -20,19 +22,46 @@ export async function chooseMany<const Q extends Questions>(
   choose: Chooser, request: Request<Q>,
 ): Promise<Results<Q>> {
   if (typeof choose !== "function") throw new TypeError("choose must be a function.");
-  const { context, questions, images, signal } = request;
+  const { context, questions, images, signal, concurrency } = request;
   signal?.throwIfAborted();
   if (!isRecord(questions) || Object.getOwnPropertySymbols(questions).length !== 0) {
     throw new TypeError("questions must be an object of named questions.");
   }
-  const entries = Object.entries(questions);
+  if (concurrency !== undefined && (!Number.isSafeInteger(concurrency) || concurrency < 1)) {
+    throw new TypeError("concurrency must be a positive safe integer.");
+  }
+  const entries = Object.entries(questions).map(([name, { question, choices }]) => [
+    name, question,
+    isRecord(choices) && Object.getOwnPropertySymbols(choices).length === 0
+      ? Object.freeze({ ...choices }) : choices,
+  ] as const);
   if (entries.length === 0) throw new TypeError("Provide at least one question.");
+  const sharedImages = Array.isArray(images)
+    ? Object.freeze(images.map((image: ImageInput) => isRecord(image)
+      ? Object.freeze({ mediaType: image.mediaType, base64: image.base64 }) : image))
+    : images;
 
-  const decisions = await Promise.all(entries.map(async ([name, { question, choices }]) =>
-    [name, await choose({
-      context, question, choices,
-      ...(images === undefined ? {} : { images }),
-      ...(signal === undefined ? {} : { signal }),
-    })] as const));
+  const decisions: [string, Decision<string>][] = new Array(entries.length);
+  let next = 0;
+  let failed = false;
+  await Promise.all(Array.from({ length: Math.min(concurrency ?? entries.length, entries.length) }, async () => {
+    try {
+      while (!failed && next < entries.length) {
+        signal?.throwIfAborted();
+        const index = next++;
+        const [name, question, choices] = entries[index]!;
+        const decision = await choose({
+          context, question, choices,
+          ...(sharedImages === undefined ? {} : { images: sharedImages }),
+          ...(signal === undefined ? {} : { signal }),
+        });
+        signal?.throwIfAborted();
+        decisions[index] = [name, decision];
+      }
+    } catch (error) {
+      failed = true;
+      throw error;
+    }
+  }));
   return Object.freeze(Object.fromEntries(decisions)) as Results<Q>;
 }

tests/choose-many.test.mjs

@@ -107,3 +107,202 @@ test("passes a shared signal to every in-flight question", async () => {
   controller.abort(reason);
   await assert.rejects(pending, (error) => error === reason);
 });
+
+function questions(count) {
+  return Object.fromEntries(Array.from({ length: count }, (_, i) => [
+    `q${i}`, { question: `Question ${i}?`, choices: { yes: "Yes", no: "No" } },
+  ]));
+}
+
+for (const concurrency of [1, 3]) {
+  test(`snapshots queued questions, choices and images (concurrency: ${concurrency})`, async () => {
+    const input = questions(5);
+    const images = [{ mediaType: "image/png", base64: "aW1hZ2U=" }];
+    const seen = [];
+    let release;
+    const first = new Promise((resolve) => { release = resolve; });
+    const choose = createChooser(async (request) => {
+      seen.push(request);
+      if (seen.length === 1) await first;
+      return { logprobs: [-0.1, -2] };
+    });
+    const pending = chooseMany(choose, {
+      context: "Shared.", questions: input, images, concurrency,
+    });
+
+    input.q4.question = "Changed question?";
+    input.q4.choices.yes = "Changed description";
+    input.q4.choices.extra = "New choice";
+    images[0].base64 = "Y2hhbmdlZA==";
+    images.push({ mediaType: "image/jpeg", base64: "bmV3" });
+    release();
+
+    const result = await pending;
+    assert.deepEqual(Object.keys(result.q4.distribution), ["yes", "no"]);
+    assert.equal(seen.length, 5);
+    const queued = seen.find(({ prompt }) => prompt.includes("Question 4?"));
+    assert.ok(queued);
+    assert.ok(queued.prompt.includes('"yes": "Yes"'));
+    for (const request of seen) {
+      assert.deepEqual(request.images, [{ mediaType: "image/png", base64: "aW1hZ2U=" }]);
+    }
+  });
+}
+
+test("snapshots image fields without depending on property enumeration", async () => {
+  const expected = { mediaType: "image/png", base64: "aW1hZ2U=" };
+  const images = [
+    new class {
+      get mediaType() { return expected.mediaType; }
+      get base64() { return expected.base64; }
+    }(),
+    Object.defineProperties({}, {
+      mediaType: { value: expected.mediaType },
+      base64: { value: expected.base64 },
+    }),
+    Object.defineProperty({ ...expected }, "metadata", {
+      enumerable: true,
+      get() { assert.fail("Unrelated image properties must not be read."); },
+    }),
+  ];
+  for (const image of images) {
+    const seen = [];
+    const choose = createChooser((request) => {
+      seen.push(request.images);
+      return { logprobs: [-0.1, -2] };
+    });
+    await chooseMany(choose, { context: "", questions: questions(2), images: [image], concurrency: 1 });
+    assert.deepEqual(seen, [[expected], [expected]]);
+  }
+});
+
+test("snapshotting preserves chooser validation for malformed inputs", async () => {
+  let calls = 0;
+  const choose = createChooser(() => { calls++; return { logprobs: [-0.1, -2] }; });
+  const symbolChoices = Object.defineProperty({ yes: "Yes", no: "No" }, Symbol("hidden"), { value: "Invalid" });
+  const arrayImage = Object.assign([], { mediaType: "image/png", base64: "aW1hZ2U=" });
+  for (const choices of [["Yes", "No"], symbolChoices]) {
+    await assert.rejects(chooseMany(choose, {
+      context: "", questions: { one: { question: "One?", choices } },
+    }), /choices must be a record/);
+  }
+  for (const images of [{ 0: arrayImage }, [arrayImage]]) {
+    await assert.rejects(chooseMany(choose, {
+      context: "", questions: questions(1), images,
+    }), TypeError);
+  }
+  assert.equal(calls, 0);
+});
+
+for (const [count, concurrency] of [[1000, 3], [5, 1], [2, 3], [5, undefined]]) {
+  test(`bounds ${count} questions and preserves input order (concurrency: ${concurrency ?? "default"})`, async () => {
+    const input = questions(count);
+    const active = new Map();
+    let peak = 0;
+    let calls = 0;
+    const choose = ({ question }) => new Promise((resolve) => {
+      calls++;
+      active.set(question, () => {
+        active.delete(question);
+        resolve({ choice: question });
+      });
+      peak = Math.max(peak, active.size);
+    });
+    const pending = chooseMany(choose, {
+      context: "", questions: input,
+      ...(concurrency === undefined ? {} : { concurrency }),
+    });
+    const limit = Math.min(concurrency ?? count, count);
+    // The initial window starts immediately, without serializing small groups.
+    assert.equal(calls, limit);
+    while (active.size > 0) {
+      // Complete each window backwards to detect results stored in completion order.
+      for (const release of [...active.values()].reverse()) release();
+      await new Promise(setImmediate);
+      assert.ok(active.size <= limit);
+    }
+    const result = await pending;
+    assert.equal(peak, limit);
+    assert.equal(calls, count);
+    assert.deepEqual(Object.keys(result), Object.keys(input));
+    for (const [name, value] of Object.entries(input)) {
+      assert.equal(result[name].choice, value.question);
+    }
+    assert.ok(Object.isFrozen(result));
+  });
+}
+
+test("rejects invalid concurrency before invoking the chooser", async () => {
+  let calls = 0;
+  const choose = async () => { calls++; };
+  for (const concurrency of [0, -1, 1.5, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1, "2", null]) {
+    await assert.rejects(chooseMany(choose, {
+      context: "", questions: questions(2), concurrency,
+    }), /concurrency must be a positive safe integer/);
+  }
+  assert.equal(calls, 0);
+});
+
+for (const failure of [new Error("Backend failed"), undefined]) {
+  test(`stops queued work on rejection and observes late sibling errors (${failure})`, async () => {
+    const started = [];
+    const rejectCalls = [];
+    const choose = ({ question }) => new Promise((_, reject) => {
+      started.push(question);
+      rejectCalls.push(reject);
+    });
+    const pending = chooseMany(choose, { context: "", questions: questions(4), concurrency: 3 });
+    assert.equal(started.length, 3);
+    rejectCalls[1](failure);
+    await assert.rejects(pending, (error) => error === failure);
+    for (const reject of rejectCalls) reject(new Error("Late sibling failure"));
+    await new Promise(setImmediate);
+    assert.equal(started.length, 3);
+  });
+}
+
+test("stops queued work even when the chooser throws synchronously", async () => {
+  const failure = new Error("Synchronous failure");
+  let calls = 0;
+  await assert.rejects(chooseMany(() => { calls++; throw failure; }, {
+    context: "", questions: questions(4), concurrency: 3,
+  }), (error) => error === failure);
+  assert.equal(calls, 1);
+});
+
+for (const honorsAbort of [true, false]) {
+  test(`cancellation prevents queued work (chooser honors abort: ${honorsAbort})`, async () => {
+    const controller = new AbortController();
+    const reason = new Error("Cancelled");
+    const received = [];
+    const releases = [];
+    const choose = ({ signal }) => new Promise((resolve, reject) => {
+      received.push(signal);
+      releases.push(() => resolve({ choice: "yes" }));
+      if (honorsAbort) signal.addEventListener("abort", () => reject(signal.reason), { once: true });
+    });
+    const pending = chooseMany(choose, {
+      context: "", questions: questions(4), signal: controller.signal, concurrency: 3,
+    });
+    assert.deepEqual(received, Array(3).fill(controller.signal));
+    controller.abort(reason);
+    for (const release of releases) release();
+    await assert.rejects(pending, (error) => error === reason);
+    await new Promise(setImmediate);
+    assert.equal(received.length, 3);
+  });
+}
+
+test("refills a free slot without waiting for the rest of its window", async () => {
+  const releases = [];
+  const pending = chooseMany(() => new Promise((resolve) => {
+    releases.push(() => resolve({ choice: "yes" }));
+  }), { context: "", questions: questions(3), concurrency: 2 });
+  assert.equal(releases.length, 2);
+  releases[1]();
+  await new Promise(setImmediate);
+  assert.equal(releases.length, 3);
+  releases[0]();
+  releases[2]();
+  await pending;
+});

tests/types/api.cts

@@ -9,6 +9,7 @@ void fromLlamaCpp({ baseURL: "http://127.0.0.1:8080" });
 void fromOllama({ model: "test-model" });
 void fromOpenRouter({ apiKey: "test-key", model: "test/model" });
 void chooseMany(choose, {
+  concurrency: 2,
   context: "", questions: { route: { question: "Next?", choices: { test: "Test", done: "Done" } } },
 }).then((decisions) => {
   const key: "test" | "done" = decisions.route.choice;

tests/types/api.mts

@@ -56,6 +56,7 @@ void cached;
 decision.confidence;
 
 const decisions = await chooseMany(choose, {
+  concurrency: 2,
   context: "A payout failed.",
   images: [image],
   questions: {