From ff9fe548386d4ecb1d42538c4414983edbf796b8 Mon Sep 17 00:00:00 2001 From: Swayam Gupta Date: Sun, 13 Sep 2026 04:37:08 +0530 Subject: [PATCH 1/6] fix(agents): reject late inline tasks during handoff drain Reject inline task admission on an outgoing activity before it can wait for the session transition lock held by handoff. Preserve normal sibling task pause/resume and non-cancellable tool behavior. Add five regression cases covering sequential completion, both handoff windows, and concurrent shutdown, plus an agents patch changeset. Fixes #2483 --- .changeset/quiet-handoff-tasks.md | 5 + agents/src/voice/agent_activity.ts | 5 + .../voice/agent_task_handoff_drain.test.ts | 151 ++++++++++++++++++ 3 files changed, 161 insertions(+) create mode 100644 .changeset/quiet-handoff-tasks.md create mode 100644 agents/src/voice/agent_task_handoff_drain.test.ts diff --git a/.changeset/quiet-handoff-tasks.md b/.changeset/quiet-handoff-tasks.md new file mode 100644 index 000000000..945d7fbb1 --- /dev/null +++ b/.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. diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 153ef000e..c2ade7181 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -4997,6 +4997,11 @@ 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 || this.schedulingPaused) { + throw new ToolError('the activity that awaited the inline task is draining'); + } // 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. diff --git a/agents/src/voice/agent_task_handoff_drain.test.ts b/agents/src/voice/agent_task_handoff_drain.test.ts new file mode 100644 index 000000000..2914fd382 --- /dev/null +++ b/agents/src/voice/agent_task_handoff_drain.test.ts @@ -0,0 +1,151 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { expect, it, vi } from 'vitest'; +import { FunctionCall } from '../llm/chat_context.js'; +import { ToolError, handoff, tool } from '../llm/tool_context.js'; +import { Future } from '../utils.js'; +import { Agent, AgentTask } from './agent.js'; +import { AgentSession } from './agent_session.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +// The two turns must not reuse FakeLLM's default call ID while a tool is still running. +class UniqueCallLLM extends FakeLLM { + private generation = 0; + + override chat(options: Parameters[0]) { + const stream = super.chat(options); + const generation = ++this.generation; + const next = stream.next.bind(stream); + stream.next = async () => { + const result = await next(); + if (!result.done && result.value.delta?.toolCalls) { + result.value.delta.toolCalls = result.value.delta.toolCalls.map((call) => + FunctionCall.create({ ...call, callId: `${generation}_${call.callId}` }), + ); + } + return result; + }; + return stream; + } +} + +it.each([ + { timing: 'before handoff', shutdown: false }, + { timing: 'during handoff onExit', shutdown: false }, + { timing: 'during handoff drain', shutdown: false }, + { timing: 'during handoff onExit', shutdown: true }, + { timing: 'during handoff drain', shutdown: true }, +] as const)( + 'settles a non-cancellable tool that starts an AgentTask $timing (shutdown=$shutdown)', + async ({ timing, shutdown }) => { + const admission = new Future(); + const started = new Future(); + const result = new Future(); + const exiting = new Future(); + const finishExit = new Future(); + const finishTool = new Future(); + const taskEntered = vi.fn(); + const targetEntered = vi.fn(); + const target = Agent.create({ instructions: 'target', onEnter: targetEntered }); + const task = AgentTask.create({ + instructions: 'complete immediately', + onEnter: async () => { + taskEntered(); + task.complete('done'); + }, + }); + const agent = Agent.create({ + instructions: 'source', + onExit: async () => { + exiting.resolve(); + if (timing === 'during handoff onExit') await finishExit.await; + }, + tools: [ + tool({ + name: 'transfer', + description: 'Run an inline task after admission.', + execute: async (_args, { ctx }) => { + ctx.speechHandle.allowInterruptions = false; + started.resolve(); + await admission.await; + try { + const value = await task.run(); + result.resolve(value); + return value; + } catch (error) { + result.resolve(error); + if (shutdown) await finishTool.await; + throw error; + } + }, + }), + tool({ + name: 'switch', + description: 'Hand off to the target.', + execute: async () => handoff({ agent: target }), + }), + ], + }); + const session = new AgentSession({ + llm: new UniqueCallLLM([ + { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] }, + { input: 'switch', toolCalls: [{ name: 'switch', args: {} }] }, + ]), + turnHandling: { turnDetection: 'manual' }, + }); + let closing: Promise | undefined; + + try { + await session.start({ agent }); + const sourceActivity = agent._agentActivity!; + session.generateReply({ userInput: 'transfer' }); + await started.await; + await vi.waitFor(() => expect(sourceActivity.currentSpeech).toBeUndefined()); + + if (timing === 'before handoff') { + admission.resolve(); + expect(await result.await).toBe('done'); + } + + session.generateReply({ userInput: 'switch' }); + if (timing !== 'before handoff') { + if (timing === 'during handoff onExit') { + await exiting.await; + expect(sourceActivity.schedulingPaused).toBe(false); + } else { + await vi.waitFor(() => expect(sourceActivity.schedulingPaused).toBe(true)); + } + admission.resolve(); + expect(await result.await).toBeInstanceOf(ToolError); + expect(await result.await).toHaveProperty( + 'message', + 'the activity that awaited the inline task is draining', + ); + expect(taskEntered).not.toHaveBeenCalled(); + if (shutdown) closing = session.close(); + finishTool.resolve(); + finishExit.resolve(); + } else { + expect(taskEntered).toHaveBeenCalledOnce(); + } + + if (shutdown) { + await closing; + expect(targetEntered).not.toHaveBeenCalled(); + } else { + await vi.waitFor(() => expect(targetEntered).toHaveBeenCalledOnce()); + expect(session.currentAgent).toBe(target); + await session.close(); + } + expect(agent._agentActivity).toBeUndefined(); + expect(task._agentActivity).toBeUndefined(); + expect(target._agentActivity).toBeUndefined(); + } finally { + admission.resolve(); + finishTool.resolve(); + finishExit.resolve(); + await session.close(); + } + }, +); From c2419ee9063a7aa72e31091e4eb57b829d7a4cdb Mon Sep 17 00:00:00 2001 From: Swayam Gupta Date: Sun, 13 Sep 2026 05:14:13 +0530 Subject: [PATCH 2/6] fix(agents): block task admission before queued handoff exit Make every draining activity block new turns before onExit, including intermediate activities that were not current when updateAgent calls were queued. Add queued handoff and shutdown regressions, and verify that drain prevents new preemptive generation while preserving parked-reply cancellation coverage. --- agents/src/voice/agent_activity.ts | 4 +++ ...activity_preemptive_pause_deadlock.test.ts | 7 +++-- .../voice/agent_task_handoff_drain.test.ts | 27 ++++++++++++++----- 3 files changed, 30 insertions(+), 8 deletions(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index c2ade7181..f85d8a152 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -5140,6 +5140,10 @@ export class AgentActivity implements RecognitionHooks { try { if (this._schedulingPaused) return undefined; + // Queued handoffs may have blocked an earlier activity, not this one. + // Close admission before onExit can start an inline task that awaits this drain. + this.blockNewTurns(); + this._onExitTask = this.createSpeechTask({ taskFn: () => tracer.startActiveSpan(async () => this.agent.onExit(), { diff --git a/agents/src/voice/agent_activity_preemptive_pause_deadlock.test.ts b/agents/src/voice/agent_activity_preemptive_pause_deadlock.test.ts index 94f82adf1..9c46d5bd7 100644 --- a/agents/src/voice/agent_activity_preemptive_pause_deadlock.test.ts +++ b/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 { 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(); } }); }); diff --git a/agents/src/voice/agent_task_handoff_drain.test.ts b/agents/src/voice/agent_task_handoff_drain.test.ts index 2914fd382..22c12398a 100644 --- a/agents/src/voice/agent_task_handoff_drain.test.ts +++ b/agents/src/voice/agent_task_handoff_drain.test.ts @@ -36,9 +36,13 @@ it.each([ { timing: 'during handoff drain', shutdown: false }, { timing: 'during handoff onExit', shutdown: true }, { timing: 'during handoff drain', shutdown: true }, + { timing: 'during queued handoff onExit', shutdown: false }, + { timing: 'during queued handoff onExit', shutdown: true }, ] as const)( 'settles a non-cancellable tool that starts an AgentTask $timing (shutdown=$shutdown)', async ({ timing, shutdown }) => { + const queued = timing === 'during queued handoff onExit'; + const duringOnExit = timing === 'during handoff onExit' || queued; const admission = new Future(); const started = new Future(); const result = new Future(); @@ -47,6 +51,7 @@ it.each([ const finishTool = new Future(); const taskEntered = vi.fn(); const targetEntered = vi.fn(); + const root = Agent.create({ instructions: 'root' }); const target = Agent.create({ instructions: 'target', onEnter: targetEntered }); const task = AgentTask.create({ instructions: 'complete immediately', @@ -57,9 +62,12 @@ it.each([ }); const agent = Agent.create({ instructions: 'source', + onEnter: async () => { + if (queued) session.generateReply({ userInput: 'transfer' }); + }, onExit: async () => { exiting.resolve(); - if (timing === 'during handoff onExit') await finishExit.await; + if (duringOnExit) await finishExit.await; }, tools: [ tool({ @@ -97,10 +105,16 @@ it.each([ let closing: Promise | undefined; try { - await session.start({ agent }); - const sourceActivity = agent._agentActivity!; - session.generateReply({ userInput: 'transfer' }); + await session.start({ agent: queued ? root : agent }); + if (queued) { + // Both requests initially block root, not the intermediate source activity. + session.updateAgent(agent); + session.updateAgent(target); + } else { + session.generateReply({ userInput: 'transfer' }); + } await started.await; + const sourceActivity = agent._agentActivity!; await vi.waitFor(() => expect(sourceActivity.currentSpeech).toBeUndefined()); if (timing === 'before handoff') { @@ -108,9 +122,9 @@ it.each([ expect(await result.await).toBe('done'); } - session.generateReply({ userInput: 'switch' }); + if (!queued) session.generateReply({ userInput: 'switch' }); if (timing !== 'before handoff') { - if (timing === 'during handoff onExit') { + if (duringOnExit) { await exiting.await; expect(sourceActivity.schedulingPaused).toBe(false); } else { @@ -139,6 +153,7 @@ it.each([ await session.close(); } expect(agent._agentActivity).toBeUndefined(); + expect(root._agentActivity).toBeUndefined(); expect(task._agentActivity).toBeUndefined(); expect(target._agentActivity).toBeUndefined(); } finally { From 054fb017f85f1fd6a54b7323ed84ce98c6dc8a44 Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Tue, 15 Sep 2026 10:59:30 +0100 Subject: [PATCH 3/6] fix: align handoff task admission with Python Block inline tasks only during agent transitions, so standalone drain can still finish an in-flight AgentTask. Match the transition error and add the applicable regression cases from livekit/agents#7268. Addresses AGT-3502 Refs livekit/agents-js#2483 Co-authored-by: Dan Tran --- agents/src/voice/agent_activity.ts | 12 +- agents/src/voice/agent_session.ts | 1 + .../src/voice/agent_session_handoff.test.ts | 5 + .../voice/agent_task_handoff_drain.test.ts | 194 ++++++++++++++++-- 4 files changed, 192 insertions(+), 20 deletions(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index f85d8a152..26c4d4a7e 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -4999,8 +4999,12 @@ export class AgentActivity implements RecognitionHooks { } // 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 || this.schedulingPaused) { - throw new ToolError('the activity that awaited the inline task is draining'); + 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 @@ -5140,10 +5144,6 @@ export class AgentActivity implements RecognitionHooks { try { if (this._schedulingPaused) return undefined; - // Queued handoffs may have blocked an earlier activity, not this one. - // Close admission before onExit can start an inline task that awaits this drain. - this.blockNewTurns(); - this._onExitTask = this.createSpeechTask({ taskFn: () => tracer.startActiveSpan(async () => this.agent.onExit(), { diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index 2a1faf914..138197c45 100644 --- a/agents/src/voice/agent_session.ts +++ b/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, }); diff --git a/agents/src/voice/agent_session_handoff.test.ts b/agents/src/voice/agent_session_handoff.test.ts index c0585ccfc..a0e4e2ec8 100644 --- a/agents/src/voice/agent_session_handoff.test.ts +++ b/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), diff --git a/agents/src/voice/agent_task_handoff_drain.test.ts b/agents/src/voice/agent_task_handoff_drain.test.ts index 22c12398a..9079db057 100644 --- a/agents/src/voice/agent_task_handoff_drain.test.ts +++ b/agents/src/voice/agent_task_handoff_drain.test.ts @@ -9,6 +9,11 @@ import { Agent, AgentTask } from './agent.js'; import { AgentSession } from './agent_session.js'; import { FakeLLM } from './testing/fake_llm.js'; +const transitionError = + '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.'; + // The two turns must not reuse FakeLLM's default call ID while a tool is still running. class UniqueCallLLM extends FakeLLM { private generation = 0; @@ -32,6 +37,9 @@ class UniqueCallLLM extends FakeLLM { it.each([ { timing: 'before handoff', shutdown: false }, + { timing: 'handoff requested', shutdown: false }, + { timing: 'direct handoff', shutdown: false }, + { timing: 'tool starts during drain', shutdown: false }, { timing: 'during handoff onExit', shutdown: false }, { timing: 'during handoff drain', shutdown: false }, { timing: 'during handoff onExit', shutdown: true }, @@ -42,7 +50,12 @@ it.each([ 'settles a non-cancellable tool that starts an AgentTask $timing (shutdown=$shutdown)', async ({ timing, shutdown }) => { const queued = timing === 'during queued handoff onExit'; - const duringOnExit = timing === 'during handoff onExit' || queued; + const duringOnExit = + timing === 'during handoff onExit' || timing === 'direct handoff' || queued; + const switchStarted = new Future(); + const finishSwitch = new Future(); + const generationStarted = new Future(); + const emitTool = new Future(); const admission = new Future(); const started = new Future(); const result = new Future(); @@ -91,18 +104,42 @@ it.each([ tool({ name: 'switch', description: 'Hand off to the target.', - execute: async () => handoff({ agent: target }), + execute: async () => { + switchStarted.resolve(); + if (timing === 'tool starts during drain') await finishSwitch.await; + return handoff({ agent: target }); + }, }), ], }); + const llm = new UniqueCallLLM([ + { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] }, + { input: 'switch', toolCalls: [{ name: 'switch', args: {} }] }, + ]); + const chat = llm.chat.bind(llm); + llm.chat = (options) => { + const stream = chat(options); + const next = stream.next.bind(stream); + stream.next = async () => { + const chunk = await next(); + if ( + timing === 'tool starts during drain' && + !chunk.done && + chunk.value.delta?.toolCalls?.some((call) => call.name === 'transfer') + ) { + generationStarted.resolve(); + await emitTool.await; + } + return chunk; + }; + return stream; + }; const session = new AgentSession({ - llm: new UniqueCallLLM([ - { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] }, - { input: 'switch', toolCalls: [{ name: 'switch', args: {} }] }, - ]), + llm, turnHandling: { turnDetection: 'manual' }, }); let closing: Promise | undefined; + let transition: Promise | undefined; try { await session.start({ agent: queued ? root : agent }); @@ -111,31 +148,58 @@ it.each([ session.updateAgent(agent); session.updateAgent(target); } else { - session.generateReply({ userInput: 'transfer' }); + if (timing === 'tool starts during drain') { + session.generateReply({ userInput: 'switch' }); + await switchStarted.await; + await vi.waitFor(() => expect(agent._agentActivity!.currentSpeech).toBeUndefined()); + } + session.generateReply({ userInput: 'transfer', allowInterruptions: false }); + if (timing === 'tool starts during drain') { + await generationStarted.await; + finishSwitch.resolve(); + await vi.waitFor(() => expect(agent._agentActivity!.schedulingPaused).toBe(true)); + expect(started.done).toBe(false); + emitTool.resolve(); + } } await started.await; const sourceActivity = agent._agentActivity!; - await vi.waitFor(() => expect(sourceActivity.currentSpeech).toBeUndefined()); + if (timing !== 'tool starts during drain') { + await vi.waitFor(() => expect(sourceActivity.currentSpeech).toBeUndefined()); + } if (timing === 'before handoff') { admission.resolve(); expect(await result.await).toBe('done'); } - if (!queued) session.generateReply({ userInput: 'switch' }); + if (timing === 'handoff requested') { + // Keep the handoff queued while the tool attempts its inline task. + const unlock = await session['activityLock'].lock(); + try { + session.updateAgent(target); + admission.resolve(); + expect(await result.await).toBeInstanceOf(ToolError); + expect(exiting.done).toBe(false); + expect(sourceActivity.schedulingPaused).toBe(false); + } finally { + unlock(); + } + } else if (timing === 'direct handoff') { + transition = session._updateActivity(target, { waitOnEnter: false }); + } else if (!queued && timing !== 'tool starts during drain') { + session.generateReply({ userInput: 'switch' }); + } if (timing !== 'before handoff') { if (duringOnExit) { await exiting.await; expect(sourceActivity.schedulingPaused).toBe(false); - } else { + } else if (timing !== 'handoff requested') { await vi.waitFor(() => expect(sourceActivity.schedulingPaused).toBe(true)); } admission.resolve(); expect(await result.await).toBeInstanceOf(ToolError); - expect(await result.await).toHaveProperty( - 'message', - 'the activity that awaited the inline task is draining', - ); + expect(await result.await).toHaveProperty('message', transitionError); expect(taskEntered).not.toHaveBeenCalled(); if (shutdown) closing = session.close(); finishTool.resolve(); @@ -152,15 +216,117 @@ it.each([ expect(session.currentAgent).toBe(target); await session.close(); } + if (timing !== 'before handoff' && !shutdown) { + const calls = agent.chatCtx.items.filter( + (item) => item.type === 'function_call' && item.name === 'transfer', + ); + expect(calls).toHaveLength(1); + const outputs = agent.chatCtx.items.filter( + (item) => item.type === 'function_call_output' && item.callId === calls[0]!.callId, + ); + expect(outputs).toHaveLength(1); + expect(outputs[0]).toMatchObject({ isError: true, output: transitionError }); + } expect(agent._agentActivity).toBeUndefined(); expect(root._agentActivity).toBeUndefined(); expect(task._agentActivity).toBeUndefined(); expect(target._agentActivity).toBeUndefined(); } finally { + emitTool.resolve(); + finishSwitch.resolve(); admission.resolve(); finishTool.resolve(); finishExit.resolve(); await session.close(); + await transition; + } + }, +); + +it.each([false, true])( + 'allows an in-flight tool to await an AgentTask during standalone drain (generationPending=%s)', + async (generationPending) => { + const started = new Future(); + const releaseTool = new Future(); + const releaseGeneration = new Future(); + const completed = new Future(); + let drained = false; + let draining: Promise | undefined; + const task = AgentTask.create({ + instructions: 'complete immediately', + onEnter: () => { + expect(drained).toBe(true); + task.complete('done'); + }, + }); + const agent = Agent.create({ + instructions: 'source', + tools: [ + tool({ + name: 'transfer', + description: 'Run an inline task.', + execute: async (_args, { ctx }) => { + ctx.speechHandle.allowInterruptions = false; + started.resolve(); + await releaseTool.await; + try { + const value = await task.run(); + completed.resolve(value); + return value; + } catch (error) { + completed.resolve(error); + throw error; + } + }, + }), + ], + }); + const llm = new UniqueCallLLM([ + { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] }, + ]); + const chat = llm.chat.bind(llm); + llm.chat = (options) => { + const stream = chat(options); + const next = stream.next.bind(stream); + stream.next = async () => { + const chunk = await next(); + if (chunk.done) await releaseGeneration.await; + return chunk; + }; + return stream; + }; + const session = new AgentSession({ llm, turnHandling: { turnDetection: 'manual' } }); + try { + await session.start({ agent }); + const activity = agent._agentActivity!; + const speech = session.generateReply({ userInput: 'transfer' }); + await started.await; + if (!generationPending) { + releaseGeneration.resolve(); + await vi.waitFor(() => expect(activity.currentSpeech).toBeUndefined()); + } + + // Node has no public session.drain(); exercise the same activity drain directly. + draining = activity.drain().then(() => { + drained = true; + }); + await vi.waitFor(() => expect(activity.schedulingPaused).toBe(true)); + if (generationPending) expect(activity.currentSpeech).toBe(speech); + releaseTool.resolve(); + if (generationPending) { + await vi.waitFor(() => expect(activity['_drainBlockedTasks'].size).toBeGreaterThan(0)); + expect(drained).toBe(false); + releaseGeneration.resolve(); + } + await draining; + expect(await completed.await).toBe('done'); + expect(session._activity).toBe(activity); + expect(activity.schedulingPaused).toBe(false); + } finally { + releaseTool.resolve(); + releaseGeneration.resolve(); + await draining; + await session.close(); } }, ); From 2633f0466a84e34a79f39ea15cf64f9d564c1f0d Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Tue, 15 Sep 2026 11:14:40 +0100 Subject: [PATCH 4/6] test: separate handoff regression timelines Give each handoff trigger its own event sequence. Share the fixture and keep shutdown variants paired so admission, drain, and result ordering remain visible. Addresses AGT-3502 --- .../voice/agent_task_handoff_drain.test.ts | 468 ++++++++++-------- 1 file changed, 273 insertions(+), 195 deletions(-) diff --git a/agents/src/voice/agent_task_handoff_drain.test.ts b/agents/src/voice/agent_task_handoff_drain.test.ts index 9079db057..a703464b2 100644 --- a/agents/src/voice/agent_task_handoff_drain.test.ts +++ b/agents/src/voice/agent_task_handoff_drain.test.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2026 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import { expect, it, vi } from 'vitest'; +import { expect, it, onTestFinished, vi } from 'vitest'; import { FunctionCall } from '../llm/chat_context.js'; import { ToolError, handoff, tool } from '../llm/tool_context.js'; import { Future } from '../utils.js'; @@ -35,210 +35,288 @@ class UniqueCallLLM extends FakeLLM { } } -it.each([ - { timing: 'before handoff', shutdown: false }, - { timing: 'handoff requested', shutdown: false }, - { timing: 'direct handoff', shutdown: false }, - { timing: 'tool starts during drain', shutdown: false }, - { timing: 'during handoff onExit', shutdown: false }, - { timing: 'during handoff drain', shutdown: false }, - { timing: 'during handoff onExit', shutdown: true }, - { timing: 'during handoff drain', shutdown: true }, - { timing: 'during queued handoff onExit', shutdown: false }, - { timing: 'during queued handoff onExit', shutdown: true }, -] as const)( - 'settles a non-cancellable tool that starts an AgentTask $timing (shutdown=$shutdown)', - async ({ timing, shutdown }) => { - const queued = timing === 'during queued handoff onExit'; - const duringOnExit = - timing === 'during handoff onExit' || timing === 'direct handoff' || queued; - const switchStarted = new Future(); - const finishSwitch = new Future(); - const generationStarted = new Future(); - const emitTool = new Future(); - const admission = new Future(); - const started = new Future(); - const result = new Future(); - const exiting = new Future(); - const finishExit = new Future(); - const finishTool = new Future(); - const taskEntered = vi.fn(); - const targetEntered = vi.fn(); - const root = Agent.create({ instructions: 'root' }); - const target = Agent.create({ instructions: 'target', onEnter: targetEntered }); - const task = AgentTask.create({ - instructions: 'complete immediately', - onEnter: async () => { - taskEntered(); - task.complete('done'); - }, - }); - const agent = Agent.create({ - instructions: 'source', - onEnter: async () => { - if (queued) session.generateReply({ userInput: 'transfer' }); - }, - onExit: async () => { - exiting.resolve(); - if (duringOnExit) await finishExit.await; - }, - tools: [ - tool({ - name: 'transfer', - description: 'Run an inline task after admission.', - execute: async (_args, { ctx }) => { - ctx.speechHandle.allowInterruptions = false; - started.resolve(); - await admission.await; - try { - const value = await task.run(); - result.resolve(value); - return value; - } catch (error) { - result.resolve(error); - if (shutdown) await finishTool.await; - throw error; - } - }, - }), - tool({ - name: 'switch', - description: 'Hand off to the target.', - execute: async () => { - switchStarted.resolve(); - if (timing === 'tool starts during drain') await finishSwitch.await; - return handoff({ agent: target }); - }, - }), - ], - }); - const llm = new UniqueCallLLM([ - { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] }, - { input: 'switch', toolCalls: [{ name: 'switch', args: {} }] }, - ]); - const chat = llm.chat.bind(llm); - llm.chat = (options) => { - const stream = chat(options); - const next = stream.next.bind(stream); - stream.next = async () => { - const chunk = await next(); - if ( - timing === 'tool starts during drain' && - !chunk.done && - chunk.value.delta?.toolCalls?.some((call) => call.name === 'transfer') - ) { - generationStarted.resolve(); - await emitTool.await; - } - return chunk; - }; - return stream; - }; - const session = new AgentSession({ - llm, - turnHandling: { turnDetection: 'manual' }, - }); - let closing: Promise | undefined; - let transition: Promise | undefined; +function createFixture(toolErrorGate?: Future) { + const admission = new Future(); + const started = new Future(); + const result = new Future(); + const exiting = new Future(); + const finishExit = new Future(); + const switchStarted = new Future(); + const finishSwitch = new Future(); + const taskEntered = vi.fn(); + const targetEntered = vi.fn(); + const target = Agent.create({ instructions: 'target', onEnter: targetEntered }); + const task = AgentTask.create({ + instructions: 'complete immediately', + onEnter: () => { + taskEntered(); + task.complete('done'); + }, + }); + const agent = Agent.create({ + instructions: 'source', + onExit: async () => { + exiting.resolve(); + await finishExit.await; + }, + tools: [ + tool({ + name: 'transfer', + description: 'Run an inline task after admission.', + execute: async (_args, { ctx }) => { + ctx.speechHandle.allowInterruptions = false; + started.resolve(); + await admission.await; + try { + const value = await task.run(); + result.resolve(value); + return value; + } catch (error) { + result.resolve(error); + if (toolErrorGate) await toolErrorGate.await; + throw error; + } + }, + }), + tool({ + name: 'switch', + description: 'Hand off to the target.', + execute: async () => { + switchStarted.resolve(); + await finishSwitch.await; + return handoff({ agent: target }); + }, + }), + ], + }); + const llm = new UniqueCallLLM([ + { input: 'transfer', toolCalls: [{ name: 'transfer', args: {} }] }, + { input: 'switch', toolCalls: [{ name: 'switch', args: {} }] }, + ]); + const session = new AgentSession({ llm, turnHandling: { turnDetection: 'manual' } }); + onTestFinished(async () => { + admission.resolve(); + finishExit.resolve(); + toolErrorGate?.resolve(); + finishSwitch.resolve(); + await session.close(); + expect(agent._agentActivity).toBeUndefined(); + expect(task._agentActivity).toBeUndefined(); + expect(target._agentActivity).toBeUndefined(); + }); + return { + session, + agent, + target, + taskEntered, + targetEntered, + llm, + admission, + started, + result, + exiting, + finishExit, + switchStarted, + finishSwitch, + }; +} - try { - await session.start({ agent: queued ? root : agent }); - if (queued) { - // Both requests initially block root, not the intermediate source activity. - session.updateAgent(agent); - session.updateAgent(target); - } else { - if (timing === 'tool starts during drain') { - session.generateReply({ userInput: 'switch' }); - await switchStarted.await; - await vi.waitFor(() => expect(agent._agentActivity!.currentSpeech).toBeUndefined()); - } - session.generateReply({ userInput: 'transfer', allowInterruptions: false }); - if (timing === 'tool starts during drain') { - await generationStarted.await; - finishSwitch.resolve(); - await vi.waitFor(() => expect(agent._agentActivity!.schedulingPaused).toBe(true)); - expect(started.done).toBe(false); - emitTool.resolve(); - } - } - await started.await; - const sourceActivity = agent._agentActivity!; - if (timing !== 'tool starts during drain') { - await vi.waitFor(() => expect(sourceActivity.currentSpeech).toBeUndefined()); - } +type Fixture = ReturnType; - if (timing === 'before handoff') { - admission.resolve(); - expect(await result.await).toBe('done'); - } +async function startPendingTool(f: Fixture) { + await f.session.start({ agent: f.agent }); + f.session.generateReply({ userInput: 'transfer', allowInterruptions: false }); + await f.started.await; + await vi.waitFor(() => expect(f.agent._agentActivity!.currentSpeech).toBeUndefined()); +} - if (timing === 'handoff requested') { - // Keep the handoff queued while the tool attempts its inline task. - const unlock = await session['activityLock'].lock(); - try { - session.updateAgent(target); - admission.resolve(); - expect(await result.await).toBeInstanceOf(ToolError); - expect(exiting.done).toBe(false); - expect(sourceActivity.schedulingPaused).toBe(false); - } finally { - unlock(); - } - } else if (timing === 'direct handoff') { - transition = session._updateActivity(target, { waitOnEnter: false }); - } else if (!queued && timing !== 'tool starts during drain') { - session.generateReply({ userInput: 'switch' }); - } - if (timing !== 'before handoff') { - if (duringOnExit) { - await exiting.await; - expect(sourceActivity.schedulingPaused).toBe(false); - } else if (timing !== 'handoff requested') { - await vi.waitFor(() => expect(sourceActivity.schedulingPaused).toBe(true)); - } - admission.resolve(); - expect(await result.await).toBeInstanceOf(ToolError); - expect(await result.await).toHaveProperty('message', transitionError); - expect(taskEntered).not.toHaveBeenCalled(); - if (shutdown) closing = session.close(); - finishTool.resolve(); - finishExit.resolve(); - } else { - expect(taskEntered).toHaveBeenCalledOnce(); +async function expectRejected(f: Fixture) { + expect(await f.result.await).toBeInstanceOf(ToolError); + expect(await f.result.await).toHaveProperty('message', transitionError); + expect(f.taskEntered).not.toHaveBeenCalled(); +} + +async function expectHandoff(f: Fixture) { + await vi.waitFor(() => expect(f.targetEntered).toHaveBeenCalledOnce()); + expect(f.session.currentAgent).toBe(f.target); + await f.session.close(); + const calls = f.agent.chatCtx.items.filter( + (item) => item.type === 'function_call' && item.name === 'transfer', + ); + expect(calls).toHaveLength(1); + const outputs = f.agent.chatCtx.items.filter( + (item) => item.type === 'function_call_output' && item.callId === calls[0]!.callId, + ); + expect(outputs).toHaveLength(1); + expect(outputs[0]).toMatchObject({ isError: true, output: transitionError }); +} + +it('completes an inline task before handoff', async () => { + const f = createFixture(); + await startPendingTool(f); + f.admission.resolve(); + expect(await f.result.await).toBe('done'); + expect(f.taskEntered).toHaveBeenCalledOnce(); + f.finishSwitch.resolve(); + f.finishExit.resolve(); + f.session.generateReply({ userInput: 'switch' }); + await vi.waitFor(() => expect(f.targetEntered).toHaveBeenCalledOnce()); + expect(f.session.currentAgent).toBe(f.target); +}); + +it('rejects an inline task while handoff waits for the transition lock', async () => { + const f = createFixture(); + await startPendingTool(f); + const unlock = await f.session['activityLock'].lock(); + try { + f.session.updateAgent(f.target); + f.admission.resolve(); + await expectRejected(f); + expect(f.exiting.done).toBe(false); + expect(f.agent._agentActivity!.schedulingPaused).toBe(false); + } finally { + unlock(); + } + f.finishExit.resolve(); + await expectHandoff(f); +}); + +it('rejects an inline task during a direct activity transition', async () => { + const f = createFixture(); + await startPendingTool(f); + const transition = f.session._updateActivity(f.target, { waitOnEnter: false }); + try { + await f.exiting.await; + expect(f.agent._agentActivity!.schedulingPaused).toBe(false); + f.admission.resolve(); + await expectRejected(f); + } finally { + f.admission.resolve(); + f.finishExit.resolve(); + await transition; + } + await expectHandoff(f); +}); + +it('rejects a tool that starts after handoff drain has begun', async () => { + const f = createFixture(); + const generationStarted = new Future(); + const emitTool = new Future(); + const chat = f.llm.chat.bind(f.llm); + const chatSpy = vi.spyOn(f.llm, 'chat').mockImplementation((options) => { + const stream = chat(options); + const next = stream.next.bind(stream); + stream.next = async () => { + const chunk = await next(); + if (!chunk.done && chunk.value.delta?.toolCalls?.some((call) => call.name === 'transfer')) { + generationStarted.resolve(); + await emitTool.await; } + return chunk; + }; + return stream; + }); + try { + await f.session.start({ agent: f.agent }); + f.session.generateReply({ userInput: 'switch' }); + await f.switchStarted.await; + await vi.waitFor(() => expect(f.agent._agentActivity!.currentSpeech).toBeUndefined()); + f.session.generateReply({ userInput: 'transfer', allowInterruptions: false }); + await generationStarted.await; + f.finishSwitch.resolve(); + f.finishExit.resolve(); + await vi.waitFor(() => expect(f.agent._agentActivity!.schedulingPaused).toBe(true)); + expect(f.started.done).toBe(false); + emitTool.resolve(); + await f.started.await; + f.admission.resolve(); + await expectRejected(f); + await expectHandoff(f); + } finally { + emitTool.resolve(); + chatSpy.mockRestore(); + } +}); - if (shutdown) { +it.each([false, true])( + 'rejects an inline task during handoff onExit (shutdown=%s)', + async (shutdown) => { + const finishTool = new Future(); + const f = createFixture(shutdown ? finishTool : undefined); + await startPendingTool(f); + f.finishSwitch.resolve(); + f.session.generateReply({ userInput: 'switch' }); + await f.exiting.await; + expect(f.agent._agentActivity!.schedulingPaused).toBe(false); + f.admission.resolve(); + await expectRejected(f); + const closing = shutdown ? f.session.close() : undefined; + finishTool.resolve(); + f.finishExit.resolve(); + if (closing) { + await closing; + expect(f.targetEntered).not.toHaveBeenCalled(); + } else { + await expectHandoff(f); + } + }, +); + +it.each([false, true])( + 'rejects an inline task during handoff drain (shutdown=%s)', + async (shutdown) => { + const finishTool = new Future(); + const f = createFixture(shutdown ? finishTool : undefined); + await startPendingTool(f); + f.finishSwitch.resolve(); + f.finishExit.resolve(); + f.session.generateReply({ userInput: 'switch' }); + await vi.waitFor(() => expect(f.agent._agentActivity!.schedulingPaused).toBe(true)); + f.admission.resolve(); + await expectRejected(f); + const closing = shutdown ? f.session.close() : undefined; + finishTool.resolve(); + if (closing) { + await closing; + expect(f.targetEntered).not.toHaveBeenCalled(); + } else { + await expectHandoff(f); + } + }, +); + +it.each([false, true])( + 'rejects an inline task on a queued intermediate agent (shutdown=%s)', + async (shutdown) => { + const finishTool = new Future(); + const f = createFixture(shutdown ? finishTool : undefined); + const root = Agent.create({ instructions: 'root' }); + const enter = vi.spyOn(f.agent, 'onEnter').mockImplementation(async () => { + f.session.generateReply({ userInput: 'transfer' }); + }); + try { + await f.session.start({ agent: root }); + // Both requests initially block root, not the intermediate source activity. + f.session.updateAgent(f.agent); + f.session.updateAgent(f.target); + await f.started.await; + await vi.waitFor(() => expect(f.agent._agentActivity!.currentSpeech).toBeUndefined()); + await f.exiting.await; + expect(f.agent._agentActivity!.schedulingPaused).toBe(false); + f.admission.resolve(); + await expectRejected(f); + const closing = shutdown ? f.session.close() : undefined; + finishTool.resolve(); + f.finishExit.resolve(); + if (closing) { await closing; - expect(targetEntered).not.toHaveBeenCalled(); + expect(f.targetEntered).not.toHaveBeenCalled(); } else { - await vi.waitFor(() => expect(targetEntered).toHaveBeenCalledOnce()); - expect(session.currentAgent).toBe(target); - await session.close(); + await expectHandoff(f); } - if (timing !== 'before handoff' && !shutdown) { - const calls = agent.chatCtx.items.filter( - (item) => item.type === 'function_call' && item.name === 'transfer', - ); - expect(calls).toHaveLength(1); - const outputs = agent.chatCtx.items.filter( - (item) => item.type === 'function_call_output' && item.callId === calls[0]!.callId, - ); - expect(outputs).toHaveLength(1); - expect(outputs[0]).toMatchObject({ isError: true, output: transitionError }); - } - expect(agent._agentActivity).toBeUndefined(); expect(root._agentActivity).toBeUndefined(); - expect(task._agentActivity).toBeUndefined(); - expect(target._agentActivity).toBeUndefined(); } finally { - emitTool.resolve(); - finishSwitch.resolve(); - admission.resolve(); - finishTool.resolve(); - finishExit.resolve(); - await session.close(); - await transition; + enter.mockRestore(); } }, ); From e44b9f150d78c13da33ddd62acb3f35fc6e10394 Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Tue, 15 Sep 2026 11:26:14 +0100 Subject: [PATCH 5/6] fix: finish handoff after onExit errors Log ordinary onExit errors and continue draining so inline tasks return to their parent with the original result. Preserve cancellation propagation, matching Python activity teardown at d3961b903edebf82ca16489e0447212e337d25ce. Addresses AGT-3502 --- agents/src/voice/agent_activity.ts | 7 +- agents/src/voice/agent_task_close.test.ts | 78 ++++++++++++++++++++++- 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 26c4d4a7e..700e81b23 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -5156,7 +5156,12 @@ 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(error, 'error in agent onExit'); + } await this._pauseSchedulingTask([]); // detach after speech tasks are done but before _closeSessionResources diff --git a/agents/src/voice/agent_task_close.test.ts b/agents/src/voice/agent_task_close.test.ts index 836b42efc..aa7b0f90b 100644 --- a/agents/src/voice/agent_task_close.test.ts +++ b/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(); + const failure = new ToolError('task failed'); + const task = AgentTask.create({ + 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(); + } +}); From 70910dbef73208354188d37e90da28a959a74454 Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Tue, 15 Sep 2026 11:29:24 +0100 Subject: [PATCH 6/6] fix: mark onExit error logs as PII Keep hook failure details in a PII-marked structured field so the collector can redact customer content. Addresses AGT-3502 --- agents/src/voice/agent_activity.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 700e81b23..c6482d4fd 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -5160,7 +5160,10 @@ export class AgentActivity implements RecognitionHooks { await this._onExitTask.result; } catch (error) { if (this._onExitTask.cancelled) throw error; - this.logger.error(error, 'error in agent onExit'); + this.logger.error( + { 'lk.pii.error': error instanceof Error ? error.message : String(error) }, + 'error in agent onExit', + ); } await this._pauseSchedulingTask([]);