{"version":3,"sources":["../../src/voice/agent_task_handoff_eou.test.ts"],"sourcesContent":["// SPDX-FileCopyrightText: 2026 LiveKit, Inc.\n//\n// SPDX-License-Identifier: Apache-2.0\n//\n// Regression test for the end-of-turn stall after AgentTask handoffs\n// (1.4.4 -> 1.4.5 regression; reported in production as multi-minute agent\n// silences, minimal repro contributed by the affected customer).\n//\n// Mechanism: across a task handoff the STT pipeline is reused, so provider\n// transcript timestamps (`SpeechData.endTime`) stay cumulative from the stream\n// start. If the input anchor is re-stamped at handoff time and VAD misses the\n// user's speech, `lastSpeakingTime` falls into the future by the stream's age at\n// handoff, and the end-of-turn timer waits out the difference.\n//\n// Guarded fixes: `inputStartedAt` lives on the reused STT pipeline and\n// `lastSpeakingTime` is clamped to wall clock. Without both,\n// the three steps below stall by ~2s/~3s/~8s instead of committing within the\n// endpointing window.\nimport { AudioFrame } from '@livekit/rtc-node';\nimport { ReadableStream, type ReadableStreamDefaultController } from 'node:stream/web';\nimport { describe, expect, it, vi } from 'vitest';\nimport { asLanguageCode } from '../language.js';\nimport { tool } from '../llm/tool_context.js';\nimport { initializeLogger } from '../log.js';\nimport { type SpeechEvent, SpeechEventType } from '../stt/stt.js';\nimport type { FakeRecognizeStream } from '../stt/testing/fake_stt.js';\nimport { FakeSTT } from '../stt/testing/fake_stt.js';\nimport { VAD, VADStream } from '../vad.js';\nimport { Agent, AgentTask } from './agent.js';\nimport { AgentSession } from './agent_session.js';\nimport { AudioInput } from './io.js';\nimport { FakeLLM } from './testing/fake_llm.js';\n\n// Mock `ws` so adaptive interruption connects to an in-memory socket instead of\n// the real LiveKit inference gateway. It opens cleanly and never returns any\n// inference events (so no interruption is ever detected), but the audio\n// forwarding loop still runs. This keeps the test hermetic and avoids the noisy\n// unhandled rejections a dead network address would produce.\nvi.mock('ws', async () => {\n  const { EventEmitter } = await import('node:events');\n  class AutoOpenWebSocket extends EventEmitter {\n    static readonly CONNECTING = 0;\n    static readonly OPEN = 1;\n    static readonly CLOSING = 2;\n    static readonly CLOSED = 3;\n    readyState = AutoOpenWebSocket.CONNECTING;\n    constructor(_url: string, _opts?: unknown) {\n      super();\n      // Open on the next tick, after connectWebSocket has attached its listeners.\n      setTimeout(() => {\n        this.readyState = AutoOpenWebSocket.OPEN;\n        this.emit('open');\n      }, 0);\n    }\n    send(_data: unknown, cb?: (err?: Error) => void) {\n      cb?.();\n    }\n    ping() {}\n    close() {\n      this.readyState = AutoOpenWebSocket.CLOSED;\n      this.emit('close', 1000, Buffer.alloc(0));\n    }\n    terminate() {\n      this.readyState = AutoOpenWebSocket.CLOSED;\n      this.emit('close', 1006, Buffer.alloc(0));\n    }\n  }\n  return { default: AutoOpenWebSocket, WebSocket: AutoOpenWebSocket };\n});\n\nconst STEP_GAP_MS = 2_000; // silence between turns; grows the stream-age offset\nconst ENDPOINTING_MS = 500; // expected healthy commit delay\nconst STEP_INPUTS = ['step one done', 'step two done', 'step three done'] as const;\n\nconst sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));\n\nasync function waitFor(check: () => boolean, timeoutMs: number, label: string): Promise<void> {\n  const startedAt = Date.now();\n  while (!check()) {\n    if (Date.now() - startedAt > timeoutMs) throw new Error(`timed out waiting for ${label}`);\n    await sleep(20);\n  }\n}\n\nfunction silenceFrame(durationMs: number, sampleRate = 16_000): AudioFrame {\n  const samples = Math.floor((sampleRate * durationMs) / 1000);\n  return new AudioFrame(new Int16Array(samples), sampleRate, 1, samples);\n}\n\n/**\n * A VAD that consumes audio but never detects speech (a missed quiet\n * utterance), forcing `lastSpeakingTime` to fall back to STT timestamps.\n * Subclasses the real VAD/VADStream so `instanceof` checks in AgentActivity hold\n * and the audio tee stays drained (otherwise backpressure would stall the\n * pipeline). It never writes to the output, so `lastSpeakingTime` is never set\n * from VAD — which is exactly the precondition for the buggy STT fallback.\n */\nclass DeafVADStream extends VADStream {\n  constructor(vad: VAD) {\n    super(vad);\n    void this.#drain();\n  }\n\n  // Drain the (tee'd) input so the shared audio pipeline doesn't block, but\n  // never emit a VAD event.\n  async #drain(): Promise<void> {\n    try {\n      while (!this.closed) {\n        const { done } = await this.inputReader.read();\n        if (done) break;\n      }\n    } catch {\n      /* stream detached/closed */\n    }\n  }\n}\n\nclass DeafVAD extends VAD {\n  label = 'deaf-vad';\n\n  constructor() {\n    super({ updateInterval: 1 });\n  }\n\n  stream(): VADStream {\n    return new DeafVADStream(this);\n  }\n}\n\n/** Audio input that lets the test push frames into the session, like a SIP line. */\nclass ScriptedAudioInput extends AudioInput {\n  #controller!: ReadableStreamDefaultController<AudioFrame>;\n\n  constructor() {\n    super();\n    const source = new ReadableStream<AudioFrame>({\n      start: (controller) => {\n        this.#controller = controller;\n      },\n    });\n    this.multiStream.addInputStream(source);\n  }\n\n  push(frame: AudioFrame): void {\n    this.#controller.enqueue(frame);\n  }\n}\n\n/** `SpeechStream.queue` is protected; we inject scripted FINAL_TRANSCRIPTs with a\n * provider-cumulative `endTime` (which `sendFakeTranscript` hard-codes to 0). */\ntype QueueAccess = { queue: { put(ev: SpeechEvent): void } };\n\ndescribe('AgentTask handoff end-of-turn timing', () => {\n  initializeLogger({ pretty: false, level: 'silent' });\n\n  it('commits end-of-turn promptly after each task handoff when VAD misses the user speech', async () => {\n    // Adaptive interruption needs LiveKit credentials to construct (else the\n    // detector is silently disabled and the regression path is not exercised).\n    // The `ws` mock above means the URL is never actually dialed.\n    const savedEnv = {\n      LIVEKIT_API_KEY: process.env.LIVEKIT_API_KEY,\n      LIVEKIT_API_SECRET: process.env.LIVEKIT_API_SECRET,\n      LIVEKIT_INFERENCE_URL: process.env.LIVEKIT_INFERENCE_URL,\n    };\n    process.env.LIVEKIT_API_KEY = 'test-api-key';\n    process.env.LIVEKIT_API_SECRET = 'test-api-secret';\n    process.env.LIVEKIT_INFERENCE_URL = 'ws://127.0.0.1:1';\n\n    // Streaming STT with aligned transcripts. We inject transcript events with\n    // provider-cumulative timestamps directly.\n    const stt = new FakeSTT({\n      label: 'scripted-stt',\n      capabilities: { streaming: true, interimResults: true, alignedTranscript: 'word' },\n    });\n\n    const sttStreams: Array<{ stream: FakeRecognizeStream; openedAt: number }> = [];\n    void (async () => {\n      for await (const stream of stt.streamCh) {\n        sttStreams.push({ stream, openedAt: Date.now() });\n      }\n    })();\n\n    const injectFinalTranscript = (text: string) => {\n      const opened = sttStreams[0]!;\n      const endS = (Date.now() - opened.openedAt) / 1000; // cumulative provider timestamp\n      (opened.stream as unknown as QueueAccess).queue.put({\n        type: SpeechEventType.FINAL_TRANSCRIPT,\n        alternatives: [\n          {\n            text,\n            language: asLanguageCode('en'),\n            startTime: Math.max(0, endS - 0.8),\n            endTime: endS,\n            confidence: 0.95,\n          },\n        ],\n      });\n    };\n\n    // Each user input maps to the current task's completion tool call.\n    const fakeLlm = new FakeLLM(\n      STEP_INPUTS.map((input) => ({ input, toolCalls: [{ name: 'completeStep', args: {} }] })),\n    );\n\n    // ---- Parent agent chaining three AgentTasks (mirrors the production shape) ----\n    const toolExecutedAt: Record<number, number> = {};\n\n    const makeStepTask = (step: number): AgentTask<{ step: number }> => {\n      const task: AgentTask<{ step: number }> = new AgentTask<{ step: number }>({\n        instructions: `You are handling step ${step}. Wait for the caller to finish it.`,\n        tools: [\n          tool({\n            name: 'completeStep',\n            description: `Record that step ${step} is complete.`,\n            execute: async () => {\n              toolExecutedAt[step] = Date.now();\n              task.complete({ step });\n            },\n          }),\n        ],\n      });\n      return task;\n    };\n\n    let allTasksDone = false;\n    const parent = new (class extends Agent {\n      async onEnter(): Promise<void> {\n        for (const step of [1, 2, 3]) {\n          await makeStepTask(step).run(); // task handoff: STT pipeline is reused\n        }\n        allTasksDone = true;\n      }\n    })({ instructions: 'Supervisor: delegate the caller through three steps.' });\n\n    const session = new AgentSession({\n      stt,\n      vad: new DeafVAD(),\n      llm: fakeLlm,\n      turnHandling: {\n        interruption: { mode: 'adaptive' },\n        endpointing: { minDelay: ENDPOINTING_MS, maxDelay: 3_000 },\n      },\n    });\n\n    const audioInput = new ScriptedAudioInput();\n    session.input.audio = audioInput;\n\n    await session.start({ agent: parent });\n\n    // Continuous caller audio, like an open SIP line.\n    const pump = setInterval(() => audioInput.push(silenceFrame(50)), 50);\n\n    const delays: number[] = [];\n    try {\n      await waitFor(() => sttStreams.length >= 1, 5_000, 'STT stream to open');\n\n      for (const [i, input] of STEP_INPUTS.entries()) {\n        const step = i + 1;\n        await sleep(STEP_GAP_MS);\n        const injectedAt = Date.now();\n        injectFinalTranscript(input);\n        await waitFor(() => toolExecutedAt[step] !== undefined, 30_000, `step ${step} tool call`);\n        delays.push(toolExecutedAt[step]! - injectedAt);\n      }\n\n      await waitFor(() => allTasksDone, 5_000, 'parent onEnter to finish');\n    } finally {\n      clearInterval(pump);\n      await session.close().catch(() => {});\n      for (const key of Object.keys(savedEnv) as Array<keyof typeof savedEnv>) {\n        if (savedEnv[key] === undefined) delete process.env[key];\n        else process.env[key] = savedEnv[key];\n      }\n    }\n\n    // The STT pipeline should be reused (a single stream) across all handoffs;\n    // that reuse is what makes `endTime` cumulative across activities.\n    expect(sttStreams.length).toBe(1);\n\n    // every step must commit within the endpointing window — no step may\n    // stall by the stream age at its handoff\n    const [step1, step2, step3] = delays;\n    expect(step1!).toBeLessThan(STEP_GAP_MS / 2);\n    expect(step2!).toBeLessThan(STEP_GAP_MS / 2);\n    expect(step3!).toBeLessThan(STEP_GAP_MS / 2);\n  }, 60_000);\n});\n"],"mappings":";;;;;;;;;;;;;;;;;;;;;;;AAkBA,sBAA2B;AAC3B,iBAAqE;AACrE,oBAAyC;AACzC,sBAA+B;AAC/B,0BAAqB;AACrB,iBAAiC;AACjC,iBAAkD;AAElD,sBAAwB;AACxB,iBAA+B;AAC/B,mBAAiC;AACjC,2BAA6B;AAC7B,gBAA2B;AAC3B,sBAAwB;AAOxB,iBAAG,KAAK,MAAM,YAAY;AACxB,QAAM,EAAE,aAAa,IAAI,MAAM,OAAO,aAAa;AAAA,EACnD,MAAM,0BAA0B,aAAa;AAAA,IAC3C,OAAgB,aAAa;AAAA,IAC7B,OAAgB,OAAO;AAAA,IACvB,OAAgB,UAAU;AAAA,IAC1B,OAAgB,SAAS;AAAA,IACzB,aAAa,kBAAkB;AAAA,IAC/B,YAAY,MAAc,OAAiB;AACzC,YAAM;AAEN,iBAAW,MAAM;AACf,aAAK,aAAa,kBAAkB;AACpC,aAAK,KAAK,MAAM;AAAA,MAClB,GAAG,CAAC;AAAA,IACN;AAAA,IACA,KAAK,OAAgB,IAA4B;AAC/C;AAAA,IACF;AAAA,IACA,OAAO;AAAA,IAAC;AAAA,IACR,QAAQ;AACN,WAAK,aAAa,kBAAkB;AACpC,WAAK,KAAK,SAAS,KAAM,OAAO,MAAM,CAAC,CAAC;AAAA,IAC1C;AAAA,IACA,YAAY;AACV,WAAK,aAAa,kBAAkB;AACpC,WAAK,KAAK,SAAS,MAAM,OAAO,MAAM,CAAC,CAAC;AAAA,IAC1C;AAAA,EACF;AACA,SAAO,EAAE,SAAS,mBAAmB,WAAW,kBAAkB;AACpE,CAAC;AAED,MAAM,cAAc;AACpB,MAAM,iBAAiB;AACvB,MAAM,cAAc,CAAC,iBAAiB,iBAAiB,iBAAiB;AAExE,MAAM,QAAQ,CAAC,OAAe,IAAI,QAAQ,CAAC,MAAM,WAAW,GAAG,EAAE,CAAC;AAElE,eAAe,QAAQ,OAAsB,WAAmB,OAA8B;AAC5F,QAAM,YAAY,KAAK,IAAI;AAC3B,SAAO,CAAC,MAAM,GAAG;AACf,QAAI,KAAK,IAAI,IAAI,YAAY,UAAW,OAAM,IAAI,MAAM,yBAAyB,KAAK,EAAE;AACxF,UAAM,MAAM,EAAE;AAAA,EAChB;AACF;AAEA,SAAS,aAAa,YAAoB,aAAa,MAAoB;AACzE,QAAM,UAAU,KAAK,MAAO,aAAa,aAAc,GAAI;AAC3D,SAAO,IAAI,2BAAW,IAAI,WAAW,OAAO,GAAG,YAAY,GAAG,OAAO;AACvE;AAUA,MAAM,sBAAsB,qBAAU;AAAA,EACpC,YAAY,KAAU;AACpB,UAAM,GAAG;AACT,SAAK,KAAK,OAAO;AAAA,EACnB;AAAA;AAAA;AAAA,EAIA,MAAM,SAAwB;AAC5B,QAAI;AACF,aAAO,CAAC,KAAK,QAAQ;AACnB,cAAM,EAAE,KAAK,IAAI,MAAM,KAAK,YAAY,KAAK;AAC7C,YAAI,KAAM;AAAA,MACZ;AAAA,IACF,QAAQ;AAAA,IAER;AAAA,EACF;AACF;AAEA,MAAM,gBAAgB,eAAI;AAAA,EACxB,QAAQ;AAAA,EAER,cAAc;AACZ,UAAM,EAAE,gBAAgB,EAAE,CAAC;AAAA,EAC7B;AAAA,EAEA,SAAoB;AAClB,WAAO,IAAI,cAAc,IAAI;AAAA,EAC/B;AACF;AAGA,MAAM,2BAA2B,qBAAW;AAAA,EAC1C;AAAA,EAEA,cAAc;AACZ,UAAM;AACN,UAAM,SAAS,IAAI,0BAA2B;AAAA,MAC5C,OAAO,CAAC,eAAe;AACrB,aAAK,cAAc;AAAA,MACrB;AAAA,IACF,CAAC;AACD,SAAK,YAAY,eAAe,MAAM;AAAA,EACxC;AAAA,EAEA,KAAK,OAAyB;AAC5B,SAAK,YAAY,QAAQ,KAAK;AAAA,EAChC;AACF;AAAA,IAMA,wBAAS,wCAAwC,MAAM;AACrD,mCAAiB,EAAE,QAAQ,OAAO,OAAO,SAAS,CAAC;AAEnD,wBAAG,wFAAwF,YAAY;AAIrG,UAAM,WAAW;AAAA,MACf,iBAAiB,QAAQ,IAAI;AAAA,MAC7B,oBAAoB,QAAQ,IAAI;AAAA,MAChC,uBAAuB,QAAQ,IAAI;AAAA,IACrC;AACA,YAAQ,IAAI,kBAAkB;AAC9B,YAAQ,IAAI,qBAAqB;AACjC,YAAQ,IAAI,wBAAwB;AAIpC,UAAM,MAAM,IAAI,wBAAQ;AAAA,MACtB,OAAO;AAAA,MACP,cAAc,EAAE,WAAW,MAAM,gBAAgB,MAAM,mBAAmB,OAAO;AAAA,IACnF,CAAC;AAED,UAAM,aAAuE,CAAC;AAC9E,UAAM,YAAY;AAChB,uBAAiB,UAAU,IAAI,UAAU;AACvC,mBAAW,KAAK,EAAE,QAAQ,UAAU,KAAK,IAAI,EAAE,CAAC;AAAA,MAClD;AAAA,IACF,GAAG;AAEH,UAAM,wBAAwB,CAAC,SAAiB;AAC9C,YAAM,SAAS,WAAW,CAAC;AAC3B,YAAM,QAAQ,KAAK,IAAI,IAAI,OAAO,YAAY;AAC9C,MAAC,OAAO,OAAkC,MAAM,IAAI;AAAA,QAClD,MAAM,2BAAgB;AAAA,QACtB,cAAc;AAAA,UACZ;AAAA,YACE;AAAA,YACA,cAAU,gCAAe,IAAI;AAAA,YAC7B,WAAW,KAAK,IAAI,GAAG,OAAO,GAAG;AAAA,YACjC,SAAS;AAAA,YACT,YAAY;AAAA,UACd;AAAA,QACF;AAAA,MACF,CAAC;AAAA,IACH;AAGA,UAAM,UAAU,IAAI;AAAA,MAClB,YAAY,IAAI,CAAC,WAAW,EAAE,OAAO,WAAW,CAAC,EAAE,MAAM,gBAAgB,MAAM,CAAC,EAAE,CAAC,EAAE,EAAE;AAAA,IACzF;AAGA,UAAM,iBAAyC,CAAC;AAEhD,UAAM,eAAe,CAAC,SAA8C;AAClE,YAAM,OAAoC,IAAI,uBAA4B;AAAA,QACxE,cAAc,yBAAyB,IAAI;AAAA,QAC3C,OAAO;AAAA,cACL,0BAAK;AAAA,YACH,MAAM;AAAA,YACN,aAAa,oBAAoB,IAAI;AAAA,YACrC,SAAS,YAAY;AACnB,6BAAe,IAAI,IAAI,KAAK,IAAI;AAChC,mBAAK,SAAS,EAAE,KAAK,CAAC;AAAA,YACxB;AAAA,UACF,CAAC;AAAA,QACH;AAAA,MACF,CAAC;AACD,aAAO;AAAA,IACT;AAEA,QAAI,eAAe;AACnB,UAAM,SAAS,IAAK,cAAc,mBAAM;AAAA,MACtC,MAAM,UAAyB;AAC7B,mBAAW,QAAQ,CAAC,GAAG,GAAG,CAAC,GAAG;AAC5B,gBAAM,aAAa,IAAI,EAAE,IAAI;AAAA,QAC/B;AACA,uBAAe;AAAA,MACjB;AAAA,IACF,EAAG,EAAE,cAAc,uDAAuD,CAAC;AAE3E,UAAM,UAAU,IAAI,kCAAa;AAAA,MAC/B;AAAA,MACA,KAAK,IAAI,QAAQ;AAAA,MACjB,KAAK;AAAA,MACL,cAAc;AAAA,QACZ,cAAc,EAAE,MAAM,WAAW;AAAA,QACjC,aAAa,EAAE,UAAU,gBAAgB,UAAU,IAAM;AAAA,MAC3D;AAAA,IACF,CAAC;AAED,UAAM,aAAa,IAAI,mBAAmB;AAC1C,YAAQ,MAAM,QAAQ;AAEtB,UAAM,QAAQ,MAAM,EAAE,OAAO,OAAO,CAAC;AAGrC,UAAM,OAAO,YAAY,MAAM,WAAW,KAAK,aAAa,EAAE,CAAC,GAAG,EAAE;AAEpE,UAAM,SAAmB,CAAC;AAC1B,QAAI;AACF,YAAM,QAAQ,MAAM,WAAW,UAAU,GAAG,KAAO,oBAAoB;AAEvE,iBAAW,CAAC,GAAG,KAAK,KAAK,YAAY,QAAQ,GAAG;AAC9C,cAAM,OAAO,IAAI;AACjB,cAAM,MAAM,WAAW;AACvB,cAAM,aAAa,KAAK,IAAI;AAC5B,8BAAsB,KAAK;AAC3B,cAAM,QAAQ,MAAM,eAAe,IAAI,MAAM,QAAW,KAAQ,QAAQ,IAAI,YAAY;AACxF,eAAO,KAAK,eAAe,IAAI,IAAK,UAAU;AAAA,MAChD;AAEA,YAAM,QAAQ,MAAM,cAAc,KAAO,0BAA0B;AAAA,IACrE,UAAE;AACA,oBAAc,IAAI;AAClB,YAAM,QAAQ,MAAM,EAAE,MAAM,MAAM;AAAA,MAAC,CAAC;AACpC,iBAAW,OAAO,OAAO,KAAK,QAAQ,GAAmC;AACvE,YAAI,SAAS,GAAG,MAAM,OAAW,QAAO,QAAQ,IAAI,GAAG;AAAA,YAClD,SAAQ,IAAI,GAAG,IAAI,SAAS,GAAG;AAAA,MACtC;AAAA,IACF;AAIA,8BAAO,WAAW,MAAM,EAAE,KAAK,CAAC;AAIhC,UAAM,CAAC,OAAO,OAAO,KAAK,IAAI;AAC9B,8BAAO,KAAM,EAAE,aAAa,cAAc,CAAC;AAC3C,8BAAO,KAAM,EAAE,aAAa,cAAc,CAAC;AAC3C,8BAAO,KAAM,EAAE,aAAa,cAAc,CAAC;AAAA,EAC7C,GAAG,GAAM;AACX,CAAC;","names":[]}