feat(pipelines-audio): Refine event types and error handling (#2026)

2026-07-13 11:16:26 +08:00
parent 5b47e8af55
commit 88983a1611
9 changed files with 1030 additions and 412 deletions
@@ -14,12 +14,12 @@ export const speechSegmentEvent = defineEventa<TextSegment>('proj-airi:pipelines
 export const speechSpecialEvent = defineEventa<TextSegment>('proj-airi:pipelines:output:speech:special')

 export const speechTtsRequestEvent = defineEventa<TtsRequest>('proj-airi:pipelines:output:speech:tts-request')
-export const speechTtsResultEvent = defineEventa<TtsResult<any>>('proj-airi:pipelines:output:speech:tts-result')
+export const speechTtsResultEvent = defineEventa<TtsResult<unknown>>('proj-airi:pipelines:output:speech:tts-result')

-export const speechPlaybackStartEvent = defineEventa<PlaybackStartEvent<any>>('proj-airi:pipelines:output:speech:playback-start')
-export const speechPlaybackEndEvent = defineEventa<PlaybackEndEvent<any>>('proj-airi:pipelines:output:speech:playback-end')
-export const speechPlaybackInterruptEvent = defineEventa<PlaybackInterruptEvent<any>>('proj-airi:pipelines:output:speech:playback-interrupt')
-export const speechPlaybackRejectEvent = defineEventa<PlaybackRejectEvent<any>>('proj-airi:pipelines:output:speech:playback-reject')
+export const speechPlaybackStartEvent = defineEventa<PlaybackStartEvent<unknown>>('proj-airi:pipelines:output:speech:playback-start')
+export const speechPlaybackEndEvent = defineEventa<PlaybackEndEvent<unknown>>('proj-airi:pipelines:output:speech:playback-end')
+export const speechPlaybackInterruptEvent = defineEventa<PlaybackInterruptEvent<unknown>>('proj-airi:pipelines:output:speech:playback-interrupt')
+export const speechPlaybackRejectEvent = defineEventa<PlaybackRejectEvent<unknown>>('proj-airi:pipelines:output:speech:playback-reject')

 export const speechIntentStartEvent = defineEventa<string>('proj-airi:pipelines:output:speech:intent-start')
 export const speechIntentEndEvent = defineEventa<string>('proj-airi:pipelines:output:speech:intent-end')
@@ -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<string, Set<LlmStreamingControlCallHandler>>
  callManifests: Map<string, LlmStreamingControlCallManifest>
@@ -30,6 +20,144 @@ interface StreamingControlTurnState {
  done: Promise<LlmStreamingControlTurnDone>
 }

+/**
+ * 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<LlmStreamingControlCallContext, 'observer'> | undefined,
+  payload: Parameters<NonNullable<LlmStreamingControlCallContext['observer']>>[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<LlmStreamingControlTurnDone>((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<TPayload extends Record<string, unknown>>(
+  container: Pick<StreamingControlTurnState, 'handlers' | 'callManifests'>,
+  manifest: LlmStreamingControlCallManifest,
+  handler: LlmStreamingControlCallHandler<TPayload>,
+) {
+  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<string, Set<LlmStreamingControlCallHandler>>()
  const callManifests = new Map<string, LlmStreamingControlCallManifest>()
  const turns = new Map<string, StreamingControlTurnState>()
  const signalHandlers = new Set<LlmStreamingControlSignalHandler>()
+
  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<TPayload extends Record<string, unknown> = Record<string, unknown>>(manifest, handler) {
+        return registerHandler<TPayload>(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<LlmStreamingControlCallHandler>()
-      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<LlmStreamingControlTurnDone>((resolve) => {
-        settle = (result) => {
-          if (settled)
-            return
-          settled = true
-          resolve(result)
-        }
-      })
-      const turn = {
-        handlers: new Map<string, Set<LlmStreamingControlCallHandler>>(),
-        callManifests: new Map<string, LlmStreamingControlCallManifest>(),
-        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<StreamingControlTurnState, 'handlers' | 'callManifests'>,
-    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<LlmStreamingControlCallHandler>()
-    registeredHandlers.add(handler)
-    turn.handlers.set(normalizedName, registeredHandlers)
-
-    return () => {
-      registeredHandlers.delete(handler)
-      if (registeredHandlers.size === 0) {
-        turn.handlers.delete(normalizedName)
-        turn.callManifests.delete(normalizedName)
-      }
-    }
-  }
 }
@@ -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' })
@@ -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<string>(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<string, unknown>): NormalizedActPayload {
-  const normalized: NormalizedActPayload = {}
+export function normalizeActPayload(
+  payload: Record<string, unknown>,
+): 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
 }
@@ -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<unknown> {
+  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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<string, () => void>()
+    const play = vi.fn((item: PlaybackItem<unknown>, signal) => new Promise<void>((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))
+  })
+})
@@ -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<TAudio> {
+  item: PlaybackItem<TAudio>
+  controller: AbortController
+  startedAt: number
+}
+
+interface WaitingPlayback<TAudio> {
+  item: PlaybackItem<TAudio>
+  enqueuedAt: number
+}
+
+type Listener<T> = (event: T) => void
+
 export interface PlaybackManagerOptions<TAudio> {
-  play: (item: PlaybackItem<TAudio>, signal: AbortSignal) => Promise<void>
+  play: (
+    item: PlaybackItem<TAudio>,
+    signal: AbortSignal,
+  ) => Promise<void>
+
  maxVoices?: number
  maxVoicesPerOwner?: number
  overflowPolicy?: OverflowPolicy
  ownerOverflowPolicy?: OwnerOverflowPolicy
 }

-export function createPlaybackManager<TAudio>(options: PlaybackManagerOptions<TAudio>) {
+export function createPlaybackManager<TAudio>(
+  options: PlaybackManagerOptions<TAudio>,
+) {
  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<string, {
-    item: PlaybackItem<TAudio>
-    controller: AbortController
-    startedAt: number
-  }>()
-
-  const waiting: Array<{ item: PlaybackItem<TAudio>, enqueuedAt: number }> = []
-
+  const active = new Map<string, ActivePlayback<TAudio>>()
+  const waiting: WaitingPlayback<TAudio>[] = []
  const listeners = {
-    start: [] as Array<(event: PlaybackStartEvent<TAudio>) => void>,
-    end: [] as Array<(event: PlaybackEndEvent<TAudio>) => void>,
-    interrupt: [] as Array<(event: PlaybackInterruptEvent<TAudio>) => void>,
-    reject: [] as Array<(event: PlaybackRejectEvent<TAudio>) => void>,
+    start: new Set<Listener<PlaybackStartEvent<TAudio>>>(),
+    end: new Set<Listener<PlaybackEndEvent<TAudio>>>(),
+    interrupt: new Set<Listener<PlaybackInterruptEvent<TAudio>>>(),
+    reject: new Set<Listener<PlaybackRejectEvent<TAudio>>>(),
  }

-  function onStart(listener: (event: PlaybackStartEvent<TAudio>) => void) {
-    listeners.start.push(listener)
+  function subscribe<T>(bucket: Set<Listener<T>>, listener: Listener<T>) {
+    bucket.add(listener)
+
+    return () => {
+      bucket.delete(listener)
+    }
  }

-  function onEnd(listener: (event: PlaybackEndEvent<TAudio>) => void) {
-    listeners.end.push(listener)
+  function emit<T>(bucket: Set<Listener<T>>, event: T) {
+    for (const listener of [...bucket])
+      listener(event)
  }

-  function onInterrupt(listener: (event: PlaybackInterruptEvent<TAudio>) => 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<TAudio>) => void) {
-    listeners.reject.push(listener)
-  }
-
-  function emitStart(item: PlaybackItem<TAudio>) {
-    const event = { item, startedAt: Date.now() }
-    listeners.start.forEach(listener => listener(event))
-  }
-
-  function emitEnd(item: PlaybackItem<TAudio>) {
-    const event = { item, endedAt: Date.now() }
-    listeners.end.forEach(listener => listener(event))
-  }
-
-  function emitInterrupt(item: PlaybackItem<TAudio>, reason: string) {
-    const event = { item, reason, interruptedAt: Date.now() }
-    listeners.interrupt.forEach(listener => listener(event))
-  }
-
-  function emitReject(item: PlaybackItem<TAudio>, 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<TAudio>, 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<TAudio>):
+    | '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<TAudio>, 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<TAudio>, controller: AbortController }, reason: string) {
-    entry.controller.abort(reason)
-    active.delete(entry.item.id)
-    emitInterrupt(entry.item, reason)
-  }
-
-  function canStart(item: PlaybackItem<TAudio>) {
    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<TAudio>,
+    ) => boolean,
+
+    compare?: (
+      a: ActivePlayback<TAudio>,
+      b: ActivePlayback<TAudio>,
+    ) => boolean,
+  ) {
+    let victim:
+      | ActivePlayback<TAudio>
+      | 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<TAudio>, 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<TAudio>) {
-    const controller = new AbortController()
-    const startedAt = Date.now()
-    active.set(item.id, { item, controller, startedAt })
-    emitStart(item)
+    const entry: ActivePlayback<TAudio>
+      = {
+        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<TAudio>) {
+    const queued: WaitingPlayback<TAudio>
+      = {
+        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<TAudio>, 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<TAudio>, 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<TAudio>) {
-    const { ok, reason } = canStart(item)
-    if (ok) {
+  function interrupt(entry: ActivePlayback<TAudio>, reason: string, options?: { allowStartWaiting?: boolean }) {
+    if (!active.has(entry.item.id)) {
+      return
+    }
+
+    entry.controller.abort(reason)
+    finalize(entry, reason, options)
+  }
+
+  function reject(item: PlaybackItem<TAudio>, reason: string) {
+    emit(
+      listeners.reject,
+      {
+        item,
+        reason,
+        rejectedAt:
+          Date.now(),
+      },
+    )
+  }
+
+  function stealOldest(
+    item: PlaybackItem<TAudio>,
+    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<TAudio>) {
+    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<TAudio>) {
+    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<PlaybackStartEvent<TAudio>>,
+    ) => subscribe(listeners.start, f),
+    onEnd: (
+      f: Listener<PlaybackEndEvent<TAudio>>,
+    ) => subscribe(listeners.end, f),
+    onInterrupt: (
+      f: Listener<PlaybackInterruptEvent<TAudio>>,
+    ) => subscribe(listeners.interrupt, f),
+    onReject: (
+      f: Listener<PlaybackRejectEvent<TAudio>>,
+    ) => subscribe(listeners.reject, f),
  }
 }
@@ -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<string, any> | undefined
-    let afterNext: IteratorResult<string, any> | undefined
+    let next: IteratorResult<string, void> | undefined
+    let afterNext: IteratorResult<string, void> | 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> | 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
@@ -101,7 +101,14 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
  function waitForPlayback(item: PlaybackItem<TAudio>) {
    return new Promise<void>((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<TAudio>(options: SpeechPipelineOptions<TAud
    }
    finally {
      if (intent.canceled) {
-        context.emit(speechPipelineEventMap.onIntentCancel, { intentId: intent.intentId, reason: intent.controller.signal.reason as string | undefined })
+        context.emit(speechPipelineEventMap.onIntentCancel, { intentId: intent.intentId, reason: intent.controller.signal.reason?.toString() })
        if (intent.turnId)
-          context.emit(speechPipelineEventMap.onTurnCancel, { turnId: intent.turnId, reason: intent.controller.signal.reason as string | undefined })
+          context.emit(speechPipelineEventMap.onTurnCancel, { turnId: intent.turnId, reason: intent.controller.signal.reason?.toString() })
      }
      else {
        context.emit(speechPipelineEventMap.onIntentEnd, intent.intentId)
@@ -461,8 +468,10 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
    interrupt,
    stopAll,
    on<K extends SpeechPipelineEventName>(event: K, listener: SpeechPipelineEvents<TAudio>[K]) {
-      return context.on(speechPipelineEventMap[event] as Eventa<any>, (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)
      })
    },
  }
@@ -83,6 +83,7 @@ export interface PlaybackInterruptEvent<TAudio> {
 export interface PlaybackRejectEvent<TAudio> {
  item: PlaybackItem<TAudio>
  reason: string
+  rejectedAt?: number
 }

 export type IntentBehavior = 'queue' | 'interrupt' | 'replace'