fix(voice): prevent activity race during session shutdown

.changeset/steady-otters-close.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Fix AgentSession shutdown when its active activity changes during asynchronous cleanup.

agents/src/voice/agent_session.ts

@@ -1281,15 +1281,18 @@ export class AgentSession<
           }
         }
 
-        if (this.closing && newActivity === 'start') {
+        if (this.closing) {
           this.logger.warn(
-            { agentId: this.nextActivity?.agent.id },
-            'Session is closing, skipping start of next activity',
+            { agentId: this.nextActivity?.agent.id, transition: newActivity },
+            'Session is closing, skipping activity transition',
           );
           if (reusableResources) {
             await cleanupReusableResources(reusableResources, this.logger);
             reusableResources = undefined;
           }
+          if (newActivity === 'resume') {
+            await this.nextActivity?.close();
+          }
           this.nextActivity = undefined;
           this.activity = undefined;
           return;
@@ -1704,25 +1707,26 @@ export class AgentSession<
     this._onAecWarmupExpired();
     this.off(AgentSessionEventTypes.UserInputTranscribed, this._onUserInputTranscribed);
 
-    if (this.activity) {
+    const activity = this.activity;
+    if (activity) {
       if (!drain) {
         try {
-          await this.activity.interrupt({ force: true }).await;
+          await activity.interrupt({ force: true }).await;
         } catch (error) {
           this.logger.warn({ error }, 'Error interrupting activity');
         }
       }
 
-      await this.activity.drain();
+      await activity.drain();
       // wait any uninterruptible speech to finish
-      await this.activity.currentSpeech?.waitForPlayout();
+      await activity.currentSpeech?.waitForPlayout();
 
       if (reason !== CloseReason.ERROR) {
-        this.activity.commitUserTurn({ audioDetached: true, throwIfNotReady: false });
+        activity.commitUserTurn({ audioDetached: true, throwIfNotReady: false });
       }
 
       try {
-        this.activity.detachAudioInput();
+        activity.detachAudioInput();
       } catch (error) {
         // Ignore detach errors during cleanup - source may not have been set
       }
@@ -1738,7 +1742,7 @@ export class AgentSession<
     this.output.audio = null;
     this.output.transcription = null;
 
-    await this.activity?.close();
+    await activity?.close();
     this.activity = undefined;
 
     const sessionToolsets = this._toolCtx.toolsets;

agents/src/voice/agent_session_close_race.test.ts

@@ -0,0 +1,49 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { describe, expect, it, vi } from 'vitest';
+import { initializeLogger } from '../log.js';
+import { Agent } from './agent.js';
+import type { AgentActivity } from './agent_activity.js';
+import { AgentSession } from './agent_session.js';
+import { AgentSessionEventTypes } from './events.js';
+
+describe('AgentSession close activity race', () => {
+  initializeLogger({ pretty: false, level: 'silent' });
+
+  it('finishes closing the captured activity when the session activity changes during drain', async () => {
+    const session = new AgentSession();
+    await session.start({ agent: new Agent({ instructions: 'test' }) });
+
+    const internals = session as unknown as { activity?: AgentActivity };
+    const activity = internals.activity!;
+
+    let releaseDrain!: () => void;
+    const drainGate = new Promise<void>((resolve) => {
+      releaseDrain = resolve;
+    });
+    let markDrainStarted!: () => void;
+    const drainStarted = new Promise<void>((resolve) => {
+      markDrainStarted = resolve;
+    });
+
+    vi.spyOn(activity, 'drain').mockImplementation(async () => {
+      markDrainStarted();
+      await drainGate;
+      return undefined;
+    });
+    const activityClose = vi.spyOn(activity, 'close');
+    const onClose = vi.fn();
+    session.on(AgentSessionEventTypes.Close, onClose);
+
+    const closePromise = session.close();
+    await drainStarted;
+
+    internals.activity = undefined;
+    releaseDrain();
+
+    await expect(closePromise).resolves.toBeUndefined();
+    expect(activityClose).toHaveBeenCalledOnce();
+    expect(onClose).toHaveBeenCalledOnce();
+  });
+});

agents/src/voice/agent_session_handoff.test.ts

@@ -259,4 +259,45 @@ describe('AgentSession reusable resources handoff', () => {
       startSpy.mockRestore();
     }
   });
+
+  it('skips resuming a paused activity while the session is closing and closes it', async () => {
+    const closeFn = vi.fn(async () => {});
+    const resources: ReusableResources = {
+      sttPipeline: { close: closeFn } as any,
+    };
+    const taskAgent = new Agent({ instructions: 'task' });
+    const previousActivity = {
+      agent: taskAgent,
+      drain: vi.fn(async () => resources),
+      close: vi.fn(async () => {}),
+      pause: vi.fn(async () => resources),
+    };
+    const resumedAgent = new Agent({ instructions: 'resumed' });
+    const resumedActivity = {
+      agent: resumedAgent,
+      resume: vi.fn(async () => {}),
+      start: vi.fn(async () => {}),
+      close: vi.fn(async () => {}),
+      attachAudioInput: vi.fn(),
+      _onEnterTask: undefined,
+    };
+    resumedAgent._agentActivity = resumedActivity as any;
+
+    const session = createFakeSession();
+    (session as any).activity = previousActivity as any;
+    (session as any).closing = true;
+
+    await AgentSession.prototype._updateActivity.call(session, resumedAgent, {
+      newActivity: 'resume',
+      waitOnEnter: false,
+    });
+
+    expect(previousActivity.drain).toHaveBeenCalledTimes(1);
+    expect(previousActivity.close).toHaveBeenCalledTimes(1);
+    expect(closeFn).toHaveBeenCalledTimes(1);
+    expect(resumedActivity.resume).not.toHaveBeenCalled();
+    expect(resumedActivity.close).toHaveBeenCalledTimes(1);
+    expect((session as any).activity).toBeUndefined();
+    expect((session as any).nextActivity).toBeUndefined();
+  });
 });