fix(agents): release initMutex after warming to restore pool concurrency (#1214)

Author: drain-zine

.changeset/afraid-cars-clap.md

@@ -0,0 +1,5 @@
+---
+"@livekit/agents": patch
+---
+
+fix(agents): release initMutex after warming to restore pool concurrency

agents/src/ipc/proc_pool.test.ts

@@ -5,6 +5,7 @@ import type { Throws } from '@livekit/throws-transformer/throws';
 import { describe, expect, it, vi } from 'vitest';
 import type { RunningJobInfo } from '../job.js';
 import { type JobExecutor, JobStatus } from './job_executor.js';
+import * as jobProcExecutorModule from './job_proc_executor.js';
 import { ProcPool } from './proc_pool.js';
 
 function createMockExecutor() {
@@ -22,6 +23,13 @@ function createMockExecutor() {
   return executor;
 }
 
+/** Flush the microtask queue enough times for an async chain to settle. */
+async function flushMicrotasks(ticks = 10): Promise<void> {
+  for (let i = 0; i < ticks; i++) {
+    await Promise.resolve();
+  }
+}
+
 describe('ProcPool warmed process lock handling', () => {
   it('releases lock token from the dequeued warmed process entry', async (): Promise<
     Throws<void, Error>
@@ -69,4 +77,84 @@ describe('ProcPool warmed process lock handling', () => {
     expect(initUnlock).toHaveBeenCalledTimes(1);
     expect(procUnlock).toHaveBeenCalledTimes(1);
   });
+
+  it('releases initMutex after warming so concurrent procWatchTasks can initialise', async (): Promise<
+    Throws<void, Error>
+  > => {
+    // Regression: initMutex must be released after enqueue, not after join().
+    // Child procs are one-shot, so holding initMutex through join() serialises
+    // the pool to effective concurrency 1 regardless of numIdleProcesses.
+    const pool = new ProcPool('agent', 1, 1000, 1000, undefined, 0, 0);
+    const initUnlock = vi.fn();
+    const procUnlock = vi.fn();
+
+    let joinResolve: () => void = () => {};
+    const joinPromise = new Promise<void>((resolve) => {
+      joinResolve = resolve;
+    });
+    const mockProc: JobExecutor = {
+      ...createMockExecutor(),
+      join: vi.fn(() => joinPromise),
+    };
+
+    const jobProcExecutorSpy = vi
+      .spyOn(jobProcExecutorModule, 'JobProcExecutor')
+      .mockImplementation(function MockJobProcExecutor(this: unknown) {
+        return mockProc as unknown as jobProcExecutorModule.JobProcExecutor;
+      } as unknown as typeof jobProcExecutorModule.JobProcExecutor);
+
+    pool.initMutex.lock = vi.fn(async () => initUnlock);
+
+    try {
+      const watchPromise = pool.procWatchTask(procUnlock);
+      await flushMicrotasks();
+
+      // initMutex released while proc.join() is still pending.
+      expect(initUnlock).toHaveBeenCalledTimes(1);
+      expect(pool.warmedProcQueue.items.length).toBe(1);
+      expect(mockProc.join).toHaveBeenCalledTimes(1);
+
+      joinResolve();
+      await watchPromise;
+
+      // finally block must not double-release.
+      expect(initUnlock).toHaveBeenCalledTimes(1);
+      expect(procUnlock).not.toHaveBeenCalled();
+    } finally {
+      jobProcExecutorSpy.mockRestore();
+    }
+  });
+
+  it('releases initMutex in finally when initialization fails before enqueue', async (): Promise<
+    Throws<void, Error>
+  > => {
+    const pool = new ProcPool('agent', 1, 1000, 1000, undefined, 0, 0);
+    const initUnlock = vi.fn();
+    const procUnlock = vi.fn();
+
+    const mockProc: JobExecutor = {
+      ...createMockExecutor(),
+      initialize: vi.fn(async () => {
+        throw new Error('simulated initialization failure');
+      }),
+    };
+
+    const jobProcExecutorSpy = vi
+      .spyOn(jobProcExecutorModule, 'JobProcExecutor')
+      .mockImplementation(function MockJobProcExecutor(this: unknown) {
+        return mockProc as unknown as jobProcExecutorModule.JobProcExecutor;
+      } as unknown as typeof jobProcExecutorModule.JobProcExecutor);
+
+    pool.initMutex.lock = vi.fn(async () => initUnlock);
+
+    try {
+      await pool.procWatchTask(procUnlock);
+
+      expect(initUnlock).toHaveBeenCalledTimes(1);
+      expect(procUnlock).toHaveBeenCalledTimes(1);
+      expect(pool.warmedProcQueue.items.length).toBe(0);
+    } finally {
+      jobProcExecutorSpy.mockRestore();
+    }
+  });
 });

agents/src/ipc/proc_pool.ts

@@ -97,6 +97,7 @@ export class ProcPool {
       this.executors.push(proc);
 
       const unlock = await this.initMutex.lock();
+      let initReleased = false;
       let procUnlockTransferred = false;
       try {
         if (this.closed) {
@@ -108,13 +109,19 @@ export class ProcPool {
           await proc.initialize();
           await this.warmedProcQueue.put({ proc, unlock: procUnlock });
           procUnlockTransferred = true;
+          // Release initMutex after enqueue — holding it through join() serialises
+          // the pool to concurrency 1 since child procs are one-shot.
+          unlock();
+          initReleased = true;
         } catch {
           // Initialization failed before enqueue, so release the acquired slot immediately.
         }
 
         await proc.join();
       } finally {
-        unlock();
+        if (!initReleased) {
+          unlock();
+        }
         if (!procUnlockTransferred) {
           procUnlock();
         }