diff --git a/.changeset/steady-otters-close.md b/.changeset/steady-otters-close.md new file mode 100644 index 000000000..95d9b6d77 --- /dev/null +++ b/.changeset/steady-otters-close.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Fix AgentSession shutdown when its active activity changes during asynchronous cleanup. diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index c317d019d..3e613ae3e 100644 --- a/agents/src/voice/agent_session.ts +++ b/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; diff --git a/agents/src/voice/agent_session_close_race.test.ts b/agents/src/voice/agent_session_close_race.test.ts new file mode 100644 index 000000000..950894ca7 --- /dev/null +++ b/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((resolve) => { + releaseDrain = resolve; + }); + let markDrainStarted!: () => void; + const drainStarted = new Promise((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(); + }); +}); diff --git a/agents/src/voice/agent_session_handoff.test.ts b/agents/src/voice/agent_session_handoff.test.ts index 5690f7412..b12544fe6 100644 --- a/agents/src/voice/agent_session_handoff.test.ts +++ b/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(); + }); });