⚡ refactor: Bound Concurrent Office-HTML Rendering for Code Artifacts (#12951)

* 🧰 feat: add `createConcurrencyLimiter` promise utility

Lightweight, dependency-free FIFO concurrency limiter for bounding
parallelism of expensive async work that fans out from a single
producer. Tasks queue rather than reject when the cap is reached;
slots release on both fulfillment and rejection so a single failure
cannot stall the queue. Each task runs inside a thunk so timeouts
and other side effects do not start until a slot is acquired.

* ⚡ perf: bound concurrent office-HTML rendering for code artifacts

A tool result with N office files (DOCX/XLSX/XLS/ODS/CSV/PPTX)
previously fanned out into N parallel mammoth/SheetJS/PPTX renders,
all CPU-bound and synchronous. Under bursty agent output this
competes with the still-running inference loop for event-loop time
and inflates end-of-run "finalize" waits in non-streaming flows
(BaseClient chat-completions, non-streaming Responses, the tools.js
direct endpoint) which all `await Promise.all(artifactPromises)`
before responding.

Cap the parser layer at 2 concurrent office-HTML renders process-
wide via a shared `createConcurrencyLimiter`. Tasks queue in FIFO
and the per-render timeout starts only after a slot is acquired so
queue waits do not consume the timeout budget. The HTML-or-null
contract (no text fallback for office types) is preserved.

packages/api/src/files/code/extract.spec.ts

@@ -597,3 +597,82 @@ describe('getExtractedTextFormat', () => {
     expect(getExtractedTextFormat('', '', 'whatever')).toBe('text');
   });
 });
+
+describe('extractCodeArtifactText office-html concurrency', () => {
+  beforeEach(() => {
+    mockOfficeHtml.mockReset();
+    parseDocumentCalls.length = 0;
+  });
+
+  /* The limiter inside extract.ts caps simultaneous office-HTML renders so
+   * a tool result with many office artifacts can't fan out N parallel
+   * mammoth/SheetJS invocations and starve the still-running agent loop.
+   * The cap is a module-private constant; this test asserts the observable
+   * contract without coupling to the literal value. */
+  it('caps concurrent office-HTML renders so a burst of 10 files does not all run at once', async () => {
+    let active = 0;
+    let peak = 0;
+
+    mockOfficeHtml.mockImplementation(async () => {
+      active++;
+      peak = Math.max(peak, active);
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      active--;
+      return '<!DOCTYPE html><html><body>ok</body></html>';
+    });
+
+    const results = await Promise.all(
+      Array.from({ length: 10 }, (_, i) =>
+        extractCodeArtifactText(
+          Buffer.from('PK'),
+          `report-${i}.docx`,
+          'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+          'document',
+        ),
+      ),
+    );
+
+    expect(results).toHaveLength(10);
+    results.forEach((text) => {
+      expect(text).toContain('<!DOCTYPE html>');
+    });
+    /* Hard upper bound from the module's OFFICE_HTML_CONCURRENCY=2.
+     * We assert <= a small constant rather than the exact value to avoid
+     * coupling, but it MUST be well below the request count. */
+    expect(peak).toBeGreaterThan(0);
+    expect(peak).toBeLessThanOrEqual(4);
+    expect(peak).toBeLessThan(10);
+  });
+
+  it('continues processing the queue when an earlier render rejects', async () => {
+    /* A failing render must release its slot — otherwise a single bad
+     * file in a burst would permanently shrink the limiter and stall
+     * subsequent extractions. */
+    let call = 0;
+    mockOfficeHtml.mockImplementation(async () => {
+      call++;
+      if (call === 1) {
+        throw new Error('mammoth blew up');
+      }
+      return '<!DOCTYPE html><html><body>ok</body></html>';
+    });
+
+    const results = await Promise.all(
+      Array.from({ length: 5 }, (_, i) =>
+        extractCodeArtifactText(
+          Buffer.from('PK'),
+          `report-${i}.docx`,
+          'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+          'document',
+        ),
+      ),
+    );
+
+    /* First call's failure surfaces as null per the HTML-or-null contract;
+     * the remaining four must still produce HTML. */
+    const nullCount = results.filter((t) => t === null).length;
+    const htmlCount = results.filter((t) => t != null && t.includes('<!DOCTYPE html>')).length;
+    expect(nullCount).toBe(1);
+    expect(htmlCount).toBe(4);
+  });
+});

packages/api/src/files/code/extract.ts

@@ -7,7 +7,7 @@ import type { CodeArtifactCategory } from './classify';
 import { parseDocument } from '~/files/documents/crud';
 import { bufferToOfficeHtml, officeHtmlBucket } from '~/files/documents/html';
 import { isBinaryBuffer } from '~/skills/binary';
-import { withTimeout } from '~/utils/promise';
+import { createConcurrencyLimiter, withTimeout } from '~/utils/promise';
 
 export const MAX_TEXT_CACHE_BYTES = 512 * 1024;
 export const MAX_TEXT_EXTRACT_BYTES = 1024 * 1024;
@@ -16,6 +16,21 @@ const OFFICE_HTML_TIMEOUT_MS = 12_000;
 const TRUNCATION_MARKER = '\n\n…[truncated]';
 const TRUNCATION_MARKER_BYTES = Buffer.byteLength(TRUNCATION_MARKER, 'utf-8');
 
+/**
+ * Cap simultaneous office-HTML renders process-wide. Mammoth (DOCX), SheetJS
+ * (XLSX/XLS/ODS), the CSV producer, and the PPTX renderer are CPU-heavy and
+ * synchronous; left unbounded, a tool result with N office files fans out
+ * into N parallel parses that compete with the still-running agent inference
+ * for event-loop time. Two slots keeps a single bursty tool result from
+ * starving inference while still allowing meaningful parallelism. Tasks
+ * queue in FIFO — none are dropped, only their start is deferred. Shared
+ * across every flow: streaming Responses, non-streaming Responses,
+ * chat-completions BaseClient, and the `tools.js` direct endpoint all
+ * funnel through this module.
+ */
+const OFFICE_HTML_CONCURRENCY = 2;
+const officeHtmlLimit = createConcurrencyLimiter(OFFICE_HTML_CONCURRENCY);
+
 /**
  * Decide whether a buffer is a candidate for rich HTML preview. Wraps the
  * shared `officeHtmlBucket` predicate from `~/files/documents/html` so the
@@ -181,10 +196,12 @@ const renderOfficeHtml = async (
   mimeType: string,
 ): Promise<string | null> => {
   try {
-    const html = await withTimeout(
-      bufferToOfficeHtml(buffer, name, mimeType),
-      OFFICE_HTML_TIMEOUT_MS,
-      `bufferToOfficeHtml exceeded ${OFFICE_HTML_TIMEOUT_MS}ms`,
+    const html = await officeHtmlLimit(() =>
+      withTimeout(
+        bufferToOfficeHtml(buffer, name, mimeType),
+        OFFICE_HTML_TIMEOUT_MS,
+        `bufferToOfficeHtml exceeded ${OFFICE_HTML_TIMEOUT_MS}ms`,
+      ),
     );
     if (html == null) {
       return null;

packages/api/src/utils/promise.spec.ts

@@ -1,4 +1,20 @@
-import { withTimeout } from './promise';
+import { createConcurrencyLimiter, withTimeout } from './promise';
+
+const tick = (ms = 0): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
+
+const deferred = <T>(): {
+  promise: Promise<T>;
+  resolve: (value: T) => void;
+  reject: (reason?: unknown) => void;
+} => {
+  let resolve!: (value: T) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+};
 
 describe('withTimeout', () => {
   beforeEach(() => {
@@ -113,3 +129,186 @@ describe('withTimeout', () => {
     );
   });
 });
+
+describe('createConcurrencyLimiter', () => {
+  it('runs tasks immediately while under the cap', async () => {
+    const limit = createConcurrencyLimiter(2);
+    const a = deferred<string>();
+    const b = deferred<string>();
+    let aStarted = false;
+    let bStarted = false;
+
+    const pa = limit(() => {
+      aStarted = true;
+      return a.promise;
+    });
+    const pb = limit(() => {
+      bStarted = true;
+      return b.promise;
+    });
+
+    await tick();
+    expect(aStarted).toBe(true);
+    expect(bStarted).toBe(true);
+
+    a.resolve('a');
+    b.resolve('b');
+    expect(await pa).toBe('a');
+    expect(await pb).toBe('b');
+  });
+
+  it('queues tasks beyond the cap and starts them as slots free', async () => {
+    const limit = createConcurrencyLimiter(2);
+    const a = deferred<string>();
+    const b = deferred<string>();
+    const c = deferred<string>();
+    let cStarted = false;
+
+    const pa = limit(() => a.promise);
+    const pb = limit(() => b.promise);
+    const pc = limit(() => {
+      cStarted = true;
+      return c.promise;
+    });
+
+    await tick();
+    expect(cStarted).toBe(false);
+
+    a.resolve('a');
+    expect(await pa).toBe('a');
+    await tick();
+    expect(cStarted).toBe(true);
+
+    b.resolve('b');
+    c.resolve('c');
+    expect(await pb).toBe('b');
+    expect(await pc).toBe('c');
+  });
+
+  it('dequeues in FIFO order', async () => {
+    const limit = createConcurrencyLimiter(1);
+    const order: string[] = [];
+    const head = deferred<void>();
+
+    const pHead = limit(async () => {
+      order.push('head:start');
+      await head.promise;
+      order.push('head:end');
+    });
+    const p1 = limit(async () => {
+      order.push('1');
+    });
+    const p2 = limit(async () => {
+      order.push('2');
+    });
+    const p3 = limit(async () => {
+      order.push('3');
+    });
+
+    await tick();
+    expect(order).toEqual(['head:start']);
+
+    head.resolve();
+    await Promise.all([pHead, p1, p2, p3]);
+    expect(order).toEqual(['head:start', 'head:end', '1', '2', '3']);
+  });
+
+  it('releases the slot when a task rejects, allowing queued work to proceed', async () => {
+    const limit = createConcurrencyLimiter(1);
+    const failing = deferred<never>();
+    let nextStarted = false;
+
+    const pFail = limit(() => failing.promise);
+    const pNext = limit(async () => {
+      nextStarted = true;
+      return 'ok';
+    });
+
+    await tick();
+    expect(nextStarted).toBe(false);
+
+    failing.reject(new Error('boom'));
+    await expect(pFail).rejects.toThrow('boom');
+    expect(await pNext).toBe('ok');
+    expect(nextStarted).toBe(true);
+  });
+
+  it('isolates rejections — one failing task does not reject sibling tasks', async () => {
+    const limit = createConcurrencyLimiter(2);
+    const ok = limit(async () => 'ok');
+    const bad = limit(async () => {
+      throw new Error('nope');
+    });
+
+    expect(await ok).toBe('ok');
+    await expect(bad).rejects.toThrow('nope');
+  });
+
+  it('does not exceed the configured cap under heavy load', async () => {
+    const concurrency = 3;
+    const limit = createConcurrencyLimiter(concurrency);
+    let active = 0;
+    let peak = 0;
+
+    const tasks = Array.from({ length: 20 }, (_, i) =>
+      limit(async () => {
+        active++;
+        peak = Math.max(peak, active);
+        await tick(5);
+        active--;
+        return i;
+      }),
+    );
+
+    const results = await Promise.all(tasks);
+    expect(results).toEqual(Array.from({ length: 20 }, (_, i) => i));
+    expect(peak).toBeLessThanOrEqual(concurrency);
+    expect(peak).toBe(concurrency);
+  });
+
+  it('serializes when concurrency is 1', async () => {
+    const limit = createConcurrencyLimiter(1);
+    let active = 0;
+    let peak = 0;
+
+    await Promise.all(
+      Array.from({ length: 5 }, () =>
+        limit(async () => {
+          active++;
+          peak = Math.max(peak, active);
+          await tick(2);
+          active--;
+        }),
+      ),
+    );
+
+    expect(peak).toBe(1);
+  });
+
+  it('throws when concurrency is not a positive integer', () => {
+    expect(() => createConcurrencyLimiter(0)).toThrow(/positive integer/);
+    expect(() => createConcurrencyLimiter(-1)).toThrow(/positive integer/);
+    expect(() => createConcurrencyLimiter(1.5)).toThrow(/positive integer/);
+    expect(() => createConcurrencyLimiter(Infinity)).toThrow(/positive integer/);
+    expect(() => createConcurrencyLimiter(NaN)).toThrow(/positive integer/);
+  });
+
+  it('does not invoke the task until a slot is available', async () => {
+    const limit = createConcurrencyLimiter(1);
+    const block = deferred<void>();
+    let queuedTaskCalled = false;
+
+    const pHead = limit(() => block.promise);
+    const pQueued = limit(async () => {
+      queuedTaskCalled = true;
+    });
+
+    await tick();
+    await tick();
+    expect(queuedTaskCalled).toBe(false);
+
+    block.resolve();
+    await Promise.all([pHead, pQueued]);
+    expect(queuedTaskCalled).toBe(true);
+  });
+});

packages/api/src/utils/promise.ts

@@ -40,3 +40,69 @@ export async function withTimeout<T>(
     clearTimeout(timeoutId!);
   }
 }
+
+/**
+ * Create an in-process concurrency limiter. Returns a `run` function that
+ * wraps async tasks: at most `concurrency` invocations may execute at once;
+ * additional calls queue and dequeue in FIFO order as slots free.
+ *
+ * Use to bound the parallelism of expensive CPU-or-IO work that fans out
+ * from a single producer (e.g. an agent emitting many office artifacts in
+ * one tool result), so the work doesn't compete with the still-running
+ * agent inference for event-loop time. Tasks remain queued — they are
+ * never dropped or rejected by the limiter itself — so the overall workload
+ * still completes; only peak concurrency is capped.
+ *
+ * Each task is wrapped in a thunk so timeouts and other side effects do
+ * not start until the limiter actually invokes it.
+ *
+ * @example
+ * ```typescript
+ * const limit = createConcurrencyLimiter(2);
+ * const results = await Promise.all(files.map((f) => limit(() => parse(f))));
+ * ```
+ */
+export function createConcurrencyLimiter(
+  concurrency: number,
+): <T>(task: () => Promise<T>) => Promise<T> {
+  if (!Number.isInteger(concurrency) || concurrency < 1) {
+    throw new Error(
+      `createConcurrencyLimiter: concurrency must be a positive integer (got ${concurrency})`,
+    );
+  }
+
+  let active = 0;
+  const queue: Array<() => void> = [];
+
+  const release = (): void => {
+    active--;
+    const next = queue.shift();
+    if (next) {
+      next();
+    }
+  };
+
+  return <T>(task: () => Promise<T>): Promise<T> =>
+    new Promise<T>((resolve, reject) => {
+      const run = (): void => {
+        active++;
+        Promise.resolve()
+          .then(task)
+          .then(
+            (value) => {
+              release();
+              resolve(value);
+            },
+            (error) => {
+              release();
+              reject(error);
+            },
+          );
+      };
+      if (active < concurrency) {
+        run();
+      } else {
+        queue.push(run);
+      }
+    });
+}