diff --git a/packages/api/src/files/code/extract.spec.ts b/packages/api/src/files/code/extract.spec.ts
index a2847b6428..1baa2490f1 100644
--- a/packages/api/src/files/code/extract.spec.ts
+++ b/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 '
ok';
+ });
+
+ 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('');
+ });
+ /* 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 'ok';
+ });
+
+ 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('')).length;
+ expect(nullCount).toBe(1);
+ expect(htmlCount).toBe(4);
+ });
+});
diff --git a/packages/api/src/files/code/extract.ts b/packages/api/src/files/code/extract.ts
index d61337df20..85f9321d71 100644
--- a/packages/api/src/files/code/extract.ts
+++ b/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 => {
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;
diff --git a/packages/api/src/utils/promise.spec.ts b/packages/api/src/utils/promise.spec.ts
index c43c8bf739..93c5fd8893 100644
--- a/packages/api/src/utils/promise.spec.ts
+++ b/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 => new Promise((resolve) => setTimeout(resolve, ms));
+
+const deferred = (): {
+ promise: Promise;
+ resolve: (value: T) => void;
+ reject: (reason?: unknown) => void;
+} => {
+ let resolve!: (value: T) => void;
+ let reject!: (reason?: unknown) => void;
+ const promise = new Promise((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();
+ const b = deferred();
+ 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();
+ const b = deferred();
+ const c = deferred();
+ 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();
+
+ 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();
+ 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();
+ 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);
+ });
+});
diff --git a/packages/api/src/utils/promise.ts b/packages/api/src/utils/promise.ts
index 72719a3ff0..06a3e440d4 100644
--- a/packages/api/src/utils/promise.ts
+++ b/packages/api/src/utils/promise.ts
@@ -40,3 +40,69 @@ export async function withTimeout(
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,
+): (task: () => Promise) => Promise {
+ 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 (task: () => Promise): Promise =>
+ new Promise((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);
+ }
+ });
+}