diff --git a/.changeset/guard-agent-task-handoff-run-state.md b/.changeset/guard-agent-task-handoff-run-state.md new file mode 100644 index 000000000..6a737d9db --- /dev/null +++ b/.changeset/guard-agent-task-handoff-run-state.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Fix `session.run()` stalling or racing the activity transition when an AgentTask handoff is triggered by a speech that predates the run (e.g. created in `onEnter`): the blocked handoff tasks are now watched by the active run for the duration of the transition. diff --git a/agents/src/voice/agent.ts b/agents/src/voice/agent.ts index 47a2a3dd4..05536663b 100644 --- a/agents/src/voice/agent.ts +++ b/agents/src/voice/agent.ts @@ -684,6 +684,13 @@ export class AgentTask extends Agent extends Agent> = []; + await session._updateActivity(this, { previousActivity: 'pause', newActivity: 'start', @@ -702,16 +711,26 @@ export class AgentTask extends Agent 1) { - runState._unwatchHandle(speechHandle); + if (runState && !runState.done()) { + if (speechHandle && runState._unwatchHandle(speechHandle)) { + suspendedHandles.push(speechHandle); + } + + for (const task of blockedTasks) { + if (runState._unwatchHandle(task)) { + suspendedHandles.push(task); + } + } + + // It is OK to call _markDoneIfNeeded here: _updateActivity has started + // the AgentTask activity, and any onEnter-generated speech is now watched. + // Only call it when something was actually suspended — a run created + // mid-transition has no watched handles yet (its generateReply is still + // deferred behind the activity lock), and marking done on an empty + // handle set would resolve it before it produced any events. + if (suspendedHandles.length > 0) { + runState._markDoneIfNeeded(); } - // it is OK to call _markDoneIfNeeded here, the above _updateActivity will call onEnter - // and newly added handles keep the run alive. - runState._markDoneIfNeeded(); } try { @@ -727,8 +746,10 @@ export class AgentTask extends Agent { + constructor() { + super({ + instructions: 'simple task', + tools: [ + tool({ + name: 'finish', + description: 'Called to complete the task.', + execute: async () => { + this.complete(null); + return 'done'; + }, + }), + ], + }); + } + + async onEnter(): Promise { + // Widen the mid-transition window (old activity paused, new activity + // still starting) so a run completing early is deterministically caught. + await new Promise((r) => setTimeout(r, 500)); + this.session.generateReply({ userInput: 'task_greeting' }); + taskOnEnterCompletedAt = Date.now(); + } +} + +class EnterHandoffAgent extends Agent { + constructor() { + super({ + instructions: 'root agent', + tools: [ + tool({ + name: 'start_task', + description: 'Transitions into SimpleTask.', + execute: async () => { + await new SimpleTask().run(); + return 'task completed'; + }, + }), + ], + }); + } + + async onEnter(): Promise { + // This speech predates any session.run(), so no run watches it. + const handle = this.session.generateReply({ userInput: 'enter_greeting' }); + await handle.waitForPlayout(); + } +} + +function withTimeout(p: Promise, ms: number, label: string): Promise { + return Promise.race([ + p, + new Promise((_, reject) => + setTimeout(() => reject(new Error(`timed out waiting for ${label}`)), ms), + ), + ]); +} + +describe('AgentTask handoff from pre-run speech', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + it('keeps the run alive until the handoff settles', async () => { + const llm = new FakeLLM([ + // onEnter reply -> calls start_task; slow enough that run('hi') starts + // before the tool call lands + { + input: 'enter_greeting', + content: '', + ttft: 1000, + duration: 1000, + toolCalls: [{ name: 'start_task', args: {} }], + }, + // user says "hi" while the handoff is in flight; the only speech the + // first run watches on its own + { input: 'hi', content: 'hello!', ttft: 1000, duration: 2000 }, + // SimpleTask onEnter greeting + { input: 'task_greeting', content: 'hello from task' }, + // user says "bye" -> LLM calls finish + { + input: 'bye', + content: '', + toolCalls: [{ name: 'finish', args: {} }], + }, + // after start_task tool output, LLM responds + { input: 'task completed', content: 'all done' }, + ]); + + const session = new AgentSession({ llm }); + + try { + await session.start({ agent: new EnterHandoffAgent() }); + + await withTimeout(session.run({ userInput: 'hi' }).wait(), 10_000, "run('hi')"); + const runResolvedAt = Date.now(); + expect(session.currentAgent).toBeInstanceOf(SimpleTask); + + // the run must not complete mid-handoff: the new activity's onEnter + // must already have finished when the run resolves + expect(taskOnEnterCompletedAt).toBeGreaterThan(0); + expect(runResolvedAt).toBeGreaterThanOrEqual(taskOnEnterCompletedAt); + + await withTimeout(session.run({ userInput: 'bye' }).wait(), 10_000, "run('bye')"); + expect(session.currentAgent).toBeInstanceOf(EnterHandoffAgent); + } finally { + await session.close().catch(() => {}); + } + }, 30_000); +}); diff --git a/agents/src/voice/testing/run_result.ts b/agents/src/voice/testing/run_result.ts index 21c3a9e11..4033c9436 100644 --- a/agents/src/voice/testing/run_result.ts +++ b/agents/src/voice/testing/run_result.ts @@ -195,8 +195,8 @@ export class RunResult { * @internal * Unwatch a handle. */ - _unwatchHandle(handle: SpeechHandle | Task): void { - this.handles.delete(handle); + _unwatchHandle(handle: SpeechHandle | Task): boolean { + const wasWatched = this.handles.delete(handle); const doneCallback = this.doneCallbacks.get(handle); if (doneCallback) { @@ -207,6 +207,8 @@ export class RunResult { if (isSpeechHandle(handle)) { handle._removeItemAddedCallback(this.itemAddedCallback); } + + return wasWatched; } /** @internal */