From 88983a16119e7a845511d5fe2e9d8ca93c801b67 Mon Sep 17 00:00:00 2001 From: Iro <155815508+Iro96@users.noreply.github.com> Date: Mon, 13 Jul 2026 10:16:26 +0700 Subject: [PATCH] feat(pipelines-audio): Refine event types and error handling (#2026) --- packages/pipelines-audio/src/eventa.ts | 10 +- .../src/llm-streaming-control/controller.ts | 497 +++++++++++------ .../src/llm-streaming-control/index.test.ts | 17 + .../src/llm-streaming-control/payloads.ts | 128 +++-- .../src/managers/playback-manager.test.ts | 203 +++++++ .../src/managers/playback-manager.ts | 506 +++++++++++------- .../src/processors/tts-chunker.ts | 61 ++- .../pipelines-audio/src/speech-pipeline.ts | 19 +- packages/pipelines-audio/src/types.ts | 1 + 9 files changed, 1030 insertions(+), 412 deletions(-) create mode 100644 packages/pipelines-audio/src/managers/playback-manager.test.ts diff --git a/packages/pipelines-audio/src/eventa.ts b/packages/pipelines-audio/src/eventa.ts index 634b7cc6a..3c65f17b8 100644 --- a/packages/pipelines-audio/src/eventa.ts +++ b/packages/pipelines-audio/src/eventa.ts @@ -14,12 +14,12 @@ export const speechSegmentEvent = defineEventa('proj-airi:pipelines export const speechSpecialEvent = defineEventa('proj-airi:pipelines:output:speech:special') export const speechTtsRequestEvent = defineEventa('proj-airi:pipelines:output:speech:tts-request') -export const speechTtsResultEvent = defineEventa>('proj-airi:pipelines:output:speech:tts-result') +export const speechTtsResultEvent = defineEventa>('proj-airi:pipelines:output:speech:tts-result') -export const speechPlaybackStartEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-start') -export const speechPlaybackEndEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-end') -export const speechPlaybackInterruptEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-interrupt') -export const speechPlaybackRejectEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-reject') +export const speechPlaybackStartEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-start') +export const speechPlaybackEndEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-end') +export const speechPlaybackInterruptEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-interrupt') +export const speechPlaybackRejectEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-reject') export const speechIntentStartEvent = defineEventa('proj-airi:pipelines:output:speech:intent-start') export const speechIntentEndEvent = defineEventa('proj-airi:pipelines:output:speech:intent-end') diff --git a/packages/pipelines-audio/src/llm-streaming-control/controller.ts b/packages/pipelines-audio/src/llm-streaming-control/controller.ts index ca9145556..833d772c0 100644 --- a/packages/pipelines-audio/src/llm-streaming-control/controller.ts +++ b/packages/pipelines-audio/src/llm-streaming-control/controller.ts @@ -5,6 +5,7 @@ import type { LlmStreamingControlCallManifest, LlmStreamingControlOptions, LlmStreamingControlSignal, + LlmStreamingControlSignalContext, LlmStreamingControlSignalHandler, LlmStreamingControlTurnDone, } from './types' @@ -12,17 +13,6 @@ import type { import { tokenAct, tokenCall, tokenDelay } from './parsers' import { renderCallManifestPrompt } from './parsers/call' -function parsedParameter(signal: LlmStreamingControlSignal): string | undefined { - switch (signal.type) { - case 'act': - return JSON.stringify(signal.payload) - case 'call': - return signal.payload ? JSON.stringify(signal.payload) : undefined - case 'delay': - return `${signal.seconds}s` - } -} - interface StreamingControlTurnState { handlers: Map> callManifests: Map @@ -30,6 +20,144 @@ interface StreamingControlTurnState { done: Promise } +/** + * Converts parsed signal payload into observer-friendly text. + * + * Use when: + * - Observer logs need a compact human-readable parameter + * + * Notice: + * - Intentionally serializes payloads once + * - Returns undefined for empty CALL payload + */ +function parsedParameter(signal: LlmStreamingControlSignal): string | undefined { + switch (signal.type) { + case 'act': + return JSON.stringify(signal.payload) + + case 'call': + return signal.payload != null + ? JSON.stringify(signal.payload) + : undefined + + case 'delay': + return `${signal.seconds}s` + } +} + +function createTurnId() { + return `turn:${ + globalThis.crypto?.randomUUID?.() + ?? `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}` + }` +} + +/** + * Normalizes manifest values before registration. + * + * Notice: + * - Empty names/prompts are rejected + * - Prevents duplicated trim logic + */ +function normalizeManifest( + manifest: LlmStreamingControlCallManifest, +): LlmStreamingControlCallManifest | undefined { + const name = manifest.name.trim() + const prompt = manifest.prompt.trim() + + if (!name || !prompt) + return + + return { + ...manifest, + name, + prompt, + } +} + +/** + * Emits observer events safely. + * + * Notice: + * - Observer failures must never break dispatch + */ +function emit( + context: Pick | undefined, + payload: Parameters>[0], +) { + try { + context?.observer?.(payload) + } + catch {} +} + +/** + * Creates isolated turn state. + * + * Notice: + * - Promise resolves once only + * - Prevents accidental double completion + */ +function createTurnState(): StreamingControlTurnState { + let settled = false + let settle!: (result: LlmStreamingControlTurnDone) => void + + const done = new Promise((resolve) => { + settle = (result) => { + if (settled) + return + + settled = true + resolve(result) + } + }) + + return { + handlers: new Map(), + callManifests: new Map(), + settle, + done, + } +} + +/** + * Registers handler and keeps cleanup centralized. + * + * Notice: + * - Avoid duplicated Map allocations + * - Removes manifest automatically once empty + */ +function registerHandler>( + container: Pick, + manifest: LlmStreamingControlCallManifest, + handler: LlmStreamingControlCallHandler, +) { + const normalized = normalizeManifest(manifest) + + if (!normalized) + return () => undefined + + let set = container.handlers.get(normalized.name) + + if (!set) { + set = new Set() + container.handlers.set(normalized.name, set) + } + + container.callManifests.set(normalized.name, normalized) + + set.add(handler as LlmStreamingControlCallHandler) + + return () => { + set!.delete(handler) + + if (set!.size === 0) { + container.handlers.delete(normalized.name) + container.callManifests.delete(normalized.name) + } + } +} + /** * Creates a controller over LLM streaming-control tokens. * @@ -38,121 +166,244 @@ interface StreamingControlTurnState { * - A plugin bridge needs to register CALL callbacks against the same controller instance * * Expects: - * - The caller owns the controller lifetime, usually through a Pinia store + * - The caller owns the controller lifetime * * Returns: - * - A controller with `match`, `dispatchWith`, and `on` + * - A controller with `match`, `dispatchWith`, `on` + * + * Notice: + * - Core behavior intentionally preserved + * - Refactored for readability and lower duplication + * - Handler execution order preserved */ -export function createStreamingControlParser(options: LlmStreamingControlOptions = {}): LlmStreamingControl { +export function createStreamingControlParser( + options: LlmStreamingControlOptions = {}, +): LlmStreamingControl { const handlers = new Map>() const callManifests = new Map() const turns = new Map() const signalHandlers = new Set() + const parsers = options.parsers ?? [ tokenAct(), tokenDelay(), tokenCall(), ] + /** + * Finds matching parser. + * + * Notice: + * - Single lookup reused everywhere + */ + function findParser(input: string) { + return parsers.find(parser => parser.match(input)) + } + + /** + * Completes and destroys turn. + * + * Notice: + * - Centralized cleanup path + */ + function finalizeTurn( + turnId: string, + type: LlmStreamingControlTurnDone['type'], + ) { + const turn = turns.get(turnId) + + if (!turn) + return + + turn.settle({ type }) + // always delete after settle to prevent stale turn references + turns.delete(turnId) + } + + function createTurnApi( + turnId: string, + turn: StreamingControlTurnState, + ) { + return { + turnId, + + on = Record>(manifest, handler) { + return registerHandler(turn, manifest, handler) + }, + + renderManifestPrompt() { + return renderCallManifestPrompt( + [...turn.callManifests.values()], + ) + }, + + complete() { + finalizeTurn(turnId, 'completed') + }, + + cancel() { + finalizeTurn(turnId, 'cancelled') + }, + + done: turn.done, + } + } + return { - match(special) { - return parsers.some(parser => parser.match(special)) + match(input) { + return !!findParser(input) }, + async dispatchWith(special, context) { - const parser = parsers.find(item => item.match(special)) + const parser = findParser(special) + if (!parser) { - context?.observer?.({ type: 'rejected', reason: 'no-matching-parser' }) + emit(context, { + type: 'rejected', + reason: 'no-matching-parser', + }) + return false } const parsed = parser.parse(special) + if (!parsed) { - context?.observer?.({ type: 'rejected', reason: 'parse-failed', parserName: parser.name }) + emit(context, { + type: 'rejected', + reason: 'parse-failed', + parserName: parser.name, + }) + return false } - context?.observer?.({ + emit(context, { type: 'parsed', parserName: parser.name, tokenType: parsed.type, - callName: parsed.type === 'call' ? parsed.name : undefined, + callName: + parsed.type === 'call' + ? parsed.name + : undefined, parameter: parsedParameter(parsed), }) - const { observer: _observer, ...dispatchContext } = context ?? {} - const signalContext: LlmStreamingControlCallContext = { ...dispatchContext, createdAt: Date.now() } + const { + observer: _observer, + ...dispatchContext + } = context ?? {} - for (const handler of signalHandlers) { + const signalContext: LlmStreamingControlSignalContext = { + ...dispatchContext, + createdAt: Date.now(), + } + + // snapshot prevents mutation during iteration + for (const handler of [...signalHandlers]) { try { await handler(parsed, signalContext) } catch (error) { - context?.observer?.({ type: 'signal-handler-error', tokenType: parsed.type, error }) - console.warn('[llm-streaming-control] signal handler failed', error) + emit(context, { + type: 'signal-handler-error', + tokenType: parsed.type, + error, + }) + + console.warn( + '[llm-streaming-control] signal handler failed', + error, + ) } } - if (parsed.type !== 'call') { - return true - } - const turnHandlers = dispatchContext.turnId - ? turns.get(dispatchContext.turnId)?.handlers.get(parsed.name) + if (parsed.type !== 'call') + return true + + const turnState = dispatchContext.turnId + ? turns.get(dispatchContext.turnId) : undefined - const globalHandlers = handlers.get(parsed.name) - const registeredHandlers = turnHandlers?.size - ? [...turnHandlers] - : [...(globalHandlers ?? [])] - context?.observer?.({ type: 'call-handler-count', count: registeredHandlers.length }) + const turnHandlers = turnState?.handlers.get(parsed.name) + + const activeHandlers + = turnHandlers?.size && turnState + ? turnHandlers + : handlers.get(parsed.name) + + const registeredHandlers = [ + ...(activeHandlers ?? []), + ] + + emit(context, { + type: 'call-handler-count', + count: registeredHandlers.length, + }) + if (!registeredHandlers.length) { - context?.observer?.({ type: 'call-handler-missing', callName: parsed.name, payload: parsed.payload }) + emit(context, { + type: 'call-handler-missing', + callName: parsed.name, + payload: parsed.payload, + }) + return true } + // preserve sequential execution + // parallel execution would change semantics for (const handler of registeredHandlers) { try { - context?.observer?.({ type: 'call-handler-start', callName: parsed.name }) - await handler(parsed.payload, signalContext) - context?.observer?.({ type: 'call-handler-end', callName: parsed.name }) + emit(context, { + type: 'call-handler-start', + callName: parsed.name, + }) + + await handler( + parsed.payload, + signalContext, + ) + + emit(context, { + type: 'call-handler-end', + callName: parsed.name, + }) } catch (error) { - context?.observer?.({ type: 'call-handler-error', callName: parsed.name, error }) - console.warn('[llm-streaming-control] handler failed', error) + emit(context, { + type: 'call-handler-error', + callName: parsed.name, + error, + }) + + console.warn( + '[llm-streaming-control] handler failed', + error, + ) } } return true }, + on(manifest, handler) { - const normalizedName = manifest.name.trim() - const normalizedPrompt = manifest.prompt.trim() - - if (!normalizedName || !normalizedPrompt) { - return () => undefined - } - - callManifests.set(normalizedName, { - ...manifest, - name: normalizedName, - prompt: normalizedPrompt, - }) - - const registeredHandlers = handlers.get(normalizedName) ?? new Set() - registeredHandlers.add(handler as LlmStreamingControlCallHandler) - handlers.set(normalizedName, registeredHandlers) - - return () => { - registeredHandlers.delete(handler as LlmStreamingControlCallHandler) - - if (registeredHandlers.size === 0) { - handlers.delete(normalizedName) - callManifests.delete(normalizedName) - } - } + return registerHandler( + { + handlers, + callManifests, + }, + manifest, + handler, + ) }, + renderManifestPrompt() { - return renderCallManifestPrompt([...callManifests.values()]) + return renderCallManifestPrompt( + [...callManifests.values()], + ) }, + onSignal(handler) { signalHandlers.add(handler) @@ -160,105 +411,31 @@ export function createStreamingControlParser(options: LlmStreamingControlOptions signalHandlers.delete(handler) } }, - beginTurn(options) { - const turnId = options?.turnId?.trim() || `turn:${Date.now()}:${Math.random().toString(36).slice(2, 8)}` - const existing = turns.get(turnId) - if (existing) { - return { - turnId, - on(manifest, handler) { - return registerTurnHandler(existing, manifest, handler as LlmStreamingControlCallHandler) - }, - renderManifestPrompt() { - return renderCallManifestPrompt([...existing.callManifests.values()]) - }, - complete() { - existing.settle({ type: 'completed' }) - turns.delete(turnId) - }, - cancel() { - existing.settle({ type: 'cancelled' }) - turns.delete(turnId) - }, - done: existing.done, - } - } - let settle!: (result: LlmStreamingControlTurnDone) => void - let settled = false - const done = new Promise((resolve) => { - settle = (result) => { - if (settled) - return - settled = true - resolve(result) - } - }) - const turn = { - handlers: new Map>(), - callManifests: new Map(), - settle, - done, - } + beginTurn(options) { + // crypto UUID avoids collision under concurrency + const turnId + = options?.turnId?.trim() + || createTurnId() + + const existing = turns.get(turnId) + + if (existing) + return createTurnApi(turnId, existing) + + const turn = createTurnState() + turns.set(turnId, turn) - return { - turnId, - on(manifest, handler) { - return registerTurnHandler(turn, manifest, handler as LlmStreamingControlCallHandler) - }, - renderManifestPrompt() { - return renderCallManifestPrompt([...turn.callManifests.values()]) - }, - complete() { - settle({ type: 'completed' }) - turns.delete(turnId) - }, - cancel() { - settle({ type: 'cancelled' }) - turns.delete(turnId) - }, - done, - } + return createTurnApi(turnId, turn) }, + completeTurn(turnId) { - const turn = turns.get(turnId) - if (!turn) - return - turn.settle({ type: 'completed' }) - turns.delete(turnId) + finalizeTurn(turnId, 'completed') }, + cancelTurn(turnId) { - const turn = turns.get(turnId) - if (!turn) - return - turn.settle({ type: 'cancelled' }) - turns.delete(turnId) + finalizeTurn(turnId, 'cancelled') }, } - - function registerTurnHandler( - turn: Pick, - manifest: LlmStreamingControlCallManifest, - handler: LlmStreamingControlCallHandler, - ) { - const normalizedName = manifest.name.trim() - const normalizedPrompt = manifest.prompt.trim() - if (!normalizedName || !normalizedPrompt) { - return () => undefined - } - - turn.callManifests.set(normalizedName, { ...manifest, name: normalizedName, prompt: normalizedPrompt }) - const registeredHandlers = turn.handlers.get(normalizedName) ?? new Set() - registeredHandlers.add(handler) - turn.handlers.set(normalizedName, registeredHandlers) - - return () => { - registeredHandlers.delete(handler) - if (registeredHandlers.size === 0) { - turn.handlers.delete(normalizedName) - turn.callManifests.delete(normalizedName) - } - } - } } diff --git a/packages/pipelines-audio/src/llm-streaming-control/index.test.ts b/packages/pipelines-audio/src/llm-streaming-control/index.test.ts index 25feee7c1..ebb99b135 100644 --- a/packages/pipelines-audio/src/llm-streaming-control/index.test.ts +++ b/packages/pipelines-audio/src/llm-streaming-control/index.test.ts @@ -106,6 +106,23 @@ describe('createStreamingControlParser', () => { turn.complete() }) + it('falls back to global handlers when the turn id is stale', async () => { + const control = createStreamingControlParser() + const globalHandler = vi.fn() + const disposeGlobal = control.on({ + name: 'plugin.action', + prompt: 'Run the global plugin action.', + }, globalHandler) + + await expect(control.dispatchWith('<|CALL ["plugin.action"]|>', { + turnId: 'missing-turn', + })).resolves.toBe(true) + + expect(globalHandler).toHaveBeenCalledTimes(1) + + disposeGlobal() + }) + /** * @example * const turn = control.beginTurn({ turnId: 'turn-1' }) diff --git a/packages/pipelines-audio/src/llm-streaming-control/payloads.ts b/packages/pipelines-audio/src/llm-streaming-control/payloads.ts index 3143eed14..c9c26e8d0 100644 --- a/packages/pipelines-audio/src/llm-streaming-control/payloads.ts +++ b/packages/pipelines-audio/src/llm-streaming-control/payloads.ts @@ -1,3 +1,8 @@ +/** + * Supported emotion values emitted through ACT tokens. + * + * Keep this list synchronized with renderer/runtime support. + */ const emotionValues = [ 'happy', 'sad', @@ -10,10 +15,15 @@ const emotionValues = [ 'neutral', ] as const -export type StreamingControlEmotion = typeof emotionValues[number] +/** Constant-time membership lookup. */ +const emotionSet = new Set(emotionValues) + +export type StreamingControlEmotion = (typeof emotionValues)[number] export interface StreamingControlEmotionPayload { + /** Canonical normalized emotion. */ name: StreamingControlEmotion + /** Emotion strength in range [0–1]. */ intensity: number } @@ -24,77 +34,135 @@ export interface NormalizedActPayload { motion?: string } -function normalizeEmotionName(value: string): StreamingControlEmotion | undefined { +/** + * Converts arbitrary emotion text into canonical emotion. + * + * Examples: + * - "Surprised" → "surprised" + * - " HAPPY " → "happy" + */ +function normalizeEmotionName( + value: string, +): StreamingControlEmotion | undefined { const normalized = value.trim().toLowerCase() - if (emotionValues.includes(normalized as StreamingControlEmotion)) { - return normalized as StreamingControlEmotion - } - return undefined + return emotionSet.has(normalized) + ? (normalized as StreamingControlEmotion) + : undefined } +/** + * Normalizes intensity into [0, 1]. + * + * Invalid values fallback to 1. + * + * // content of things need notice: + * // Accept numeric strings because many streaming payloads arrive serialized. + */ function normalizeIntensity(value: unknown): number { - if (typeof value !== 'number' || Number.isNaN(value)) { + const numeric + = typeof value === 'number' + ? value + : typeof value === 'string' + ? Number(value) + : Number.NaN + + if (!Number.isFinite(numeric)) { return 1 } - return Math.min(1, Math.max(0, value)) + return Math.max(0, Math.min(1, numeric)) } -function normalizeEmotion(value: unknown): StreamingControlEmotionPayload | undefined { +/** + * Converts arbitrary emotion payload into normalized structure. + * + * Supported: + * - "happy" + * - { name: "happy" } + * - { name: "happy", intensity: 0.8 } + */ +function normalizeEmotion( + value: unknown, +): StreamingControlEmotionPayload | undefined { if (typeof value === 'string') { const name = normalizeEmotionName(value) - return name ? { name, intensity: 1 } : undefined + + return name + ? { + name, + intensity: 1, + } + : undefined } - if (!value || typeof value !== 'object' || Array.isArray(value)) { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { return undefined } - if (!('name' in value) || typeof value.name !== 'string') { - return undefined + const normalizedValue = value as { + name?: unknown + intensity?: unknown } - const name = normalizeEmotionName(value.name) + const name + = typeof normalizedValue.name === 'string' + ? normalizeEmotionName(normalizedValue.name) + : undefined + if (!name) { return undefined } return { name, - intensity: normalizeIntensity('intensity' in value ? value.intensity : undefined), + intensity: normalizeIntensity(normalizedValue.intensity), } } +/** + * Trims and validates motion values. + * + * Empty strings become undefined. + */ function normalizeMotion(value: unknown): string | undefined { if (typeof value !== 'string') { return undefined } - const trimmed = value.trim() - return trimmed.length > 0 ? trimmed : undefined + const normalized = value.trim() + + return normalized.length > 0 + ? normalized + : undefined } /** * Normalizes ACT token payloads. * - * Before: - * - `{ emotion: "Surprised", motion: " nod " }` + * Input: + * { + * emotion: "Surprised", + * motion: " nod " + * } * - * After: - * - `{ emotion: { name: "surprised", intensity: 1 }, motion: "nod" }` + * Output: + * { + * emotion: { + * name: "surprised", + * intensity: 1 + * }, + * motion: "nod" + * } */ -export function normalizeActPayload(payload: Record): NormalizedActPayload { - const normalized: NormalizedActPayload = {} +export function normalizeActPayload( + payload: Record, +): NormalizedActPayload { const emotion = normalizeEmotion(payload.emotion) const motion = normalizeMotion(payload.motion) - if (emotion) { - normalized.emotion = emotion + return { + ...(emotion && { emotion }), + ...(motion && { motion }), } - if (motion) { - normalized.motion = motion - } - - return normalized } diff --git a/packages/pipelines-audio/src/managers/playback-manager.test.ts b/packages/pipelines-audio/src/managers/playback-manager.test.ts new file mode 100644 index 000000000..230bbc016 --- /dev/null +++ b/packages/pipelines-audio/src/managers/playback-manager.test.ts @@ -0,0 +1,203 @@ +import type { PlaybackItem } from '../types' + +import { describe, expect, it, vi } from 'vitest' + +import { createPlaybackManager } from './playback-manager' + +function createPlaybackItem(id: string, priority: number, intentId: string, ownerId?: string): PlaybackItem { + return { + id, + streamId: 'stream-1', + intentId, + segmentId: `${id}-segment`, + sequence: 1, + ownerId, + priority, + text: `${id} text`, + special: null, + audio: { id }, + createdAt: Date.now(), + } +} + +describe('createPlaybackManager', () => { + it.each(['stopByIntent', 'stopAll'])( + 'does not restart queued playback when stopping with %s', + (method) => { + const play = vi.fn((_item, signal) => new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }) + })) + const manager = createPlaybackManager({ + maxVoices: 1, + overflowPolicy: 'queue', + play, + }) + + manager.schedule(createPlaybackItem('active', 10, 'intent-1')) + manager.schedule(createPlaybackItem('queued', 5, 'intent-2')) + + if (method === 'stopByIntent') + manager.stopByIntent('intent-1', 'stop') + else + manager.stopAll('stop') + + expect(play).toHaveBeenCalledTimes(1) + }, + ) + + it('rejects lower-priority overflow items with steal-lowest-priority policy', () => { + const play = vi.fn((_item, signal) => new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }) + })) + const rejected: string[] = [] + const manager = createPlaybackManager({ + maxVoices: 1, + overflowPolicy: 'steal-lowest-priority', + play, + }) + + manager.onReject((event) => { + rejected.push(event.item.id) + }) + + manager.schedule(createPlaybackItem('active', 10, 'intent-1')) + manager.schedule(createPlaybackItem('lower', 5, 'intent-2')) + + expect(play).toHaveBeenCalledTimes(1) + expect(rejected).toEqual(['lower']) + }) + + it('rejects equal-priority overflow items with steal-lowest-priority policy', () => { + const play = vi.fn((_item, signal) => new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }) + })) + const rejected: string[] = [] + const manager = createPlaybackManager({ + maxVoices: 1, + overflowPolicy: 'steal-lowest-priority', + play, + }) + + manager.onReject((event) => { + rejected.push(event.item.id) + }) + + manager.schedule(createPlaybackItem('active', 10, 'intent-1')) + manager.schedule(createPlaybackItem('equal', 10, 'intent-2')) + + expect(play).toHaveBeenCalledTimes(1) + expect(rejected).toEqual(['equal']) + }) + + it('rejects an owner-overflow item after stealing a different-owner victim with steal-oldest', () => { + const play = vi.fn((_item, signal) => new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }) + })) + const rejected: string[] = [] + const manager = createPlaybackManager({ + maxVoices: 2, + maxVoicesPerOwner: 1, + overflowPolicy: 'steal-oldest', + ownerOverflowPolicy: 'reject', + play, + }) + + manager.onReject((event) => { + rejected.push(event.item.id) + }) + + manager.schedule(createPlaybackItem('b', 9, 'intent-2', 'owner-y')) + manager.schedule(createPlaybackItem('a', 10, 'intent-1', 'owner-x')) + manager.schedule(createPlaybackItem('a2', 8, 'intent-3', 'owner-x')) + + expect(play).toHaveBeenCalledTimes(2) + expect(rejected).toEqual(['a2']) + }) + + it('rejects an owner-overflow item after stealing a lower-priority victim with steal-lowest-priority', () => { + const play = vi.fn((_item, signal) => new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }) + })) + const rejected: string[] = [] + const manager = createPlaybackManager({ + maxVoices: 2, + maxVoicesPerOwner: 1, + overflowPolicy: 'steal-lowest-priority', + ownerOverflowPolicy: 'reject', + play, + }) + + manager.onReject((event) => { + rejected.push(event.item.id) + }) + + manager.schedule(createPlaybackItem('a', 10, 'intent-1', 'owner-x')) + manager.schedule(createPlaybackItem('b', 1, 'intent-2', 'owner-y')) + manager.schedule(createPlaybackItem('a2', 5, 'intent-3', 'owner-x')) + + expect(play).toHaveBeenCalledTimes(2) + expect(rejected).toEqual(['a2']) + }) + + it('steals the oldest active item for queued owner-overflow when a slot frees up', async () => { + let resolvePlayback: (() => void) | undefined + const play = vi.fn((_item, signal) => new Promise((resolve) => { + resolvePlayback = () => { + signal.aborted ? resolve() : signal.addEventListener('abort', () => resolve(), { once: true }) + resolve() + } + })) + const manager = createPlaybackManager({ + maxVoices: 2, + maxVoicesPerOwner: 1, + overflowPolicy: 'queue', + ownerOverflowPolicy: 'steal-oldest', + play, + }) + + manager.schedule(createPlaybackItem('a', 10, 'intent-1', 'owner-x')) + manager.schedule(createPlaybackItem('b', 9, 'intent-2', 'owner-y')) + manager.schedule(createPlaybackItem('a2', 8, 'intent-3', 'owner-x')) + + expect(play).toHaveBeenCalledTimes(2) + + resolvePlayback?.() + await Promise.resolve() + await Promise.resolve() + + expect(play).toHaveBeenCalledTimes(3) + }) + + it('does not drain the queue while stealing an owner-overflow playback slot', async () => { + const resolveMap = new Map void>() + const play = vi.fn((item: PlaybackItem, signal) => new Promise((resolve) => { + resolveMap.set(item.id, () => resolve()) + + if (!signal.aborted) { + signal.addEventListener('abort', () => resolve(), { once: true }) + } + })) + const manager = createPlaybackManager({ + maxVoices: 2, + maxVoicesPerOwner: 1, + overflowPolicy: 'queue', + ownerOverflowPolicy: 'steal-oldest', + play, + }) + + manager.schedule(createPlaybackItem('a', 10, 'intent-1', 'owner-x')) + manager.schedule(createPlaybackItem('d', 10, 'intent-2', 'owner-y')) + manager.schedule(createPlaybackItem('a2', 9, 'intent-3', 'owner-x')) + manager.schedule(createPlaybackItem('b', 8, 'intent-4', 'owner-y')) + manager.schedule(createPlaybackItem('c', 7, 'intent-5', 'owner-y')) + + expect(play).toHaveBeenCalledTimes(2) + + resolveMap.get('d')?.() + await Promise.resolve() + await Promise.resolve() + + expect(play).toHaveBeenCalledTimes(3) + expect(play).toHaveBeenNthCalledWith(3, expect.objectContaining({ id: 'a2' }), expect.any(AbortSignal)) + }) +}) diff --git a/packages/pipelines-audio/src/managers/playback-manager.ts b/packages/pipelines-audio/src/managers/playback-manager.ts index b9532cab0..0bfed7a22 100644 --- a/packages/pipelines-audio/src/managers/playback-manager.ts +++ b/packages/pipelines-audio/src/managers/playback-manager.ts @@ -9,271 +9,409 @@ import type { import { errorMessageFrom } from '@moeru/std' export type OverflowPolicy = 'queue' | 'reject' | 'steal-oldest' | 'steal-lowest-priority' + export type OwnerOverflowPolicy = 'reject' | 'steal-oldest' +interface ActivePlayback { + item: PlaybackItem + controller: AbortController + startedAt: number +} + +interface WaitingPlayback { + item: PlaybackItem + enqueuedAt: number +} + +type Listener = (event: T) => void + export interface PlaybackManagerOptions { - play: (item: PlaybackItem, signal: AbortSignal) => Promise + play: ( + item: PlaybackItem, + signal: AbortSignal, + ) => Promise + maxVoices?: number maxVoicesPerOwner?: number overflowPolicy?: OverflowPolicy ownerOverflowPolicy?: OwnerOverflowPolicy } -export function createPlaybackManager(options: PlaybackManagerOptions) { +export function createPlaybackManager( + options: PlaybackManagerOptions, +) { const maxVoices = Math.max(1, options.maxVoices ?? 1) - const maxVoicesPerOwner = options.maxVoicesPerOwner + const maxVoicesPerOwner = options.maxVoicesPerOwner != null + ? Math.max(1, options.maxVoicesPerOwner) + : undefined const overflowPolicy = options.overflowPolicy ?? 'queue' const ownerOverflowPolicy = options.ownerOverflowPolicy ?? 'steal-oldest' - - const active = new Map - controller: AbortController - startedAt: number - }>() - - const waiting: Array<{ item: PlaybackItem, enqueuedAt: number }> = [] - + const active = new Map>() + const waiting: WaitingPlayback[] = [] const listeners = { - start: [] as Array<(event: PlaybackStartEvent) => void>, - end: [] as Array<(event: PlaybackEndEvent) => void>, - interrupt: [] as Array<(event: PlaybackInterruptEvent) => void>, - reject: [] as Array<(event: PlaybackRejectEvent) => void>, + start: new Set>>(), + end: new Set>>(), + interrupt: new Set>>(), + reject: new Set>>(), } - function onStart(listener: (event: PlaybackStartEvent) => void) { - listeners.start.push(listener) + function subscribe(bucket: Set>, listener: Listener) { + bucket.add(listener) + + return () => { + bucket.delete(listener) + } } - function onEnd(listener: (event: PlaybackEndEvent) => void) { - listeners.end.push(listener) + function emit(bucket: Set>, event: T) { + for (const listener of [...bucket]) + listener(event) } - function onInterrupt(listener: (event: PlaybackInterruptEvent) => void) { - listeners.interrupt.push(listener) + function exists(id: string) { + return ( + active.has(id) + || waiting.some( + x => x.item.id === id, + ) + ) } - function onReject(listener: (event: PlaybackRejectEvent) => void) { - listeners.reject.push(listener) - } - - function emitStart(item: PlaybackItem) { - const event = { item, startedAt: Date.now() } - listeners.start.forEach(listener => listener(event)) - } - - function emitEnd(item: PlaybackItem) { - const event = { item, endedAt: Date.now() } - listeners.end.forEach(listener => listener(event)) - } - - function emitInterrupt(item: PlaybackItem, reason: string) { - const event = { item, reason, interruptedAt: Date.now() } - listeners.interrupt.forEach(listener => listener(event)) - } - - function emitReject(item: PlaybackItem, reason: string) { - const event = { item, reason } - listeners.reject.forEach(listener => listener(event)) - } - - function countByOwner(ownerId?: string) { + function ownerCount(ownerId?: string) { if (!ownerId) return 0 + let count = 0 - for (const entry of active.values()) { - if (entry.item.ownerId === ownerId) - count += 1 + for (const x of active.values()) { + if (x.item.ownerId === ownerId) { + count++ + } } + return count } - function chooseVictimByPriority() { - let victim: { item: PlaybackItem, controller: AbortController, startedAt: number } | undefined - for (const entry of active.values()) { - if (!victim) - victim = entry - else if (entry.item.priority < victim.item.priority) - victim = entry + function canStart(item: PlaybackItem): + | 'overflow' + | 'owner-overflow' + | undefined { + if ( + maxVoicesPerOwner + && item.ownerId + && ownerCount(item.ownerId) + >= maxVoicesPerOwner + ) { + return 'owner-overflow' } - return victim - } - function chooseVictimOldest(ownerId?: string) { - let victim: { item: PlaybackItem, controller: AbortController, startedAt: number } | undefined - for (const entry of active.values()) { - if (ownerId && entry.item.ownerId !== ownerId) - continue - if (!victim || entry.startedAt < victim.startedAt) - victim = entry - } - return victim - } - - function stopActive(entry: { item: PlaybackItem, controller: AbortController }, reason: string) { - entry.controller.abort(reason) - active.delete(entry.item.id) - emitInterrupt(entry.item, reason) - } - - function canStart(item: PlaybackItem) { if (active.size >= maxVoices) - return { ok: false, reason: 'overflow' as const } - if (maxVoicesPerOwner && item.ownerId) { - if (countByOwner(item.ownerId) >= maxVoicesPerOwner) - return { ok: false, reason: 'owner-overflow' as const } + return 'overflow' + + return undefined + } + + function pickVictim( + predicate?: ( + x: ActivePlayback, + ) => boolean, + + compare?: ( + a: ActivePlayback, + b: ActivePlayback, + ) => boolean, + ) { + let victim: + | ActivePlayback + | undefined + + for (const x of active.values()) { + if (predicate && !predicate(x)) { + continue + } + + if (!victim || (compare && compare(x, victim))) { + victim = x + } + } + + return victim + } + + function finalize(entry: ActivePlayback, interrupted?: string, options?: { allowStartWaiting?: boolean }) { + if (!active.delete(entry.item.id)) { + return + } + + if (interrupted) { + emit( + listeners.interrupt, + { + item: entry.item, + reason: interrupted, + interruptedAt: Date.now(), + }, + ) + } + else { + emit( + listeners.end, + { + item: entry.item, + endedAt: Date.now(), + }, + ) + } + + if (options?.allowStartWaiting !== false) { + tryStartWaiting() } - return { ok: true as const } } function start(item: PlaybackItem) { - const controller = new AbortController() - const startedAt = Date.now() - active.set(item.id, { item, controller, startedAt }) - emitStart(item) + const entry: ActivePlayback + = { + item, + controller: new AbortController(), + startedAt: Date.now(), + } - void options.play(item, controller.signal) + active.set(item.id, entry) + + emit( + listeners.start, + { + item, + startedAt: entry.startedAt, + }, + ) + + void options + .play( + item, + entry.controller.signal, + ) .then(() => { - if (!active.has(item.id)) - return - active.delete(item.id) - emitEnd(item) - void tryStartWaiting() + finalize(entry) }) .catch((err) => { - if (!active.has(item.id)) + if (entry.controller.signal.aborted) { return - active.delete(item.id) - emitInterrupt(item, errorMessageFrom(err) ?? 'playback-error') - void tryStartWaiting() + } + + finalize( + entry, + errorMessageFrom(err) ?? 'playback-error', + ) }) } + function enqueue(item: PlaybackItem) { + const queued: WaitingPlayback + = { + item, + enqueuedAt: Date.now(), + } + + let index = waiting.findIndex(x => x.item.priority < item.priority) + if (index === -1) + index = waiting.length + waiting.splice(index, 0, queued) + } + + function resolvePolicy(blocked: 'overflow' | 'owner-overflow') { + return blocked === 'owner-overflow' + ? ownerOverflowPolicy + : overflowPolicy + } + + function handleBlocked(item: PlaybackItem, blocked: 'overflow' | 'owner-overflow') { + const policy = resolvePolicy(blocked) + switch (policy) { + case 'queue': + enqueue(item) + return + case 'reject': + reject(item, blocked) + return + case 'steal-oldest': + stealOldest(item, blocked) + return + case 'steal-lowest-priority': + stealLowestPriority(item) + } + } + function tryStartWaiting() { - if (waiting.length === 0) - return + let i = 0 - const candidates = waiting - .slice() - .sort((a, b) => (b.item.priority - a.item.priority) || (a.enqueuedAt - b.enqueuedAt)) + while (i < waiting.length && active.size < maxVoices) { + const next = waiting[i] + const blocked = canStart(next.item) + + if (!blocked) { + waiting.splice(i, 1) + start(next.item) - for (const candidate of candidates) { - const { ok, reason } = canStart(candidate.item) - if (!ok) { - if (reason === 'owner-overflow' && ownerOverflowPolicy === 'steal-oldest') { - const victim = chooseVictimOldest(candidate.item.ownerId) - if (victim) - stopActive(victim, 'owner-overflow') - } continue } - const index = waiting.indexOf(candidate) - if (index >= 0) - waiting.splice(index, 1) - - start(candidate.item) - if (active.size >= maxVoices) - break - } - } - - function handleOverflow(item: PlaybackItem, reason: 'overflow' | 'owner-overflow') { - if (reason === 'owner-overflow') { - if (ownerOverflowPolicy === 'reject') { - emitReject(item, 'owner-overflow') - return + const policy = resolvePolicy(blocked) + if (policy === 'queue') { + i++ + continue } - const victim = chooseVictimOldest(item.ownerId) - if (victim) { - stopActive(victim, 'owner-overflow') - waiting.push({ item, enqueuedAt: Date.now() }) - void tryStartWaiting() - return - } - } + waiting.splice(i, 1) - switch (overflowPolicy) { - case 'reject': - emitReject(item, 'overflow') - break - case 'queue': - waiting.push({ item, enqueuedAt: Date.now() }) - break - case 'steal-oldest': { - const victim = chooseVictimOldest() - if (victim) - stopActive(victim, 'steal-oldest') - waiting.push({ item, enqueuedAt: Date.now() }) - void tryStartWaiting() - break - } - case 'steal-lowest-priority': { - const victim = chooseVictimByPriority() - if (victim && victim.item.priority <= item.priority) { - stopActive(victim, 'steal-lowest-priority') - waiting.push({ item, enqueuedAt: Date.now() }) - void tryStartWaiting() - } - else { - emitReject(item, 'lower-priority') - } - break + switch (policy) { + case 'reject': + reject(next.item, blocked) + break + case 'steal-oldest': + stealOldest(next.item, blocked) + break + case 'steal-lowest-priority': + stealLowestPriority(next.item) + break } } } - function schedule(item: PlaybackItem) { - const { ok, reason } = canStart(item) - if (ok) { + function interrupt(entry: ActivePlayback, reason: string, options?: { allowStartWaiting?: boolean }) { + if (!active.has(entry.item.id)) { + return + } + + entry.controller.abort(reason) + finalize(entry, reason, options) + } + + function reject(item: PlaybackItem, reason: string) { + emit( + listeners.reject, + { + item, + reason, + rejectedAt: + Date.now(), + }, + ) + } + + function stealOldest( + item: PlaybackItem, + blocked: + | 'overflow' + | 'owner-overflow', + ) { + const victim = pickVictim( + blocked === 'owner-overflow' + ? x => x.item.ownerId === item.ownerId + : undefined, + + (a, b) => a.startedAt < b.startedAt, + ) + + if (!victim) { + enqueue(item) + return + } + + interrupt(victim, 'overflow', { allowStartWaiting: false }) + + const recheck = canStart(item) + if (!recheck) { start(item) return } - handleOverflow(item, reason) + handleBlocked(item, recheck) } - function stopAll(reason: string) { - for (const entry of active.values()) { - stopActive(entry, reason) + function stealLowestPriority(item: PlaybackItem) { + const victim = pickVictim(undefined, (a, b) => a.item.priority < b.item.priority) + const canSteal = !!victim && victim.item.priority < item.priority + + if (!canSteal) { + reject(item, 'priority-overflow') + return } - waiting.length = 0 + + interrupt(victim, 'priority-overflow', { allowStartWaiting: false }) + + const recheck = canStart(item) + if (!recheck) { + start(item) + return + } + + handleBlocked(item, recheck) } - function stopByIntent(intentId: string, reason: string) { - for (const entry of active.values()) { - if (entry.item.intentId !== intentId) - continue - stopActive(entry, reason) + function schedule(item: PlaybackItem) { + if (exists(item.id)) { + return } - for (let i = waiting.length - 1; i >= 0; i -= 1) { + const blocked = canStart(item) + if (!blocked) { + start(item) + return + } + + handleBlocked(item, blocked) + } + + function stopByIntent(intentId: string, reason = 'stop-by-intent') { + for (let i = waiting.length - 1; i >= 0; i--) { if (waiting[i]?.item.intentId === intentId) waiting.splice(i, 1) } - } - function stopByOwner(ownerId: string, reason: string) { - for (const entry of active.values()) { - if (entry.item.ownerId !== ownerId) - continue - stopActive(entry, reason) + for (const entry of [...active.values()]) { + if (entry.item.intentId === intentId) + interrupt(entry, reason, { allowStartWaiting: false }) } - for (let i = waiting.length - 1; i >= 0; i -= 1) { + tryStartWaiting() + } + + function stopByOwner(ownerId: string, reason = 'stop-by-owner') { + for (let i = waiting.length - 1; i >= 0; i--) { if (waiting[i]?.item.ownerId === ownerId) waiting.splice(i, 1) } + + for (const entry of [...active.values()]) { + if (entry.item.ownerId === ownerId) + interrupt(entry, reason, { allowStartWaiting: false }) + } + + tryStartWaiting() } return { schedule, - stopAll, + stopAll(reason = 'stop-all') { + waiting.length = 0 + + for (const x of [...active.values()]) { + interrupt(x, reason, { allowStartWaiting: false }) + } + }, stopByIntent, stopByOwner, - onStart, - onEnd, - onInterrupt, - onReject, + onStart: ( + f: Listener>, + ) => subscribe(listeners.start, f), + onEnd: ( + f: Listener>, + ) => subscribe(listeners.end, f), + onInterrupt: ( + f: Listener>, + ) => subscribe(listeners.interrupt, f), + onReject: ( + f: Listener>, + ) => subscribe(listeners.reject, f), } } diff --git a/packages/pipelines-audio/src/processors/tts-chunker.ts b/packages/pipelines-audio/src/processors/tts-chunker.ts index 41cce54f6..3f82cef98 100644 --- a/packages/pipelines-audio/src/processors/tts-chunker.ts +++ b/packages/pipelines-audio/src/processors/tts-chunker.ts @@ -80,8 +80,8 @@ export async function* chunkTtsInput( const hard = hardPunctuations.has(value) const soft = softPunctuations.has(value) const kept = keptPunctuations.has(value) - let next: IteratorResult | undefined - let afterNext: IteratorResult | undefined + let next: IteratorResult | undefined + let afterNext: IteratorResult | undefined if (flush || special || hard || soft) { switch (value) { @@ -191,9 +191,6 @@ export async function* chunkTtsInput( current = next } - // TODO: remove later - // eslint-disable-next-line no-console - console.debug('while loop ends, chunk/buffer:', chunk, buffer) if (chunk.length > 0 || buffer.length > 0) { const text = (chunk + buffer).trim() yield { @@ -210,17 +207,15 @@ export async function chunkEmitter( options: TtsInputChunkOptions | undefined, handler: (ttsSegment: TtsChunkItem) => Promise | void, ) { - const sanitizeChunk = (text: string) => { - const cleanedText = text + function sanitizeChunk(text: string) { + return text .replaceAll(TTS_SPECIAL_TOKEN, '') .replaceAll(TTS_FLUSH_INSTRUCTION, '') - - return cleanedText.trim() + .trim() } try { for await (const chunk of chunkTtsInput(reader, options)) { - // TODO: remove later const cleanedText = sanitizeChunk(chunk.text) if (!cleanedText && chunk.reason !== 'special') { continue @@ -228,7 +223,6 @@ export async function chunkEmitter( if (chunk.reason === 'special') { const specialToken = pendingSpecials.shift() - // console.debug("special yield:", specialToken) await handler({ chunk: cleanedText, special: specialToken ?? null, reason: chunk.reason }) } else { @@ -355,24 +349,32 @@ export function processNarrative(text: string, options?: TtsInputChunkOptions): } let result = '' + + if (options?.keepNarrativeText) { + for (let i = 0; i < text.length; i++) { + if (!charsToRemove.has(i)) + result += text[i] + } + + return result + } + + rangesToRemove.sort((a, b) => a[0] - b[0]) + let rangeIndex = 0 + for (let i = 0; i < text.length; i++) { - if (options?.keepNarrativeText) { - if (!charsToRemove.has(i)) { - result += text[i] - } - } - else { - let inRange = false - for (const [start, end] of rangesToRemove) { - if (i >= start && i <= end) { - inRange = true - break - } - } - if (!inRange) { - result += text[i] - } + while ( + rangeIndex < rangesToRemove.length + && i > rangesToRemove[rangeIndex]![1] + ) { + rangeIndex += 1 } + + const activeRange = rangesToRemove[rangeIndex] + if (activeRange && i >= activeRange[0] && i <= activeRange[1]) + continue + + result += text[i] } return result @@ -479,8 +481,8 @@ export function createTtsSegmentStream( })() void (async () => { + const reader = byteStream.getReader() try { - const reader = byteStream.getReader() await chunkEmitter(reader, pendingSpecials, options, async (chunk) => { write({ turnId: meta.turnId, @@ -498,6 +500,9 @@ export function createTtsSegmentStream( catch (err) { error(err) } + finally { + reader.releaseLock() + } })() return stream diff --git a/packages/pipelines-audio/src/speech-pipeline.ts b/packages/pipelines-audio/src/speech-pipeline.ts index 8352967ab..f8f11ee06 100644 --- a/packages/pipelines-audio/src/speech-pipeline.ts +++ b/packages/pipelines-audio/src/speech-pipeline.ts @@ -101,7 +101,14 @@ export function createSpeechPipeline(options: SpeechPipelineOptions) { return new Promise((resolve) => { playbackWaiters.set(item.id, resolve) - options.playback.schedule(item) + try { + options.playback.schedule(item) + } + catch (err) { + playbackWaiters.delete(item.id) + logger.warn('Playback schedule failed:', err) + resolve() + } }) } @@ -296,9 +303,9 @@ export function createSpeechPipeline(options: SpeechPipelineOptions(options: SpeechPipelineOptions(event: K, listener: SpeechPipelineEvents[K]) { - return context.on(speechPipelineEventMap[event] as Eventa, (payload) => { - listener(payload?.body ?? payload) + const typedListener = listener as (payload: unknown) => void + + return context.on(speechPipelineEventMap[event] as Eventa<{ body?: unknown }>, (payload) => { + typedListener(payload?.body ?? payload) }) }, } diff --git a/packages/pipelines-audio/src/types.ts b/packages/pipelines-audio/src/types.ts index 758ff15bd..77e947e3e 100644 --- a/packages/pipelines-audio/src/types.ts +++ b/packages/pipelines-audio/src/types.ts @@ -83,6 +83,7 @@ export interface PlaybackInterruptEvent { export interface PlaybackRejectEvent { item: PlaybackItem reason: string + rejectedAt?: number } export type IntentBehavior = 'queue' | 'interrupt' | 'replace'