fix: address 5 Detail scan bugs from March 11 (reconnect, mutex leak, playout, ordering, retryability) (#1188)

Author: toubatbrian

.changeset/shy-pianos-sort.md

@@ -0,0 +1,5 @@
+---
+"@livekit/agents": patch
+---
+
+fix: address 5 Detail scan bugs from March 11 (reconnect, mutex leak, playout, ordering, retryability)

agents/src/_exceptions.test.ts

@@ -0,0 +1,45 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { describe, expect, it } from 'vitest';
+import { APIStatusError } from './_exceptions.js';
+
+describe('APIStatusError retryability defaults', () => {
+  it('treats 408 as retryable by default', () => {
+    const error = new APIStatusError({
+      message: 'timeout',
+      options: { statusCode: 408 },
+    });
+    expect(error.retryable).toBe(true);
+  });
+
+  it('treats 429 as retryable by default', () => {
+    const error = new APIStatusError({
+      message: 'rate limited',
+      options: { statusCode: 429 },
+    });
+    expect(error.retryable).toBe(true);
+  });
+
+  it('keeps other 4xx responses non-retryable by default', () => {
+    const error = new APIStatusError({
+      message: 'not found',
+      options: { statusCode: 404 },
+    });
+    expect(error.retryable).toBe(false);
+  });
+
+  it('respects explicit retryable override', () => {
+    const forceRetryable = new APIStatusError({
+      message: 'force retry',
+      options: { statusCode: 404, retryable: true },
+    });
+    const forceNonRetryable = new APIStatusError({
+      message: 'force no retry',
+      options: { statusCode: 429, retryable: false },
+    });
+
+    expect(forceRetryable.retryable).toBe(true);
+    expect(forceNonRetryable.retryable).toBe(false);
+  });
+});

agents/src/_exceptions.ts

@@ -79,8 +79,10 @@ export class APIStatusError extends APIError {
     options?: APIStatusErrorOptions;
   }) {
     const statusCode = options.statusCode ?? -1;
-    // 4xx errors are not retryable
-    const isRetryable = options.retryable ?? !(statusCode >= 400 && statusCode < 500);
+    // 408/429 are transient even though they are 4xx, so keep them retryable by default.
+    const isRetryable =
+      options.retryable ??
+      (statusCode === 408 || statusCode === 429 || !(statusCode >= 400 && statusCode < 500));
 
     super(message, { body: options.body, retryable: isRetryable });
     this.name = 'APIStatusError';

agents/src/inference/stt.ts

@@ -543,13 +543,20 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
       try {
         ws = await this.stt.connectWs(this.connOptions.timeoutMs);
 
-        const controller = this.abortController; // Use base class abortController for proper cancellation
-        const sendTask = Task.from(({ signal }) => send(ws!, signal), controller);
-        const wsListenerTask = Task.from(({ signal }) => createWsListener(ws!, signal), controller);
-        const recvTask = Task.from(({ signal }) => recv(signal), controller);
+        // Use a per-connection controller so reconnect loops don't inherit a permanently-aborted signal.
+        const connController = new AbortController();
+        const onStreamAbort = () => connController.abort();
+        this.abortController.signal.addEventListener('abort', onStreamAbort);
+
+        const sendTask = Task.from(({ signal }) => send(ws!, signal), connController);
+        const wsListenerTask = Task.from(
+          ({ signal }) => createWsListener(ws!, signal),
+          connController,
+        );
+        const recvTask = Task.from(({ signal }) => recv(signal), connController);
         const waitReconnectTask = Task.from(
           ({ signal }) => Promise.race([this.reconnectEvent.wait(), waitForAbort(signal)]),
-          controller,
+          connController,
         );
 
         try {
@@ -564,13 +571,16 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
           // Reconnect triggered - clear event and continue loop
           this.reconnectEvent.clear();
         } finally {
-          // Cancel all tasks to ensure cleanup
+          connController.abort();
+          this.abortController.signal.removeEventListener('abort', onStreamAbort);
           await cancelAndWait(
             [sendTask, wsListenerTask, recvTask, waitReconnectTask],
             DEFAULT_CANCEL_TIMEOUT,
           );
           resourceCleanup();
         }
+
+        if (this.abortController.signal.aborted) break;
       } finally {
         // Ensure cleanup even if connectWs throws
         resourceCleanup();

agents/src/ipc/proc_pool.test.ts

@@ -0,0 +1,69 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { describe, expect, it, vi } from 'vitest';
+import type { RunningJobInfo } from '../job.js';
+import { type JobExecutor, JobStatus } from './job_executor.js';
+import { ProcPool } from './proc_pool.js';
+
+function createMockExecutor() {
+  const executor: JobExecutor = {
+    started: true,
+    userArguments: {},
+    runningJob: undefined,
+    status: JobStatus.RUNNING,
+    start: vi.fn(async () => {}),
+    join: vi.fn(async () => {}),
+    initialize: vi.fn(async () => {}),
+    close: vi.fn(async () => {}),
+    launchJob: vi.fn(async () => {}),
+  };
+  return executor;
+}
+
+describe('ProcPool warmed process lock handling', () => {
+  it('releases lock token from the dequeued warmed process entry', async () => {
+    const pool = new ProcPool('agent', 1, 1000, 1000, undefined, 0, 0);
+    const unlock = vi.fn();
+    const executor = createMockExecutor();
+    const jobInfo = {
+      acceptArguments: { name: 'n', identity: 'i', metadata: '' },
+      job: { id: 'job-id' },
+      url: 'wss://example.com',
+      token: 'token',
+      workerId: 'worker-id',
+    } as unknown as RunningJobInfo;
+
+    await pool.warmedProcQueue.put({ proc: executor, unlock });
+    await pool.launchJob(jobInfo);
+
+    expect(unlock).toHaveBeenCalledTimes(1);
+    expect(executor.launchJob).toHaveBeenCalledWith(jobInfo);
+  });
+
+  it('releases queued lock tokens during close', async () => {
+    const pool = new ProcPool('agent', 1, 1000, 1000, undefined, 0, 0);
+    const unlock = vi.fn();
+    const executor = createMockExecutor();
+
+    await pool.warmedProcQueue.put({ proc: executor, unlock });
+    pool.started = true;
+    await pool.close();
+
+    expect(unlock).toHaveBeenCalledTimes(1);
+    expect(executor.close).toHaveBeenCalledTimes(1);
+  });
+
+  it('releases both init and proc locks when closed before proc starts', async () => {
+    const pool = new ProcPool('agent', 1, 1000, 1000, undefined, 0, 0);
+    const initUnlock = vi.fn();
+    const procUnlock = vi.fn();
+    pool.closed = true;
+    pool.initMutex.lock = vi.fn(async () => initUnlock);
+
+    await pool.procWatchTask(procUnlock);
+
+    expect(initUnlock).toHaveBeenCalledTimes(1);
+    expect(procUnlock).toHaveBeenCalledTimes(1);
+  });
+});

agents/src/ipc/proc_pool.ts

@@ -19,8 +19,8 @@ export class ProcPool {
   controller = new AbortController();
   initMutex = new Mutex();
   procMutex?: MultiMutex;
-  procUnlock?: () => void;
-  warmedProcQueue = new Queue<JobExecutor>();
+  // Keep each lock token paired with its warmed process so MultiMutex slots are always released correctly.
+  warmedProcQueue = new Queue<{ proc: JobExecutor; unlock: () => void }>();
   inferenceExecutor?: InferenceExecutor;
   memoryWarnMB: number;
   memoryLimitMB: number;
@@ -56,11 +56,10 @@ export class ProcPool {
   async launchJob(info: RunningJobInfo) {
     let proc: JobExecutor;
     if (this.procMutex) {
-      proc = await this.warmedProcQueue.get();
-      if (this.procUnlock) {
-        this.procUnlock();
-        this.procUnlock = undefined;
-      }
+      const entry = await this.warmedProcQueue.get();
+      proc = entry.proc;
+      // Release exactly the slot that produced this warmed process.
+      entry.unlock();
     } else {
       proc = new JobProcExecutor(
         this.agent,
@@ -80,7 +79,7 @@ export class ProcPool {
     await proc.launchJob(info);
   }
 
-  async procWatchTask() {
+  async procWatchTask(procUnlock: () => void) {
     const proc = new JobProcExecutor(
       this.agent,
       this.inferenceExecutor,
@@ -97,23 +96,28 @@ export class ProcPool {
       this.executors.push(proc);
 
       const unlock = await this.initMutex.lock();
-      if (this.closed) {
-        return;
-      }
-
-      await proc.start();
+      let procUnlockTransferred = false;
       try {
-        await proc.initialize();
-        await this.warmedProcQueue.put(proc);
-      } catch {
-        if (this.procUnlock) {
-          this.procUnlock();
-          this.procUnlock = undefined;
+        if (this.closed) {
+          return;
         }
-      }
 
-      unlock();
-      await proc.join();
+        await proc.start();
+        try {
+          await proc.initialize();
+          await this.warmedProcQueue.put({ proc, unlock: procUnlock });
+          procUnlockTransferred = true;
+        } catch {
+          // Initialization failed before enqueue, so release the acquired slot immediately.
+        }
+
+        await proc.join();
+      } finally {
+        unlock();
+        if (!procUnlockTransferred) {
+          procUnlock();
+        }
+      }
     } finally {
       const procIndex = this.executors.indexOf(proc);
       if (procIndex !== -1) {
@@ -136,8 +140,8 @@ export class ProcPool {
   async run(signal: AbortSignal) {
     if (this.procMutex) {
       while (!signal.aborted) {
-        this.procUnlock = await this.procMutex.lock();
-        const task = this.procWatchTask();
+        const procUnlock = await this.procMutex.lock();
+        const task = this.procWatchTask(procUnlock);
         this.tasks.push(task);
         task.finally(() => {
           const taskIndex = this.tasks.indexOf(task);
@@ -157,7 +161,10 @@ export class ProcPool {
     }
     this.closed = true;
     this.controller.abort();
-    this.warmedProcQueue.items.forEach((e) => e.close());
+    this.warmedProcQueue.items.forEach((e) => {
+      e.unlock();
+      e.proc.close();
+    });
     this.executors.forEach((e) => e.close());
     await Promise.allSettled(this.tasks);
   }

agents/src/llm/provider_format/utils.test.ts

@@ -0,0 +1,28 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { describe, expect, it } from 'vitest';
+import { ChatContext, ChatMessage } from '../chat_context.js';
+import { groupToolCalls } from './utils.js';
+
+describe('groupToolCalls', () => {
+  it('preserves insertion order for non-assistant message groups', () => {
+    const chatCtx = new ChatContext([
+      ChatMessage.create({
+        id: '10',
+        role: 'user',
+        content: 'first',
+      }),
+      ChatMessage.create({
+        id: '2',
+        role: 'system',
+        content: 'second',
+      }),
+    ]);
+
+    const groups = groupToolCalls(chatCtx);
+    const itemIds = groups.map((group) => group.flatten()[0]!.id);
+
+    expect(itemIds).toEqual(['10', '2']);
+  });
+});

agents/src/llm/provider_format/utils.ts

@@ -51,6 +51,8 @@ class ChatItemGroup {
       this.toolCalls.push(item);
     } else if (item.type === 'function_call_output') {
       this.toolOutputs.push(item);
+    } else if (item.type === 'agent_handoff') {
+      // provider formatters don't serialize handoff records into model input.
     }
     return this;
   }
@@ -153,6 +155,8 @@ export function groupToolCalls(chatCtx: ChatContext) {
       toolOutputs.push(item);
     } else {
       itemGroups[item.id] = ChatItemGroup.create().add(item);
+      // User/system messages and agent_handoff items also need stable insertion indices.
+      insertionOrder[item.id] = insertionIndex++;
     }
   }