fix(agents): reject late inline tasks during handoff drain

.changeset/quiet-handoff-tasks.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Reject late inline AgentTasks on an outgoing activity so handoff and shutdown do not deadlock while draining non-cancellable tools.

agents/src/voice/agent_activity.ts

@@ -4997,6 +4997,15 @@ export class AgentActivity implements RecognitionHooks {
         if (this.closed || this.agentSession._closing) {
           throw new ToolError('the activity that awaited the inline task is closing');
         }
+        // A handoff holds the session transition lock while draining this task's owner.
+        // Check after acquiring the slot so a preceding inline task can resume us first.
+        if (this.newTurnsBlocked) {
+          throw new ToolError(
+            'An agent transition is in progress, so this tool call cannot continue. ' +
+              'Wait until the transition is complete before retrying, if the tool is ' +
+              'available to the new agent.',
+          );
+        }
 
         // A run must only watch this task once it has the slot. Otherwise it would wait
         // for user input needed by the task currently ahead of it.
@@ -5147,7 +5156,15 @@ export class AgentActivity implements RecognitionHooks {
 
       this.cancelPreemptiveGeneration();
 
-      await this._onExitTask.result;
+      try {
+        await this._onExitTask.result;
+      } catch (error) {
+        if (this._onExitTask.cancelled) throw error;
+        this.logger.error(
+          { 'lk.pii.error': error instanceof Error ? error.message : String(error) },
+          'error in agent onExit',
+        );
+      }
       await this._pauseSchedulingTask([]);
 
       // detach after speech tasks are done but before _closeSessionResources

agents/src/voice/agent_activity_preemptive_pause_deadlock.test.ts

@@ -165,8 +165,9 @@ describe('AgentActivity parked preemptive generation', () => {
     }
   });
 
-  it('does not let a reply parked while pausing block the handoff', async () => {
+  it('does not start a parked preemptive reply during onExit', async () => {
     const llm = new GatedLLM();
+    const chat = vi.spyOn(llm, 'chat');
     const session = new AgentSession({ llm });
     session.output.setAudioEnabled(false);
 
@@ -178,7 +179,8 @@ describe('AgentActivity parked preemptive generation', () => {
       override async onExit(): Promise<void> {
         const activity = session._activity! as ActivityInternals;
         startPreemptiveGeneration(activity);
-        expect(activity._preemptiveGeneration).toBeDefined();
+        expect(activity._preemptiveGeneration).toBeUndefined();
+        expect(chat).not.toHaveBeenCalled();
       }
     }
 
@@ -193,6 +195,7 @@ describe('AgentActivity parked preemptive generation', () => {
     } finally {
       llm.release();
       await session.close().catch(() => {});
+      chat.mockRestore();
     }
   });
 });

agents/src/voice/agent_session.ts

@@ -1503,6 +1503,7 @@ export class AgentSession<
               newActivity: this.nextActivity,
             });
           } else {
+            prevActivityObj.blockNewTurns();
             reusableResources = await prevActivityObj.drain({
               newActivity: this.nextActivity,
             });

agents/src/voice/agent_session_handoff.test.ts

@@ -61,6 +61,7 @@ describe('AgentSession reusable resources handoff', () => {
     const nextAgent = new Agent({ instructions: 'new' });
     const previousActivity = {
       agent: previousAgent,
+      blockNewTurns: vi.fn(),
       drain: vi.fn(async () => resources),
       close: vi.fn(async () => {}),
       pause: vi.fn(async () => resources),
@@ -95,6 +96,7 @@ describe('AgentSession reusable resources handoff', () => {
     const nextAgent = new Agent({ instructions: 'new' });
     const previousActivity = {
       agent: previousAgent,
+      blockNewTurns: vi.fn(),
       drain: vi.fn(async () => resources),
       close: vi.fn(async () => {}),
       pause: vi.fn(async () => resources),
@@ -132,6 +134,7 @@ describe('AgentSession reusable resources handoff', () => {
     const nextAgent = new Agent({ instructions: 'new' });
     const previousActivity = {
       agent: previousAgent,
+      blockNewTurns: vi.fn(),
       drain: vi.fn(async () => resources),
       close: vi.fn(async () => {}),
       pause: vi.fn(async () => resources),
@@ -195,6 +198,7 @@ describe('AgentSession reusable resources handoff', () => {
     const nextAgent = new Agent({ instructions: 'new' });
     const previousActivity = {
       agent: previousAgent,
+      blockNewTurns: vi.fn(),
       drain: vi.fn(async () => undefined),
       close: vi.fn(async () => {}),
       pause: vi.fn(async () => undefined),
@@ -228,6 +232,7 @@ describe('AgentSession reusable resources handoff', () => {
     const nextAgent = new Agent({ instructions: 'new' });
     const previousActivity = {
       agent: previousAgent,
+      blockNewTurns: vi.fn(),
       drain: vi.fn(async () => undefined),
       close: vi.fn(async () => {}),
       pause: vi.fn(async () => undefined),

agents/src/voice/agent_task_close.test.ts

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 import { expect, it, vi } from 'vitest';
 import { ToolError, tool } from '../llm/tool_context.js';
-import { Future } from '../utils.js';
+import { Future, Task } from '../utils.js';
 import { Agent, AgentTask } from './agent.js';
 import { AgentActivity } from './agent_activity.js';
 import { AgentSession } from './agent_session.js';
@@ -219,3 +219,79 @@ it.each([
   },
   5_000,
 );
+
+it.each(['success', 'failure'] as const)(
+  'resumes the parent after task onExit throws (result=%s)',
+  async (outcome) => {
+    const result = new Future<unknown>();
+    const failure = new ToolError('task failed');
+    const task = AgentTask.create<string>({
+      instructions: 'finish the task',
+      onEnter: () => task.complete(outcome === 'success' ? 'done' : failure),
+      onExit: async () => {
+        throw new Error('onExit failed');
+      },
+    });
+    const answered = vi.fn();
+    const agent = Agent.create({
+      instructions: 'parent',
+      tools: [
+        tool({
+          name: 'transfer',
+          description: 'Run the task.',
+          execute: async () => {
+            try {
+              const value = await task.run();
+              result.resolve(value);
+              return value;
+            } catch (error) {
+              result.resolve(error);
+              throw error;
+            }
+          },
+        }),
+        tool({ name: 'answer', description: 'Answer the next turn.', execute: answered }),
+      ],
+    });
+    const session = new AgentSession({
+      llm: new FakeLLM([
+        { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] },
+        { input: 'next turn', toolCalls: [{ name: 'answer', args: {} }] },
+      ]),
+      turnHandling: { turnDetection: 'manual' },
+    });
+    try {
+      await session.start({ agent });
+      const parentActivity = agent._agentActivity!;
+      session.generateReply({ userInput: 'transfer' });
+      expect(await result.await).toBe(outcome === 'success' ? 'done' : failure);
+      expect(session._activity).toBe(parentActivity);
+      expect(parentActivity.schedulingPaused).toBe(false);
+      expect(task._agentActivity).toBeUndefined();
+      await vi.waitFor(() => expect(parentActivity.currentSpeech).toBeUndefined());
+      session.generateReply({ userInput: 'next turn' });
+      await vi.waitFor(() => expect(answered).toHaveBeenCalledOnce());
+    } finally {
+      await session.close();
+    }
+  },
+);
+
+it('propagates onExit cancellation from drain', async () => {
+  const cancellation = new Error('onExit cancelled');
+  cancellation.name = 'AbortError';
+  const onExit = vi.fn().mockImplementationOnce(() => {
+    Task.current()!.cancel();
+    throw cancellation;
+  });
+  const agent = Agent.create({ instructions: 'parent', onExit });
+  const session = new AgentSession({ llm: new FakeLLM([]) });
+  try {
+    await session.start({ agent });
+    const activity = agent._agentActivity!;
+    await expect(activity.drain()).rejects.toBe(cancellation);
+    expect(activity.schedulingPaused).toBe(false);
+  } finally {
+    await session.close();
+  }
+});