{"version":3,"file":"workflow-generator-D67FvbFe.cjs","names":["ReadableStream","VOICE_TEXT_FLUSH"],"sources":["../src/constants.ts","../src/metadata.ts","../src/workflow-generator.ts"],"sourcesContent":["/** Default LiveKit agent name used for explicit dispatch when none is configured. */\nexport const DEFAULT_LIVEKIT_AGENT_NAME = 'mastra-voice';\n","/**\n * Session metadata passed from the Mastra server to the LiveKit agent worker through\n * LiveKit's job dispatch metadata (a plain string, so this is JSON-serialized).\n */\nexport interface LiveKitSessionMetadata {\n  /** Mastra agent to run, by registered key or agent id. */\n  agentId?: string;\n  /** Memory thread id. Defaults to the LiveKit room name when omitted. */\n  threadId?: string;\n  /** Memory resource id (typically the end user id). */\n  resourceId?: string;\n  /** Plain-object entries restored into a RequestContext for agent execution. */\n  requestContext?: Record<string, unknown>;\n}\n\nexport function parseSessionMetadata(raw: string | undefined | null): LiveKitSessionMetadata {\n  if (!raw) return {};\n  try {\n    const parsed = JSON.parse(raw);\n    if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {\n      return parsed as LiveKitSessionMetadata;\n    }\n  } catch {\n    // Dispatch metadata is user-controlled and may not be JSON; treat as absent.\n  }\n  return {};\n}\n\nexport function serializeSessionMetadata(metadata: LiveKitSessionMetadata): string {\n  return JSON.stringify(metadata);\n}\n","import { ReadableStream } from 'node:stream/web';\nimport type { WritableStream } from 'node:stream/web';\nimport type { TracingContext } from '@mastra/core/observability';\nimport type { RequestContext } from '@mastra/core/request-context';\nimport type { Workflow } from '@mastra/core/workflows';\nimport type {\n  VoiceReplyGenerator,\n  VoiceToolCall,\n  VoiceTurnCompleteContext,\n  VoiceTurnCompleteHook,\n  VoiceTurnContext,\n} from './bridge';\nimport { VOICE_TEXT_FLUSH } from './turn-metrics';\nimport type { VoiceReplyChunk } from './turn-metrics';\n\nexport interface WorkflowReplyGeneratorOptions {\n  /** The Mastra workflow that generates each turn's reply. Runs once per turn (no suspend/resume). */\n  workflow: Workflow;\n  /**\n   * Maps a turn into the workflow's `inputData`. Required — input schemas are caller-defined.\n   * A common shape passes the full transcript so the workflow is stateless between turns, e.g.\n   * `ctx => ({ history: chatContextToMessages(ctx.chatCtx) })`.\n   */\n  workflowInput: (ctx: VoiceTurnContext) => unknown | Promise<unknown>;\n  /**\n   * Only stream text from this step (by id). Defaults to every step that writes text to its\n   * `writer`. Set when multiple steps write and only one produces the spoken reply.\n   */\n  replyStep?: string;\n  /**\n   * Fallback when the workflow streams no text via `writer`: derive the spoken reply from the\n   * final run result. Without this, a non-streaming workflow stays silent. Streaming via the\n   * step `writer` is preferred — it gives the caller low time-to-first-token.\n   */\n  resultText?: (result: unknown) => string | undefined | void;\n  /**\n   * Speak a short phrase while a tool call runs. Fires only for tool calls the reply step surfaces\n   * to its `writer` — use {@link pipeAgentReplyToWriter} (or pipe the agent's `fullStream`) so\n   * `tool-call` chunks reach the stream. See {@link MastraVoiceAgentOptions.toolFeedback}.\n   */\n  toolFeedback?: (toolCall: VoiceToolCall) => string | undefined | void;\n  /**\n   * Fired off the audio path after the reply streams, fire-and-forget. Carries the produced reply\n   * text and any tool calls the workflow surfaced. See {@link MastraVoiceAgentOptions.onTurnComplete}.\n   */\n  onTurnComplete?: VoiceTurnCompleteHook;\n}\n\n/**\n * Unwraps the text carried by a `workflow-step-output` chunk's `payload.output`. A step that\n * pipes `agent.stream().textStream` into its `writer` produces raw strings; a step built from an\n * agent (`createStep(agent)`) produces full `text-delta` chunks. Returns the text for both, or\n * `undefined` for any other shape.\n */\nexport function unwrapStepText(output: unknown): string | undefined {\n  if (typeof output === 'string') return output;\n  if (output && typeof output === 'object' && (output as { type?: unknown }).type === 'text-delta') {\n    const text = (output as { payload?: { text?: unknown } }).payload?.text;\n    return typeof text === 'string' ? text : undefined;\n  }\n  return undefined;\n}\n\n/**\n * Unwraps a tool call carried by a `workflow-step-output` chunk's `payload.output`. A step that\n * pipes the agent's `fullStream` (rather than just `textStream`) into its `writer` surfaces\n * `tool-call` chunks; this returns the {@link VoiceToolCall} for those, or `undefined` otherwise.\n */\nexport function unwrapStepToolCall(output: unknown): VoiceToolCall | undefined {\n  if (!output || typeof output !== 'object' || (output as { type?: unknown }).type !== 'tool-call') return undefined;\n  const payload = (output as { payload?: { toolCallId?: unknown; toolName?: unknown; args?: unknown } }).payload;\n  if (payload && typeof payload.toolCallId === 'string' && typeof payload.toolName === 'string') {\n    return { toolCallId: payload.toolCallId, toolName: payload.toolName, args: payload.args };\n  }\n  return undefined;\n}\n\n/** The minimal shape of a Mastra agent stream consumed by {@link pipeAgentReplyToWriter}. */\nexport interface AgentReplyStreamLike {\n  fullStream: AsyncIterable<unknown>;\n}\n\n/**\n * Streams a Mastra agent's reply into a workflow step's `writer` for the LiveKit workflow\n * entrypoint — the recommended way to drive a turn's reply from a step.\n *\n * It forwards the agent's text deltas (so text-to-speech starts before the full reply is ready)\n * AND its `tool-call` chunks (so {@link WorkflowReplyGeneratorOptions.toolFeedback} fires and\n * {@link WorkflowReplyGeneratorOptions.onTurnComplete}'s `result.toolCalls` is populated). This is\n * the difference from piping only `agent.stream().textStream`, which silently drops tool calls.\n * Speech boundaries and tool results are forwarded for flushing and timing. Reasoning is not forwarded, keeping the spoken\n * stream clean. Returns the accumulated reply text for the step to return.\n *\n * Pass the step's `abortSignal` to `agent.stream(...)` so barge-in stops generation promptly.\n *\n * ```ts\n * const generateResponse = createStep({\n *   // ...\n *   execute: async ({ inputData, mastra, writer, abortSignal }) => {\n *     const stream = await mastra.getAgent('callCenter').stream(inputData.turn, { abortSignal });\n *     const reply = await pipeAgentReplyToWriter(stream, writer);\n *     return { reply };\n *   },\n * });\n * ```\n */\nexport function pipeAgentReplyToWriter(\n  agentStream: AgentReplyStreamLike,\n  writer: WritableStream<unknown>,\n): Promise<string> {\n  let text = '';\n  // Re-emit only the chunks the voice path cares about, then pipe through the step writer — which\n  // reuses pipeTo's proven backpressure + close handling rather than driving the writer by hand.\n  const forwarded = new ReadableStream<unknown>({\n    start: async controller => {\n      for await (const chunk of agentStream.fullStream) {\n        const type = (chunk as { type?: unknown })?.type;\n        if (type === 'text-delta') {\n          const delta = (chunk as { payload?: { text?: unknown } }).payload?.text;\n          if (typeof delta === 'string' && delta) {\n            text += delta;\n            controller.enqueue(chunk);\n          }\n        } else if (['tool-call', 'tool-result', 'text-end', 'step-finish'].includes(String(type))) {\n          controller.enqueue(chunk);\n        }\n      }\n      controller.close();\n    },\n  });\n  return forwarded.pipeTo(writer).then(() => text);\n}\n\nfunction toError(error: unknown): Error {\n  if (error instanceof Error) return error;\n  return new Error(error === undefined ? 'workflow run failed' : String(error));\n}\n\n/**\n * A {@link VoiceReplyGenerator} backed by a Mastra workflow. Per turn it starts a fresh run to\n * completion (LiveKit owns the turn boundary, so there is no suspend/resume and no conversation\n * state carried between turns) and streams the text its steps write to their `writer`.\n *\n * A workflow's own stream emits structured step events, not token deltas — text only surfaces\n * when a step pipes it into the injected `writer`, arriving as `workflow-step-output` chunks. The\n * simplest correct reply step calls {@link pipeAgentReplyToWriter}, which forwards both text and\n * tool calls (and passes the step's `abortSignal` through `agent.stream` so barge-in stops\n * generation promptly). Piping only `agent.stream(...).textStream.pipeTo(writer)` works for text\n * but silently drops tool calls, so {@link WorkflowReplyGeneratorOptions.toolFeedback} and\n * {@link WorkflowReplyGeneratorOptions.onTurnComplete}'s `result.toolCalls` stay empty.\n */\nexport function createWorkflowReplyGenerator(options: WorkflowReplyGeneratorOptions): VoiceReplyGenerator {\n  const { workflow, workflowInput, replyStep, resultText, toolFeedback, onTurnComplete } = options;\n  return async ctx => {\n    const inputData = await workflowInput(ctx);\n    const run = await workflow.createRun();\n\n    const streamArgs: { inputData: unknown; tracingContext?: TracingContext; requestContext?: RequestContext } = {\n      inputData,\n    };\n    if (ctx.tracingContext) streamArgs.tracingContext = ctx.tracingContext;\n    // Forward the per-session request context so workflow steps see it, mirroring the agent path.\n    if (ctx.requestContext) streamArgs.requestContext = ctx.requestContext;\n    const output = run.stream(streamArgs);\n\n    let cancelled = false;\n    // Accumulated as the turn streams so the post-turn hook can see what was actually produced.\n    let replyText = '';\n    const toolCalls: VoiceToolCall[] = [];\n\n    // Fire-and-forget after the reply has streamed: off the audio path and not awaited, so it\n    // never delays the next turn. Errors are logged, not thrown. Mirrors createAgentReplyGenerator.\n    const emitTurnComplete = (interrupted: boolean) => {\n      if (!onTurnComplete) return;\n      const completeCtx: VoiceTurnCompleteContext = { ...ctx, result: { text: replyText, toolCalls, interrupted } };\n      Promise.resolve()\n        .then(() => onTurnComplete(completeCtx))\n        .catch(error => {\n          console.warn('@mastra/livekit: onTurnComplete hook threw', error);\n        });\n    };\n\n    return new ReadableStream<VoiceReplyChunk>({\n      start: async controller => {\n        let streamedAny = false;\n        try {\n          for await (const chunk of output.fullStream) {\n            if (cancelled) break;\n            // Step failures don't throw into fullStream; stop reading on a failed finish status and\n            // let the output.result check below throw the run's actual error.\n            if (\n              chunk.type === 'workflow-finish' &&\n              (chunk.payload as { workflowStatus?: unknown } | undefined)?.workflowStatus === 'failed'\n            ) {\n              break;\n            }\n            if (chunk.type !== 'workflow-step-output') continue;\n            const payload = chunk.payload as { output?: unknown; stepName?: unknown };\n            if (replyStep && payload.stepName !== replyStep) continue;\n            const event = payload.output as { type?: string; payload?: { toolCallId?: string } } | undefined;\n            if (event?.type === 'text-end' || event?.type === 'step-finish') {\n              controller.enqueue(VOICE_TEXT_FLUSH);\n              continue;\n            }\n            if (event?.type === 'tool-result' && event.payload?.toolCallId)\n              ctx.metrics?.toolEnd(event.payload.toolCallId);\n            const text = unwrapStepText(payload.output);\n            if (text) {\n              streamedAny = true;\n              replyText += text;\n              controller.enqueue(text);\n              continue;\n            }\n            // A tool call only surfaces when the step pipes the agent's fullStream; when it does,\n            // mirror the agent path — record it and speak any toolFeedback filler.\n            const toolCall = unwrapStepToolCall(payload.output);\n            if (toolCall) {\n              toolCalls.push(toolCall);\n              ctx.metrics?.toolStart({ toolCallId: toolCall.toolCallId, toolName: toolCall.toolName });\n              controller.enqueue(VOICE_TEXT_FLUSH);\n              if (toolFeedback) {\n                const filler = toolFeedback(toolCall);\n                if (filler) {\n                  controller.enqueue(filler.endsWith(' ') ? filler : `${filler} `);\n                  controller.enqueue(VOICE_TEXT_FLUSH);\n                }\n              }\n            }\n          }\n          if (!cancelled) {\n            // A failed run resolves (not rejects) output.result, so check its status explicitly.\n            const result = await output.result;\n            const failed = result as { status?: unknown; error?: unknown } | undefined;\n            if (failed?.status === 'failed') throw toError(failed.error);\n            if (!streamedAny && resultText) {\n              const finalText = resultText(result);\n              if (finalText) {\n                replyText += finalText;\n                controller.enqueue(finalText);\n              }\n            }\n          }\n          if (!cancelled) controller.close();\n          // Success, or a clean barge-in break out of the loop: the turn is done either way.\n          emitTurnComplete(cancelled);\n        } catch (error) {\n          // Barge-in cancels the run; that's not a failure — the turn still completed\n          // (interrupted), so the hook still fires for memory reconciliation.\n          if (cancelled) {\n            emitTurnComplete(true);\n            return;\n          }\n          controller.error(error);\n        }\n      },\n      cancel: () => {\n        cancelled = true;\n        // Barge-in fires this synchronously; swallow any rejection from the run cancellation so a\n        // failed cancel can't surface as an unhandled promise rejection.\n        void Promise.resolve(run.cancel()).catch(error => {\n          console.warn('@mastra/livekit: failed to cancel the workflow run on barge-in', error);\n        });\n      },\n    });\n  };\n}\n"],"mappings":";;;;AACA,MAAa,6BAA6B;;;ACc1C,SAAgB,qBAAqB,KAAwD;CAC3F,IAAI,CAAC,KAAK,OAAO,CAAC;CAClB,IAAI;EACF,MAAM,SAAS,KAAK,MAAM,GAAG;EAC7B,IAAI,UAAU,OAAO,WAAW,YAAY,CAAC,MAAM,QAAQ,MAAM,GAC/D,OAAO;CAEX,QAAQ,CAER;CACA,OAAO,CAAC;AACV;AAEA,SAAgB,yBAAyB,UAA0C;CACjF,OAAO,KAAK,UAAU,QAAQ;AAChC;;;;;;;;;ACwBA,SAAgB,eAAe,QAAqC;CAClE,IAAI,OAAO,WAAW,UAAU,OAAO;CACvC,IAAI,UAAU,OAAO,WAAW,YAAa,OAA8B,SAAS,cAAc;EAChG,MAAM,OAAQ,OAA4C,SAAS;EACnE,OAAO,OAAO,SAAS,WAAW,OAAO,KAAA;CAC3C;AAEF;;;;;;AAOA,SAAgB,mBAAmB,QAA4C;CAC7E,IAAI,CAAC,UAAU,OAAO,WAAW,YAAa,OAA8B,SAAS,aAAa,OAAO,KAAA;CACzG,MAAM,UAAW,OAAsF;CACvG,IAAI,WAAW,OAAO,QAAQ,eAAe,YAAY,OAAO,QAAQ,aAAa,UACnF,OAAO;EAAE,YAAY,QAAQ;EAAY,UAAU,QAAQ;EAAU,MAAM,QAAQ;CAAK;AAG5F;;;;;;;;;;;;;;;;;;;;;;;;;AA+BA,SAAgB,uBACd,aACA,QACiB;CACjB,IAAI,OAAO;CAoBX,OAAO,IAjBeA,WAAAA,eAAwB,EAC5C,OAAO,OAAM,eAAc;EACzB,WAAW,MAAM,SAAS,YAAY,YAAY;GAChD,MAAM,OAAQ,OAA8B;GAC5C,IAAI,SAAS,cAAc;IACzB,MAAM,QAAS,MAA2C,SAAS;IACnE,IAAI,OAAO,UAAU,YAAY,OAAO;KACtC,QAAQ;KACR,WAAW,QAAQ,KAAK;IAC1B;GACF,OAAO,IAAI;IAAC;IAAa;IAAe;IAAY;GAAa,CAAC,CAAC,SAAS,OAAO,IAAI,CAAC,GACtF,WAAW,QAAQ,KAAK;EAE5B;EACA,WAAW,MAAM;CACnB,EACF,CACe,CAAC,CAAC,OAAO,MAAM,CAAC,CAAC,WAAW,IAAI;AACjD;AAEA,SAAS,QAAQ,OAAuB;CACtC,IAAI,iBAAiB,OAAO,OAAO;CACnC,OAAO,IAAI,MAAM,UAAU,KAAA,IAAY,wBAAwB,OAAO,KAAK,CAAC;AAC9E;;;;;;;;;;;;;;AAeA,SAAgB,6BAA6B,SAA6D;CACxG,MAAM,EAAE,UAAU,eAAe,WAAW,YAAY,cAAc,mBAAmB;CACzF,OAAO,OAAM,QAAO;EAClB,MAAM,YAAY,MAAM,cAAc,GAAG;EACzC,MAAM,MAAM,MAAM,SAAS,UAAU;EAErC,MAAM,aAAuG,EAC3G,UACF;EACA,IAAI,IAAI,gBAAgB,WAAW,iBAAiB,IAAI;EAExD,IAAI,IAAI,gBAAgB,WAAW,iBAAiB,IAAI;EACxD,MAAM,SAAS,IAAI,OAAO,UAAU;EAEpC,IAAI,YAAY;EAEhB,IAAI,YAAY;EAChB,MAAM,YAA6B,CAAC;EAIpC,MAAM,oBAAoB,gBAAyB;GACjD,IAAI,CAAC,gBAAgB;GACrB,MAAM,cAAwC;IAAE,GAAG;IAAK,QAAQ;KAAE,MAAM;KAAW;KAAW;IAAY;GAAE;GAC5G,QAAQ,QAAQ,CAAC,CACd,WAAW,eAAe,WAAW,CAAC,CAAC,CACvC,OAAM,UAAS;IACd,QAAQ,KAAK,8CAA8C,KAAK;GAClE,CAAC;EACL;EAEA,OAAO,IAAIA,WAAAA,eAAgC;GACzC,OAAO,OAAM,eAAc;IACzB,IAAI,cAAc;IAClB,IAAI;KACF,WAAW,MAAM,SAAS,OAAO,YAAY;MAC3C,IAAI,WAAW;MAGf,IACE,MAAM,SAAS,qBACd,MAAM,SAAsD,mBAAmB,UAEhF;MAEF,IAAI,MAAM,SAAS,wBAAwB;MAC3C,MAAM,UAAU,MAAM;MACtB,IAAI,aAAa,QAAQ,aAAa,WAAW;MACjD,MAAM,QAAQ,QAAQ;MACtB,IAAI,OAAO,SAAS,cAAc,OAAO,SAAS,eAAe;OAC/D,WAAW,QAAQC,qBAAAA,gBAAgB;OACnC;MACF;MACA,IAAI,OAAO,SAAS,iBAAiB,MAAM,SAAS,YAClD,IAAI,SAAS,QAAQ,MAAM,QAAQ,UAAU;MAC/C,MAAM,OAAO,eAAe,QAAQ,MAAM;MAC1C,IAAI,MAAM;OACR,cAAc;OACd,aAAa;OACb,WAAW,QAAQ,IAAI;OACvB;MACF;MAGA,MAAM,WAAW,mBAAmB,QAAQ,MAAM;MAClD,IAAI,UAAU;OACZ,UAAU,KAAK,QAAQ;OACvB,IAAI,SAAS,UAAU;QAAE,YAAY,SAAS;QAAY,UAAU,SAAS;OAAS,CAAC;OACvF,WAAW,QAAQA,qBAAAA,gBAAgB;OACnC,IAAI,cAAc;QAChB,MAAM,SAAS,aAAa,QAAQ;QACpC,IAAI,QAAQ;SACV,WAAW,QAAQ,OAAO,SAAS,GAAG,IAAI,SAAS,GAAG,OAAO,EAAE;SAC/D,WAAW,QAAQA,qBAAAA,gBAAgB;QACrC;OACF;MACF;KACF;KACA,IAAI,CAAC,WAAW;MAEd,MAAM,SAAS,MAAM,OAAO;MAC5B,MAAM,SAAS;MACf,IAAI,QAAQ,WAAW,UAAU,MAAM,QAAQ,OAAO,KAAK;MAC3D,IAAI,CAAC,eAAe,YAAY;OAC9B,MAAM,YAAY,WAAW,MAAM;OACnC,IAAI,WAAW;QACb,aAAa;QACb,WAAW,QAAQ,SAAS;OAC9B;MACF;KACF;KACA,IAAI,CAAC,WAAW,WAAW,MAAM;KAEjC,iBAAiB,SAAS;IAC5B,SAAS,OAAO;KAGd,IAAI,WAAW;MACb,iBAAiB,IAAI;MACrB;KACF;KACA,WAAW,MAAM,KAAK;IACxB;GACF;GACA,cAAc;IACZ,YAAY;IAGZ,QAAa,QAAQ,IAAI,OAAO,CAAC,CAAC,CAAC,OAAM,UAAS;KAChD,QAAQ,KAAK,kEAAkE,KAAK;IACtF,CAAC;GACH;EACF,CAAC;CACH;AACF"}