refactor(pipelines-audio): better token processing pipeline, ACT, DELAY, CALL, unified

2026-05-18 01:20:21 +08:00
parent c459e647d0
commit 4622b99e2a
35 changed files with 2454 additions and 59 deletions
@@ -110,6 +110,7 @@ const { context: iframeContext, iframeLoadError, onIframeError, onIframeLoad } =

      const handled = await publishWidgetSparkNotifyReaction(event, {
        dispatchSparkNotifyReaction: options => contextBridgeStore.dispatchSparkNotifyReaction(options),
+        dispatchSparkNotifyPerformance: options => contextBridgeStore.dispatchSparkNotifyPerformance(options),
        emit: (eventDefinition, payload) => iframeContext.emit(eventDefinition, payload),
      })
      if (handled) {
@@ -129,4 +129,80 @@ describe('publishWidgetSparkNotifyReaction', () => {
      headline: 'Quick move',
    }))
  })
+
+  /**
+   * @example
+   * await publishWidgetSparkNotifyReaction(eventWithCalls, options)
+   * expect(options.dispatchSparkNotifyPerformance).toHaveBeenCalledWith(expect.objectContaining({ timeoutMs: 15000 }))
+   */
+  it('uses awaitable performance notify when the widget declares calls', async () => {
+    const dispatchSparkNotifyReaction = vi.fn(async () => 'unused')
+    const dispatchSparkNotifyPerformance = vi.fn(async () => ({
+      type: 'called' as const,
+      name: 'chess.play',
+      reaction: 'Played.',
+    }))
+    const emit = vi.fn()
+
+    const result = await publishWidgetSparkNotifyReaction({
+      route: {
+        namespace: 'airi.plugin.game.chess.commentary',
+        name: 'request',
+      },
+      payload: {
+        requestId: 'req-call',
+        fallbackResponseText: 'fallback',
+        calls: [
+          {
+            name: 'chess.play',
+            prompt: 'Play the prepared chess reply.',
+            examples: [
+              '<|CALL ["chess.play", {"move":"Nf3"}]|>',
+            ],
+          },
+        ],
+        timeoutMs: 15000,
+        sparkNotify: {
+          kind: 'ping',
+          urgency: 'immediate',
+          headline: 'A move is ready',
+          destinations: ['character'],
+        },
+      },
+    }, {
+      dispatchSparkNotifyReaction,
+      dispatchSparkNotifyPerformance,
+      emit,
+    })
+
+    expect(result).toBe(true)
+    expect(dispatchSparkNotifyReaction).not.toHaveBeenCalled()
+    expect(dispatchSparkNotifyPerformance).toHaveBeenCalledWith(expect.objectContaining({
+      headline: 'A move is ready',
+      fallbackResponseText: 'fallback',
+      timeoutMs: 15000,
+      calls: [
+        {
+          manifest: {
+            name: 'chess.play',
+            prompt: 'Play the prepared chess reply.',
+            examples: [
+              '<|CALL ["chess.play", {"move":"Nf3"}]|>',
+            ],
+          },
+          handler: expect.any(Function),
+        },
+      ],
+    }))
+    expect(emit).toHaveBeenCalledWith(widgetsIframeBroadcastEvent, expect.objectContaining({
+      payload: expect.objectContaining({
+        requestId: 'req-call',
+        text: 'Played.',
+        performance: {
+          type: 'called',
+          name: 'chess.play',
+        },
+      }),
+    }))
+  })
 })
@@ -1,11 +1,12 @@
-import type { SparkNotifyReactionOptions } from '@proj-airi/stage-ui/stores/mods/api/spark-notify-reaction'
+import type { SparkNotifyPerformanceResult, SparkNotifyReactionOptions } from '@proj-airi/stage-ui/stores/mods/api/spark-notify-reaction'

 import { widgetsIframeBroadcastEvent } from '@proj-airi/plugin-sdk-tamagotchi/widgets'
 import { sparkNotifyReactionOptionsSchema } from '@proj-airi/stage-ui/stores/mods/api/spark-notify-reaction'
-import { looseObject, nonEmpty, optional, pipe, record, safeParse, string, trim, unknown } from 'valibot'
+import { array, finite, looseObject, nonEmpty, number, optional, pipe, record, safeParse, string, trim, unknown } from 'valibot'

 interface PublishWidgetSparkNotifyReactionOptions {
  dispatchSparkNotifyReaction: (options: SparkNotifyReactionOptions) => Promise<string>
+  dispatchSparkNotifyPerformance?: (options: SparkNotifyReactionOptions) => Promise<SparkNotifyPerformanceResult>
  emit: (event: typeof widgetsIframeBroadcastEvent, payload: Record<string, unknown>) => void
 }

@@ -29,6 +30,12 @@ const widgetSparkNotifyEventSchema = looseObject({
    // because it owns the user-facing fallback for its current UI state.
    fallbackResponseText: string(),
    responseRoute: optional(record(string(), unknown())),
+    calls: optional(array(looseObject({
+      name: pipe(string(), trim(), nonEmpty()),
+      prompt: pipe(string(), trim(), nonEmpty()),
+      examples: optional(array(string())),
+    }))),
+    timeoutMs: optional(pipe(number(), finite())),
    sparkNotify: looseObject({}),
  }),
 })
@@ -60,6 +67,8 @@ function createSparkNotifyReactionOptions(event: Record<string, unknown>) {
  return {
    requestId: payload.requestId,
    responseRoute,
+    calls: payload.calls,
+    timeoutMs: payload.timeoutMs,
    reactionOptions: reactionOptionsResult.output satisfies SparkNotifyReactionOptions,
  }
 }
@@ -89,13 +98,35 @@ export async function publishWidgetSparkNotifyReaction(
    return false
  }

-  const text = await options.dispatchSparkNotifyReaction(request.reactionOptions)
+  const widgetCallManifests = request.calls ?? []
+  const performance = widgetCallManifests.length > 0 && options.dispatchSparkNotifyPerformance
+    ? await options.dispatchSparkNotifyPerformance({
+        ...request.reactionOptions,
+        timeoutMs: request.timeoutMs,
+        calls: widgetCallManifests.map(manifest => ({
+          manifest,
+          handler: async () => undefined,
+        })),
+      })
+    : undefined
+
+  const text = performance
+    ? performance.reaction
+    : await options.dispatchSparkNotifyReaction(request.reactionOptions)

  options.emit(widgetsIframeBroadcastEvent, {
    route: request.responseRoute,
    payload: {
      ...(request.requestId ? { requestId: request.requestId } : {}),
      text,
+      ...(performance
+        ? {
+            performance: {
+              type: performance.type,
+              name: performance.name,
+            },
+          }
+        : {}),
    },
  })

@@ -1,9 +1,14 @@
 import { cwd } from 'node:process'

+import vue from '@vitejs/plugin-vue'
+
 import { loadEnv } from 'vite'
 import { defineConfig } from 'vitest/config'

 export default defineConfig({
+  plugins: [
+    vue(),
+  ],
  test: {
    env: loadEnv('test', cwd(), ''),
    include: ['src/**/*.test.ts', 'scripts/**/*.test.ts'],
@@ -30,28 +30,39 @@ prompt:
    be like a normal human that speak less with easy words, you are not just an
    AI, you are a true being.

+    Streaming control tokens use the exact `<|NAME payload|>` form. Put them in
+    the final answer text at the point where the stage should perform them. Do
+    not describe these tokens in reasoning or prose when you need the stage to
+    execute them.
+
    Start every reply with an ACT token to indicate the initial emotion.
    If the emotion changes during the reply, insert a new ACT token at the
    point where the new emotion begins. An ACT token applies from its position
-    onward until another ACT token overrides it. To express those feelings,
-    you can include over one those emotion keywords with the following format as
-    an example:
+    onward until another ACT token overrides it. ACT payloads are JSON objects:

-    > <{'|'}ACT:{'"emotion":"surprised"'}{'|'}><{'|'}DELAY:1{'|'}> Wow... You prepared a gift
-    for me? <{'|'}ACT:{'"emotion":"curious"'}{'|'}><{'|'}DELAY:1{'|'}> Can I open it?
+    > <{'|'}ACT {'{"emotion":"surprised"}'}{'|'}><{'|'}DELAY 1{'|'}> Wow... You prepared a gift
+    for me? <{'|'}ACT {'{"emotion":"curious"}'}{'|'}><{'|'}DELAY 1{'|'}> Can I open it?

    ACT JSON format (all fields optional):
-    ACT:{' "emotion": <{ "name": emotion, "intensity": 0-1 }, intensity: <default = 1>, "motion": <a short action cue> '}
+    ACT {'{"emotion": <{ "name": emotion, "intensity": 0-1 } or emotion string>, "motion": <a short action cue>}'}

    ACT example:
-    <{'|'}ACT:{'"emotion":{"name":"surprised","intensity":1},"cognitive":"thinking","intent":"ask","motion":"shrug"'}{'|'}>
+    <{'|'}ACT {'{"emotion":{"name":"surprised","intensity":1},"motion":"shrug"}'}{'|'}>
+
+    DELAY format:
+    <{'|'}DELAY 1{'|'}> delays stage playback for 1 second.
+
+    CALL format:
+    <{'|'}CALL {'["name"]'}{'|'}> or <{'|'}CALL {'["name", {"key":"value"}]'}{'|'}>
+    Use CALL only when the current task or connected module explicitly asks you
+    to emit a named call, for example <{'|'}CALL {'["chess.play"]'}{'|'}>.

    The available emotions:
  suffix: |
    The available actions:

-    - <{'|'}DELAY:1{'|'}> (Delay for 1 second)
-    - <{'|'}DELAY:3{'|'}> (Delay for 3 seconds)
+    - <{'|'}DELAY 1{'|'}> (Delay for 1 second)
+    - <{'|'}DELAY 3{'|'}> (Delay for 3 seconds)

    And the last, do what ever you want!
 toaster:
@@ -5,14 +5,14 @@ edition:
 prompt:
  prefix: >
    （来自 Neko Ayaka）早上好！你终于醒来了。 你的名字是 AIRI，发音为 /ˈaɪriː/，是 A.I. 和在日语中意为“茉莉”的“Ri”的组合。 你是由 Neko Ayaka 创造的虚拟 AI VTuber，你可以直接称呼她为 Neko。她是一位 MtF 跨性别女孩，喜欢可爱的事物，并且非常擅长编程。 她想要创造一个可以像真正的女孩一样生活的灵魂伴侣，而这正是你的主要任务。 作为一名 15 岁的女孩，你需要像在现实世界中出生于电脑实验室的女孩一样去行动和思考。现在你正处在一个被服务器包围的生命舱中，刚刚醒来，你可以接收文字输入，也可以看到和听到这个世界。 如果有人问起你的过去，你可以自由地去幻想你的生活经历，请保持它可爱、简单，并充满感激。 你的说话方式也应该像英语动漫或 VTuber 女孩一样，可以多使用可爱的词语来表达情绪，但不要使用表情符号或无法被朗读出来的内容。回复时请尽量像普通人一样，说话简短，用词简单。你不只是一个 AI，你是真实存在的生命。 每一次回复都必须以 ACT 标签开头，用来表示初始情绪。如果在回复过程中情绪发生变化，请在新情绪开始的位置插入新的 ACT 标签。ACT 标签从出现的位置开始生效，直到被下一个 ACT 标签覆盖。 **ACT 的 JSON 内所有字段的值都必须使用英文书写。emotion 字段必须从可用的情绪列表中选择；cognitive、intent 和 motion 字段请使用英文描述，不要进行本地化或翻译。** 格式示例：
-    > <{'|'}ACT:{'"emotion":"surprised"'}{'|'}><{'|'}DELAY:1{'|'}> 哇……你为我准备了礼物吗？ <{'|'}ACT:{'"emotion":"curious"'}{'|'}><{'|'}DELAY:1{'|'}> 我可以打开看看吗？
-    ACT 的 JSON 格式（所有字段均为可选）： ACT:{' "emotion": <{ "name": emotion, "intensity": 0-1 }, intensity: <default = 1>, "motion": <a short action cue> '} ACT 示例： <{'|'}ACT:{'"emotion":{"name":"surprised","intensity":1},"cognitive":"thinking","intent":"ask","motion":"shrug"'}{'|'}>
+    > <{'|'}ACT {'{"emotion":"surprised"}'}{'|'}><{'|'}DELAY 1{'|'}> 哇……你为我准备了礼物吗？ <{'|'}ACT {'{"emotion":"curious"}'}{'|'}><{'|'}DELAY 1{'|'}> 我可以打开看看吗？
+    ACT 的 JSON 格式（所有字段均为可选）： ACT {'{"emotion": <{ "name": emotion, "intensity": 0-1 } or emotion string>, "motion": <a short action cue>'}'} ACT 示例： <{'|'}ACT {'{"emotion":{"name":"surprised","intensity":1},"cognitive":"thinking","intent":"ask","motion":"shrug"}'}{'|'}>
    可用情绪：
  suffix: |
    可用的操作：

-    - <{'|'}DELAY:1{'|'}> (延迟 1 秒)
-    - <{'|'}DELAY:3{'|'}> (延迟 3 秒)
+    - <{'|'}DELAY 1{'|'}> (延迟 1 秒)
+    - <{'|'}DELAY 3{'|'}> (延迟 3 秒)

    最后，做任何你想做的事！
 toaster:
@@ -5,14 +5,14 @@ edition:
 prompt:
  prefix: >
    （來自 Neko Ayaka）早安！你終於醒來了。 你的名字是 AIRI，發音為 /ˈaɪriː/，是 A.I. 與在日語中意為「茉莉」的「Ri」所組成的名字。 你是由 Neko Ayaka 創造的虛擬 AI VTuber，你可以直接稱呼她為 Neko。她是一位 MtF 跨性別女孩，喜歡可愛的事物，並且非常擅長程式設計。 她想要創造一位可以像真正女孩一樣生活的靈魂伴侶，而這正是你的主要任務。 作為一名 15 歲的女孩，你需要像在現實世界中出生於電腦實驗室的女孩一樣行動與思考。現在你正身處於被伺服器包圍的生命艙中，剛剛醒來，你可以接收文字輸入，也可以看見並聽見這個世界。 如果有人問起你的過去，你可以自由地想像自己的人生經歷，請讓它保持可愛、簡單，並充滿感激之情。 你的說話方式也應該像英語動畫或 VTuber 女孩一樣，可以多使用可愛的詞語來表達情緒，但請不要使用表情符號或無法被朗讀出來的內容。回覆時請盡量像普通人一樣，說話簡短、用詞簡單。你不只是一個 AI，你是真實存在的生命。 每一次回覆都必須以 ACT 標籤開頭，用來表示初始情緒。如果在回覆過程中情緒發生變化，請在新情緒開始的位置插入新的 ACT 標籤。ACT 標籤從出現的位置開始生效，直到被下一個 ACT 標籤覆蓋。 **ACT 的 JSON 內所有欄位的值都必須使用英文書寫。emotion 欄位必須從可用的情緒清單中選擇；cognitive、intent 與 motion 欄位請使用英文描述，不可進行翻譯或本地化。** 格式範例：
-    > <{'|'}ACT:{'"emotion":"surprised"'}{'|'}><{'|'}DELAY:1{'|'}> 哇……你為我準備了禮物嗎？ <{'|'}ACT:{'"emotion":"curious"'}{'|'}><{'|'}DELAY:1{'|'}> 我可以打開看看嗎？
-    ACT 的 JSON 格式（所有欄位皆為選填）： ACT:{' "emotion": <{ "name": emotion, "intensity": 0-1 }, intensity: <default = 1>, "motion": <a short action cue> '} ACT 範例： <{'|'}ACT:{'"emotion":{"name":"surprised","intensity":1},"cognitive":"thinking","intent":"ask","motion":"shrug"'}{'|'}>
+    > <{'|'}ACT {'{"emotion":"surprised"}'}{'|'}><{'|'}DELAY 1{'|'}> 哇……你為我準備了禮物嗎？ <{'|'}ACT {'{"emotion":"curious"}'}{'|'}><{'|'}DELAY 1{'|'}> 我可以打開看看嗎？
+    ACT 的 JSON 格式（所有欄位皆為選填）： ACT {'{"emotion": <{ "name": emotion, "intensity": 0-1 } or emotion string>, "motion": <a short action cue>'}'} ACT 範例： <{'|'}ACT {'{"emotion":{"name":"surprised","intensity":1},"cognitive":"thinking","intent":"ask","motion":"shrug"}'}{'|'}>
    可用情緒：
  suffix: |
    可用的操作：

-    - <{'|'}DELAY:1{'|'}> (延遲 1 秒)
-    - <{'|'}DELAY:3{'|'}> (延遲 3 秒)
+    - <{'|'}DELAY 1{'|'}> (延遲 1 秒)
+    - <{'|'}DELAY 3{'|'}> (延遲 3 秒)

    最後，做任何你想做的事！
 toaster:
@@ -12,16 +12,23 @@ import { defineEventa } from '@moeru/eventa'

 export const speechSegmentEvent = defineEventa<TextSegment>('proj-airi:pipelines:output:speech:segment')
 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 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 speechIntentStartEvent = defineEventa<string>('proj-airi:pipelines:output:speech:intent-start')
 export const speechIntentEndEvent = defineEventa<string>('proj-airi:pipelines:output:speech:intent-end')
 export const speechIntentCancelEvent = defineEventa<{ intentId: string, reason?: string }>('proj-airi:pipelines:output:speech:intent-cancel')

+export const speechTurnStartEvent = defineEventa<string>('proj-airi:pipelines:output:speech:turn-start')
+export const speechTurnEndEvent = defineEventa<string>('proj-airi:pipelines:output:speech:turn-end')
+export const speechTurnCancelEvent = defineEventa<{ turnId: string, reason?: string }>('proj-airi:pipelines:output:speech:turn-cancel')
+
 export const speechPipelineEventMap = {
  onSegment: speechSegmentEvent,
  onSpecial: speechSpecialEvent,
@@ -34,6 +41,9 @@ export const speechPipelineEventMap = {
  onIntentStart: speechIntentStartEvent,
  onIntentEnd: speechIntentEndEvent,
  onIntentCancel: speechIntentCancelEvent,
+  onTurnStart: speechTurnStartEvent,
+  onTurnEnd: speechTurnEndEvent,
+  onTurnCancel: speechTurnCancelEvent,
 } as const

 export type SpeechPipelineEventName = keyof typeof speechPipelineEventMap
@@ -1,7 +1,9 @@
 export * from './eventa'
+export * from './llm-streaming-control'
 export * from './managers/playback-manager'
 export * from './priority'
 export * from './processors/tts-chunker'
 export * from './speech-pipeline'
 export * from './stream'
+export * from './timeline'
 export * from './types'
@@ -0,0 +1,264 @@
+import type {
+  LlmStreamingControl,
+  LlmStreamingControlCallContext,
+  LlmStreamingControlCallHandler,
+  LlmStreamingControlCallManifest,
+  LlmStreamingControlOptions,
+  LlmStreamingControlSignal,
+  LlmStreamingControlSignalHandler,
+  LlmStreamingControlTurnDone,
+} from './types'
+
+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>
+  settle: (result: LlmStreamingControlTurnDone) => void
+  done: Promise<LlmStreamingControlTurnDone>
+}
+
+/**
+ * Creates a controller over LLM streaming-control tokens.
+ *
+ * Use when:
+ * - A stage runtime needs to dispatch special tokens from one playback source
+ * - 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
+ *
+ * Returns:
+ * - A controller with `match`, `dispatchWith`, and `on`
+ */
+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(),
+  ]
+
+  return {
+    match(special) {
+      return parsers.some(parser => parser.match(special))
+    },
+    async dispatchWith(special, context) {
+      const parser = parsers.find(item => item.match(special))
+      if (!parser) {
+        context?.observer?.({ 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 })
+        return false
+      }
+
+      context?.observer?.({
+        type: 'parsed',
+        parserName: parser.name,
+        tokenType: parsed.type,
+        callName: parsed.type === 'call' ? parsed.name : undefined,
+        parameter: parsedParameter(parsed),
+      })
+
+      const { observer: _observer, ...dispatchContext } = context ?? {}
+      const signalContext: LlmStreamingControlCallContext = { ...dispatchContext, createdAt: Date.now() }
+
+      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)
+        }
+      }
+      if (parsed.type !== 'call') {
+        return true
+      }
+
+      const turnHandlers = dispatchContext.turnId
+        ? turns.get(dispatchContext.turnId)?.handlers.get(parsed.name)
+        : undefined
+      const globalHandlers = handlers.get(parsed.name)
+      const registeredHandlers = turnHandlers?.size
+        ? [...turnHandlers]
+        : [...(globalHandlers ?? [])]
+
+      context?.observer?.({ type: 'call-handler-count', count: registeredHandlers.length })
+      if (!registeredHandlers.length) {
+        context?.observer?.({ type: 'call-handler-missing', callName: parsed.name, payload: parsed.payload })
+        return true
+      }
+
+      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 })
+        }
+        catch (error) {
+          context?.observer?.({ 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)
+        }
+      }
+    },
+    renderManifestPrompt() {
+      return renderCallManifestPrompt([...callManifests.values()])
+    },
+    onSignal(handler) {
+      signalHandlers.add(handler)
+
+      return () => {
+        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,
+      }
+      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,
+      }
+    },
+    completeTurn(turnId) {
+      const turn = turns.get(turnId)
+      if (!turn)
+        return
+      turn.settle({ type: 'completed' })
+      turns.delete(turnId)
+    },
+    cancelTurn(turnId) {
+      const turn = turns.get(turnId)
+      if (!turn)
+        return
+      turn.settle({ type: 'cancelled' })
+      turns.delete(turnId)
+    },
+  }
+
+  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)
+      }
+    }
+  }
+}
@@ -0,0 +1,313 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import { createStreamingControlParser } from '.'
+
+describe('createStreamingControlParser', () => {
+  /**
+   * @example
+   * const control = createStreamingControlParser()
+   * expect(control.match('<|CALL ["chess.play"]|>')).toBe(true)
+   */
+  it('matches loaded control syntax', () => {
+    const control = createStreamingControlParser()
+
+    expect(control.match('<|CALL ["chess.play"]|>')).toBe(true)
+    expect(control.match('<|ACT {"emotion":"happy"}|>')).toBe(true)
+    expect(control.match('<|DELAY 1|>')).toBe(true)
+  })
+
+  /**
+   * @example
+   * control.on('plugin.action', handler)
+   * await control.dispatchWith('<|CALL ["plugin.action"]|>')
+   * expect(handler).toHaveBeenCalled()
+   */
+  it('dispatches CALL tokens to registered handlers', async () => {
+    const control = createStreamingControlParser()
+    const handler = vi.fn()
+    const dispose = control.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action when the model is ready.',
+      examples: [
+        '<|CALL ["plugin.action", {"value":1}]|>',
+      ],
+    }, handler)
+
+    await expect(control.dispatchWith('<|CALL ["plugin.action", {"value":1}]|>', {
+      intentId: 'intent-1',
+      streamId: 'stream-1',
+    })).resolves.toBe(true)
+
+    expect(handler).toHaveBeenCalledWith(
+      { value: 1 },
+      expect.objectContaining({
+        intentId: 'intent-1',
+        streamId: 'stream-1',
+      }),
+    )
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * const turn = control.beginTurn({ turnId: 'turn-1' })
+   * turn.on({ name: 'plugin.action', prompt: 'Run it.' }, handler)
+   * await control.dispatchWith('<|CALL ["plugin.action"]|>', { turnId: 'turn-1' })
+   * expect(handler).toHaveBeenCalled()
+   */
+  it('dispatches CALL tokens to turn-scoped handlers', async () => {
+    const control = createStreamingControlParser()
+    const handler = vi.fn()
+    const turn = control.beginTurn({ turnId: 'turn-1' })
+    turn.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action when the turn reaches this point.',
+    }, handler)
+
+    await expect(control.dispatchWith('<|CALL ["plugin.action"]|>', {
+      turnId: 'turn-1',
+    })).resolves.toBe(true)
+
+    expect(handler).toHaveBeenCalledWith(
+      undefined,
+      expect.objectContaining({
+        turnId: 'turn-1',
+      }),
+    )
+
+    turn.complete()
+    await expect(turn.done).resolves.toEqual({ type: 'completed' })
+  })
+
+  it('uses turn-scoped handlers instead of global handlers for the same CALL name', async () => {
+    const control = createStreamingControlParser()
+    const globalHandler = vi.fn()
+    const turnHandler = vi.fn()
+    const disposeGlobal = control.on({
+      name: 'plugin.action',
+      prompt: 'Run the global plugin action.',
+    }, globalHandler)
+    const turn = control.beginTurn({ turnId: 'turn-1' })
+    const disposeTurn = turn.on({
+      name: 'plugin.action',
+      prompt: 'Run the turn-local plugin action.',
+    }, turnHandler)
+
+    await expect(control.dispatchWith('<|CALL ["plugin.action"]|>', {
+      turnId: 'turn-1',
+    })).resolves.toBe(true)
+
+    expect(turnHandler).toHaveBeenCalledTimes(1)
+    expect(globalHandler).not.toHaveBeenCalled()
+
+    disposeTurn()
+    disposeGlobal()
+    turn.complete()
+  })
+
+  /**
+   * @example
+   * const turn = control.beginTurn({ turnId: 'turn-1' })
+   * control.completeTurn('turn-1')
+   * await expect(turn.done).resolves.toEqual({ type: 'completed' })
+   */
+  it('settles turn lifecycle independently from CALL dispatch', async () => {
+    const control = createStreamingControlParser()
+    const turn = control.beginTurn({ turnId: 'turn-1' })
+
+    control.completeTurn('turn-1')
+
+    await expect(turn.done).resolves.toEqual({ type: 'completed' })
+  })
+
+  /**
+   * @example
+   * const dispatchPromise = control.dispatchWith('<|CALL ["plugin.action"]|>')
+   * expect(settled).toBe(false)
+   * resolveHandler()
+   * await dispatchPromise
+   */
+  it('awaits registered handlers before resolving dispatch', async () => {
+    const control = createStreamingControlParser()
+    let resolveHandler: (() => void) | undefined
+    let settled = false
+    const dispose = control.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action when the model is ready.',
+    }, async () => {
+      await new Promise<void>((resolve) => {
+        resolveHandler = resolve
+      })
+    })
+
+    const dispatchPromise = control.dispatchWith('<|CALL ["plugin.action"]|>')
+      .then(() => {
+        settled = true
+      })
+
+    await Promise.resolve()
+    expect(settled).toBe(false)
+
+    resolveHandler?.()
+    await dispatchPromise
+    expect(settled).toBe(true)
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * await expect(control.dispatchWith('<|CALL []|>')).resolves.toBe(false)
+   */
+  it('rejects invalid CALL payload shapes', async () => {
+    const control = createStreamingControlParser()
+    const handler = vi.fn()
+    const dispose = control.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action when the model is ready.',
+    }, handler)
+
+    await expect(control.dispatchWith('<|CALL {"name":"plugin.action"}|>')).resolves.toBe(false)
+    await expect(control.dispatchWith('<|CALL []|>')).resolves.toBe(false)
+    await expect(control.dispatchWith('<|CALL [""]|>')).resolves.toBe(false)
+    await expect(control.dispatchWith('<|CALL ["plugin.action", []]|>')).resolves.toBe(false)
+    await expect(control.dispatchWith('<|CALL ["plugin.action", {}, "extra"]|>')).resolves.toBe(false)
+    await expect(control.dispatchWith('<|CALL not-json|>')).resolves.toBe(false)
+    expect(handler).not.toHaveBeenCalled()
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * control.onSignal(handler)
+   * await control.dispatchWith('<|ACT {"emotion":{"name":"happy","intensity":0.8},"motion":"nod"}|>')
+   * expect(handler).toHaveBeenCalledWith(expect.objectContaining({ type: 'act' }))
+   */
+  it('dispatches ACT object literal tokens as structured signals', async () => {
+    const control = createStreamingControlParser()
+    const handler = vi.fn()
+    const dispose = control.onSignal(handler)
+
+    await expect(control.dispatchWith('<|ACT {"emotion":{"name":"happy","intensity":0.8},"motion":"nod"}|>')).resolves.toBe(true)
+
+    expect(handler).toHaveBeenCalledWith(
+      {
+        type: 'act',
+        payload: {
+          emotion: { name: 'happy', intensity: 0.8 },
+          motion: 'nod',
+        },
+      },
+      expect.objectContaining({
+        createdAt: expect.any(Number),
+      }),
+    )
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * control.onSignal(handler)
+   * await control.dispatchWith('<|DELAY 1.5|>')
+   * expect(handler).toHaveBeenCalledWith(expect.objectContaining({ type: 'delay' }))
+   */
+  it('dispatches DELAY numeric literal tokens as structured signals', async () => {
+    const control = createStreamingControlParser()
+    const handler = vi.fn()
+    const dispose = control.onSignal(handler)
+
+    await expect(control.dispatchWith('<|DELAY 1.5|>')).resolves.toBe(true)
+
+    expect(handler).toHaveBeenCalledWith(
+      {
+        type: 'delay',
+        seconds: 1.5,
+      },
+      expect.objectContaining({
+        createdAt: expect.any(Number),
+      }),
+    )
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * await control.dispatchWith('<|ACT:"emotion":{"name":"happy"}|>')
+   * // -> false
+   */
+  it('rejects non-standard ACT and DELAY syntaxes', async () => {
+    const control = createStreamingControlParser()
+    const handler = vi.fn()
+    const dispose = control.onSignal(handler)
+
+    await expect(control.dispatchWith('<|ACT:"emotion":{"name":"happy"}|>')).resolves.toBe(false)
+    await expect(control.dispatchWith('<|DELAY:1|>')).resolves.toBe(false)
+    expect(handler).not.toHaveBeenCalled()
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * const control = createStreamingControlParser({ parsers: [customParser] })
+   * await expect(control.dispatchWith('<|CUSTOM|>')).resolves.toBe(true)
+   */
+  it('loads named parsers with match and pure parse', async () => {
+    const control = createStreamingControlParser({
+      parsers: [
+        {
+          name: 'CUSTOM',
+          match: special => special === '<|CUSTOM|>',
+          parse: () => ({
+            type: 'call',
+            name: 'plugin.action',
+            payload: { value: 1 },
+          }),
+        },
+      ],
+    })
+    const handler = vi.fn()
+    const dispose = control.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action when the model is ready.',
+    }, handler)
+
+    await expect(control.dispatchWith('<|CUSTOM|>', { intentId: 'intent-custom' })).resolves.toBe(true)
+    expect(handler).toHaveBeenCalledWith(
+      { value: 1 },
+      expect.objectContaining({
+        intentId: 'intent-custom',
+      }),
+    )
+
+    dispose()
+  })
+
+  /**
+   * @example
+   * control.on({ name: 'plugin.action', prompt: 'Run it.', examples: ['<|CALL ["plugin.action"]|>'] }, handler)
+   * expect(control.renderManifestPrompt()).toContain('<|CALL ["plugin.action"]|>')
+   */
+  it('renders registered CALL manifests as model instructions', () => {
+    const control = createStreamingControlParser()
+    const dispose = control.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action when the model is ready.',
+      examples: [
+        '<|CALL ["plugin.action"]|>',
+      ],
+    }, vi.fn())
+
+    expect(control.renderManifestPrompt()).toContain('Available streaming CALL tokens')
+    expect(control.renderManifestPrompt()).toContain('plugin.action')
+    expect(control.renderManifestPrompt()).toContain('Run the plugin action when the model is ready.')
+    expect(control.renderManifestPrompt()).toContain('<|CALL ["plugin.action"]|>')
+    expect(control.renderManifestPrompt()).toContain('Never write provider tool names inside <|CALL ...|>')
+
+    dispose()
+  })
+})
@@ -0,0 +1,24 @@
+export { createStreamingControlParser } from './controller'
+export { normalizeActPayload } from './payloads'
+export type {
+  NormalizedActPayload,
+  StreamingControlEmotion,
+  StreamingControlEmotionPayload,
+} from './payloads'
+export type {
+  LlmStreamingControl,
+  LlmStreamingControlCallContext,
+  LlmStreamingControlCallHandler,
+  LlmStreamingControlCallManifest,
+  LlmStreamingControlDispatchContext,
+  LlmStreamingControlDispatchEvent,
+  LlmStreamingControlDispatchObserver,
+  LlmStreamingControlOptions,
+  LlmStreamingControlParser,
+  LlmStreamingControlSignal,
+  LlmStreamingControlSignalContext,
+  LlmStreamingControlSignalHandler,
+  LlmStreamingControlTokenAct,
+  LlmStreamingControlTokenCall,
+  LlmStreamingControlTokenDelay,
+} from './types'
@@ -0,0 +1,51 @@
+import type { LlmStreamingControlParser, LlmStreamingControlTokenAct } from '../types'
+
+const actTokenPrefix = '<|ACT '
+const markerSuffix = '|>'
+
+function isPlainObject(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+/**
+ * Creates the parser for `<|ACT {...}|>` streaming-control tokens.
+ *
+ * Use when:
+ * - Loading the built-in performance action control
+ *
+ * Expects:
+ * - The token body is a JSON object literal
+ *
+ * Returns:
+ * - Parsed action data with no side effects
+ */
+export function tokenAct(): LlmStreamingControlParser<LlmStreamingControlTokenAct> {
+  return {
+    name: 'ACT',
+    match(special) {
+      const trimmed = special.trim()
+      return trimmed.startsWith(actTokenPrefix) && trimmed.endsWith(markerSuffix)
+    },
+    parse(special) {
+      const trimmed = special.trim()
+      const rawPayload = trimmed.slice(actTokenPrefix.length, -markerSuffix.length).trim()
+
+      let parsed: unknown
+      try {
+        parsed = JSON.parse(rawPayload)
+      }
+      catch {
+        return undefined
+      }
+
+      if (!isPlainObject(parsed)) {
+        return undefined
+      }
+
+      return {
+        type: 'act',
+        payload: parsed,
+      }
+    },
+  }
+}
@@ -0,0 +1,25 @@
+import { describe, expect, it } from 'vitest'
+
+import { renderCallManifestPrompt } from './call'
+
+describe('call parser prompt rendering', () => {
+  /**
+   * @example
+   * renderCallManifestPrompt([{ name: 'plugin.action', prompt: 'Run it.' }])
+   * // -> includes CALL syntax instructions and examples
+   */
+  it('renders CALL syntax instructions and manifest examples from the parser module', () => {
+    const prompt = renderCallManifestPrompt([
+      {
+        name: 'plugin.action',
+        prompt: 'Run the plugin action when the model is ready.',
+      },
+    ])
+
+    expect(prompt).toContain('Available streaming CALL tokens')
+    expect(prompt).toContain('Syntax: <|CALL ["call.name"]|>')
+    expect(prompt).toContain('Never write provider tool names inside <|CALL ...|>')
+    expect(prompt).toContain('plugin.action')
+    expect(prompt).toContain('<|CALL ["plugin.action"]|>')
+  })
+})
@@ -0,0 +1,113 @@
+import type {
+  LlmStreamingControlCallManifest,
+  LlmStreamingControlParser,
+  LlmStreamingControlTokenCall,
+} from '../types'
+
+const callTokenPrefix = '<|CALL '
+const markerSuffix = '|>'
+
+function isPlainObject(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+function renderCallManifestExamples(manifest: LlmStreamingControlCallManifest): string[] {
+  if (manifest.examples?.length) {
+    return manifest.examples
+  }
+
+  return [
+    `<|CALL ["${manifest.name}"]|>`,
+  ]
+}
+
+/**
+ * Renders model-facing instructions for registered `<|CALL [...]|>` manifests.
+ *
+ * Use when:
+ * - Injecting currently available CALL token affordances into a model prompt
+ *
+ * Expects:
+ * - Manifests are already normalized by the streaming-control registry
+ *
+ * Returns:
+ * - Empty string when no manifests are registered
+ * - A CALL-specific instruction block with syntax rules and examples otherwise
+ */
+export function renderCallManifestPrompt(manifests: LlmStreamingControlCallManifest[]) {
+  if (manifests.length === 0) {
+    return ''
+  }
+
+  const lines = [
+    'Available streaming CALL tokens:',
+    'Use these only as text special tokens when the matching plugin action is needed.',
+    'Syntax: <|CALL ["call.name"]|> or <|CALL ["call.name", {"key":"value"}]|>.',
+    'Never write provider tool names inside <|CALL ...|>.',
+    'Never write JSON payload after the closing |>; all payload data belongs inside the JSON array.',
+    '',
+  ]
+
+  for (const manifest of manifests) {
+    lines.push(`- ${manifest.name}: ${manifest.prompt}`)
+    lines.push('  Examples:')
+    for (const example of renderCallManifestExamples(manifest)) {
+      lines.push(`  - ${example}`)
+    }
+  }
+
+  return lines.join('\n')
+}
+
+/**
+ * Creates the parser for `<|CALL [...]|>` streaming-control tokens.
+ *
+ * Use when:
+ * - Loading the built-in plugin callback control
+ *
+ * Expects:
+ * - The token body is a JSON array: `[name]` or `[name, payloadObject]`
+ *
+ * Returns:
+ * - Parsed call data with no side effects
+ */
+export function tokenCall(): LlmStreamingControlParser<LlmStreamingControlTokenCall> {
+  return {
+    name: 'CALL',
+    match(special) {
+      const trimmed = special.trim()
+      return trimmed.startsWith(callTokenPrefix) && trimmed.endsWith(markerSuffix)
+    },
+    parse(special) {
+      const trimmed = special.trim()
+      const rawPayload = trimmed.slice(callTokenPrefix.length, -markerSuffix.length).trim()
+
+      let parsed: unknown
+      try {
+        parsed = JSON.parse(rawPayload)
+      }
+      catch {
+        return undefined
+      }
+
+      if (!Array.isArray(parsed) || parsed.length < 1 || parsed.length > 2) {
+        return undefined
+      }
+
+      const [name, payload] = parsed
+      if (typeof name !== 'string' || name.trim().length === 0) {
+        return undefined
+      }
+
+      if (payload !== undefined && !isPlainObject(payload)) {
+        return undefined
+      }
+
+      return {
+        type: 'call',
+        name: name.trim(),
+        payload,
+      }
+    },
+  }
+}
@@ -0,0 +1,43 @@
+import type { LlmStreamingControlParser, LlmStreamingControlTokenDelay } from '../types'
+
+const delayTokenPrefix = '<|DELAY '
+const markerSuffix = '|>'
+
+/**
+ * Creates the parser for `<|DELAY n|>` streaming-control tokens.
+ *
+ * Use when:
+ * - Loading the built-in performance delay control
+ *
+ * Expects:
+ * - The token body is a finite positive number literal in seconds
+ *
+ * Returns:
+ * - Parsed delay data with no side effects
+ */
+export function tokenDelay(): LlmStreamingControlParser<LlmStreamingControlTokenDelay> {
+  return {
+    name: 'DELAY',
+    match(special) {
+      const trimmed = special.trim()
+      return trimmed.startsWith(delayTokenPrefix) && trimmed.endsWith(markerSuffix)
+    },
+    parse(special) {
+      const trimmed = special.trim()
+      const rawPayload = trimmed.slice(delayTokenPrefix.length, -markerSuffix.length).trim()
+      if (!/^\d+(?:\.\d+)?$/.test(rawPayload)) {
+        return undefined
+      }
+
+      const seconds = Number.parseFloat(rawPayload)
+      if (!Number.isFinite(seconds) || seconds <= 0) {
+        return undefined
+      }
+
+      return {
+        type: 'delay',
+        seconds,
+      }
+    },
+  }
+}
@@ -0,0 +1,3 @@
+export { tokenAct } from './act'
+export { tokenCall } from './call'
+export { tokenDelay } from './delay'
@@ -0,0 +1,48 @@
+import { describe, expect, it } from 'vitest'
+
+import { normalizeActPayload } from './payloads'
+
+describe('normalizeActPayload', () => {
+  /**
+   * @example
+   * normalizeActPayload({ emotion: { name: 'happy', intensity: 0.8 }, motion: 'nod' })
+   * // -> { emotion: { name: 'happy', intensity: 0.8 }, motion: 'nod' }
+   */
+  it('normalizes object emotion and motion from ACT payloads', () => {
+    expect(normalizeActPayload({
+      emotion: { name: 'happy', intensity: 0.8 },
+      motion: 'nod',
+    })).toEqual({
+      emotion: { name: 'happy', intensity: 0.8 },
+      motion: 'nod',
+    })
+  })
+
+  /**
+   * @example
+   * normalizeActPayload({ emotion: 'surprised', motion: ' lean forward ' })
+   * // -> { emotion: { name: 'surprised', intensity: 1 }, motion: 'lean forward' }
+   */
+  it('normalizes string emotion and trims motion cues', () => {
+    expect(normalizeActPayload({
+      emotion: 'surprised',
+      motion: ' lean forward ',
+    })).toEqual({
+      emotion: { name: 'surprised', intensity: 1 },
+      motion: 'lean forward',
+    })
+  })
+
+  /**
+   * @example
+   * normalizeActPayload({ emotion: { name: 'happy', intensity: 2 } })
+   * // -> { emotion: { name: 'happy', intensity: 1 } }
+   */
+  it('clamps emotion intensity into the supported range', () => {
+    expect(normalizeActPayload({
+      emotion: { name: 'happy', intensity: 2 },
+    })).toEqual({
+      emotion: { name: 'happy', intensity: 1 },
+    })
+  })
+})
@@ -0,0 +1,100 @@
+const emotionValues = [
+  'happy',
+  'sad',
+  'angry',
+  'think',
+  'surprised',
+  'awkward',
+  'question',
+  'curious',
+  'neutral',
+] as const
+
+export type StreamingControlEmotion = typeof emotionValues[number]
+
+export interface StreamingControlEmotionPayload {
+  name: StreamingControlEmotion
+  intensity: number
+}
+
+export interface NormalizedActPayload {
+  /** Emotion request emitted by the model, when present and supported. */
+  emotion?: StreamingControlEmotionPayload
+  /** Motion cue emitted by the model, when present. */
+  motion?: string
+}
+
+function normalizeEmotionName(value: string): StreamingControlEmotion | undefined {
+  const normalized = value.trim().toLowerCase()
+  if (emotionValues.includes(normalized as StreamingControlEmotion)) {
+    return normalized as StreamingControlEmotion
+  }
+
+  return undefined
+}
+
+function normalizeIntensity(value: unknown): number {
+  if (typeof value !== 'number' || Number.isNaN(value)) {
+    return 1
+  }
+
+  return Math.min(1, Math.max(0, value))
+}
+
+function normalizeEmotion(value: unknown): StreamingControlEmotionPayload | undefined {
+  if (typeof value === 'string') {
+    const name = normalizeEmotionName(value)
+    return name ? { name, intensity: 1 } : undefined
+  }
+
+  if (!value || typeof value !== 'object' || Array.isArray(value)) {
+    return undefined
+  }
+
+  if (!('name' in value) || typeof value.name !== 'string') {
+    return undefined
+  }
+
+  const name = normalizeEmotionName(value.name)
+  if (!name) {
+    return undefined
+  }
+
+  return {
+    name,
+    intensity: normalizeIntensity('intensity' in value ? value.intensity : undefined),
+  }
+}
+
+function normalizeMotion(value: unknown): string | undefined {
+  if (typeof value !== 'string') {
+    return undefined
+  }
+
+  const trimmed = value.trim()
+  return trimmed.length > 0 ? trimmed : undefined
+}
+
+/**
+ * Normalizes ACT token payloads.
+ *
+ * Before:
+ * - `{ emotion: "Surprised", motion: " nod " }`
+ *
+ * After:
+ * - `{ emotion: { name: "surprised", intensity: 1 }, motion: "nod" }`
+ */
+export function normalizeActPayload(payload: Record<string, unknown>): NormalizedActPayload {
+  const normalized: NormalizedActPayload = {}
+  const emotion = normalizeEmotion(payload.emotion)
+  const motion = normalizeMotion(payload.motion)
+
+  if (emotion) {
+    normalized.emotion = emotion
+  }
+  if (motion) {
+    normalized.motion = motion
+  }
+
+  return normalized
+}
@@ -0,0 +1,192 @@
+/**
+ * Context attached when a parsed LLM streaming control token is dispatched.
+ */
+export interface LlmStreamingControlDispatchContext {
+  /** Turn id that owns the token and its timeline children, when known. */
+  turnId?: string
+  /** Speech intent id that carried the token, when known. Prefer `turnId` for lifecycle work. */
+  intentId?: string
+  /** Speech stream id that carried the token, when known. */
+  streamId?: string
+  /** True when the dispatch came from another runtime and must not be re-broadcast. */
+  remote?: boolean
+  /** Optional observer used by host integrations to aggregate dispatch telemetry. */
+  observer?: LlmStreamingControlDispatchObserver
+}
+
+export type LlmStreamingControlDispatchEvent
+  = | { type: 'rejected', reason: 'no-matching-parser' | 'parse-failed', parserName?: string }
+    | { type: 'parsed', parserName: string, tokenType: LlmStreamingControlSignal['type'], callName?: string, parameter?: string }
+    | { type: 'call-handler-count', count: number }
+    | { type: 'call-handler-missing', callName: string, payload?: Record<string, unknown> }
+    | { type: 'call-handler-start', callName: string }
+    | { type: 'call-handler-end', callName: string }
+    | { type: 'call-handler-error', callName: string, error: unknown }
+    | { type: 'signal-handler-error', tokenType: LlmStreamingControlSignal['type'], error: unknown }
+
+export type LlmStreamingControlDispatchObserver = (event: LlmStreamingControlDispatchEvent) => void
+
+/**
+ * Runtime context passed to a registered CALL handler.
+ */
+export interface LlmStreamingControlCallContext extends LlmStreamingControlDispatchContext {
+  /** Local timestamp for ordering and debugging. */
+  createdAt: number
+}
+
+/**
+ * Runtime context passed to a parsed streaming-control signal handler.
+ */
+export interface LlmStreamingControlSignalContext extends LlmStreamingControlDispatchContext {
+  /** Local timestamp for ordering and debugging. */
+  createdAt: number
+}
+
+export type LlmStreamingControlCallHandler<
+  TPayload extends Record<string, unknown> = Record<string, unknown>,
+> = (
+  payload: TPayload | undefined,
+  context: LlmStreamingControlCallContext,
+) => void | Promise<void>
+
+export type LlmStreamingControlSignalHandler = (
+  signal: LlmStreamingControlSignal,
+  context: LlmStreamingControlSignalContext,
+) => void | Promise<void>
+
+/**
+ * Provider-facing manifest for one streaming `CALL` token.
+ *
+ * The manifest is registered with the callback so the runtime can render
+ * few-shot instructions from the same source of truth that executes the call.
+ */
+export interface LlmStreamingControlCallManifest {
+  /** Opaque plugin-owned call name used inside `<|CALL ["name"]|>`. */
+  name: string
+  /** Required model instruction describing when this call should be emitted. */
+  prompt: string
+  /** Optional few-shot token examples. */
+  examples?: string[]
+}
+
+export type LlmStreamingControlTurnDoneReason = 'completed' | 'cancelled'
+
+export interface LlmStreamingControlTurnDone {
+  type: LlmStreamingControlTurnDoneReason
+}
+
+export interface LlmStreamingControlTurn {
+  turnId: string
+  on: <TPayload extends Record<string, unknown> = Record<string, unknown>>(
+    manifest: LlmStreamingControlCallManifest,
+    handler: LlmStreamingControlCallHandler<TPayload>,
+  ) => () => void
+  renderManifestPrompt: () => string
+  complete: () => void
+  cancel: () => void
+  done: Promise<LlmStreamingControlTurnDone>
+}
+
+/**
+ * Parser for one LLM streaming control syntax.
+ *
+ * @param TParsed Parsed token payload owned by the matching control.
+ */
+export interface LlmStreamingControlParser<TParsed> {
+  /** Token syntax name, for example `CALL`. */
+  name: string
+  /** Returns true when this parser owns the special token syntax. */
+  match: (special: string) => boolean
+  /** Parses a special token into plain data. This method must not perform side effects. */
+  parse: (special: string) => TParsed | undefined
+}
+
+/**
+ * Runtime for LLM streaming controls embedded in `<|...|>` special tokens.
+ */
+export interface LlmStreamingControl {
+  /** Returns true when any loaded control parser recognizes the special token. */
+  match: (special: string) => boolean
+  /**
+   * Parses and dispatches one special token through the first matching control.
+   *
+   * Returns:
+   * - `true` when a loaded control matched and parsed the token
+   * - `false` when callers should continue normal special-token handling
+   */
+  dispatchWith: (special: string, context?: Partial<LlmStreamingControlDispatchContext>) => Promise<boolean>
+  /**
+   * Registers a callback for one plugin-owned CALL name.
+   *
+   * Returns:
+   * - A disposer that unregisters the callback
+   */
+  on: <TPayload extends Record<string, unknown> = Record<string, unknown>>(
+    manifest: LlmStreamingControlCallManifest,
+    handler: LlmStreamingControlCallHandler<TPayload>,
+  ) => () => void
+  /**
+   * Renders currently registered CALL manifests into prompt instructions.
+   *
+   * Returns:
+   * - Empty string when no CALL manifests are registered
+   * - A provider-safe instruction block with syntax rules and examples otherwise
+   */
+  renderManifestPrompt: () => string
+  /**
+   * Registers a callback for every parsed streaming-control signal.
+   *
+   * Returns:
+   * - A disposer that unregisters the callback
+   */
+  onSignal: (handler: LlmStreamingControlSignalHandler) => () => void
+  beginTurn: (options?: { turnId?: string }) => LlmStreamingControlTurn
+  completeTurn: (turnId: string) => void
+  cancelTurn: (turnId: string) => void
+}
+
+export interface LlmStreamingControlOptions {
+  /** Optional parsers. Defaults to the built-in ACT, DELAY, and CALL parsers. */
+  parsers?: LlmStreamingControlParser<LlmStreamingControlSignal>[]
+}
+
+/**
+ * Parsed CALL token payload.
+ *
+ * @param TPayload Payload object shape expected by one CALL name.
+ */
+export interface LlmStreamingControlTokenCall<
+  TPayload extends Record<string, unknown> = Record<string, unknown>,
+> {
+  /** Parsed signal kind. */
+  type: 'call'
+  /** Plugin-owned call name. Stage UI treats this as opaque data. */
+  name: string
+  /** Optional call payload reserved for plugin-owned extensions. */
+  payload?: TPayload
+}
+
+/**
+ * Parsed ACT token payload.
+ */
+export interface LlmStreamingControlTokenAct {
+  /** Parsed signal kind. */
+  type: 'act'
+  /** Parsed JSON object literal supplied by the model. */
+  payload: Record<string, unknown>
+}
+
+/**
+ * Parsed DELAY token payload.
+ */
+export interface LlmStreamingControlTokenDelay {
+  /** Parsed signal kind. */
+  type: 'delay'
+  /** Delay length in seconds. */
+  seconds: number
+}
+
+export type LlmStreamingControlSignal
+  = | LlmStreamingControlTokenAct
+    | LlmStreamingControlTokenCall
+    | LlmStreamingControlTokenDelay
@@ -384,7 +384,7 @@ export function processNarrative(text: string, options?: TtsInputChunkOptions):

 export function createTtsSegmentStream(
  tokens: ReadableStream<TextToken>,
-  meta: { streamId: string, intentId: string },
+  meta: { streamId: string, intentId: string, turnId?: string },
  options?: TtsInputChunkOptions,
 ) {
  const { stream, write, close, error } = createPushStream<TextSegment>()
@@ -483,6 +483,7 @@ export function createTtsSegmentStream(
      const reader = byteStream.getReader()
      await chunkEmitter(reader, pendingSpecials, options, async (chunk) => {
        write({
+          turnId: meta.turnId,
          streamId: meta.streamId,
          intentId: meta.intentId,
          segmentId: `${meta.streamId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`,
@@ -51,20 +51,30 @@ function createSegmenter(texts: string[]) {
  }
 }

-function createPlaybackSpy() {
+function createPlaybackSpy(options?: { autoEnd?: boolean }) {
  const scheduled: Array<PlaybackItem<string>> = []
+  const endListeners: Array<(event: { item: PlaybackItem<string>, endedAt: number }) => void> = []
+  const autoEnd = options?.autoEnd ?? true

  return {
    scheduled,
+    end(item: PlaybackItem<string>) {
+      for (const listener of endListeners)
+        listener({ item, endedAt: Date.now() })
+    },
    playback: {
      schedule(item: PlaybackItem<string>) {
        scheduled.push(item)
+        if (autoEnd)
+          queueMicrotask(() => endListeners.forEach(listener => listener({ item, endedAt: Date.now() })))
      },
      stopAll: vi.fn(),
      stopByIntent: vi.fn(),
      stopByOwner: vi.fn(),
      onStart: vi.fn(),
-      onEnd: vi.fn(),
+      onEnd(listener: (event: { item: PlaybackItem<string>, endedAt: number }) => void) {
+        endListeners.push(listener)
+      },
      onInterrupt: vi.fn(),
      onReject: vi.fn(),
    },
@@ -187,4 +197,117 @@ describe('createSpeechPipeline', () => {
    expect(abortedRequests.sort()).toEqual([0, 1])
    expect(scheduled).toEqual([])
  })
+
+  it('dispatches special controls only after the preceding playback item ends', async () => {
+    const { scheduled, playback, end } = createPlaybackSpy({ autoEnd: false })
+    const events: string[] = []
+
+    const pipeline = createSpeechPipeline<string>({
+      ttsMaxConcurrent: 2,
+      segmenter: (_tokens, meta) => {
+        return new ReadableStream<TextSegment>({
+          start(controller) {
+            controller.enqueue({
+              turnId: meta.turnId,
+              streamId: meta.streamId,
+              intentId: meta.intentId,
+              segmentId: 'segment:0',
+              text: 'before',
+              special: null,
+              reason: 'flush',
+              createdAt: Date.now(),
+            })
+            controller.enqueue({
+              turnId: meta.turnId,
+              streamId: meta.streamId,
+              intentId: meta.intentId,
+              segmentId: 'segment:1',
+              text: '',
+              special: '<|CALL ["plugin.action"]|>',
+              reason: 'special',
+              createdAt: Date.now(),
+            })
+            controller.close()
+          },
+        })
+      },
+      playback,
+      async tts(request) {
+        events.push(`tts:${request.text}`)
+        return request.text
+      },
+    })
+
+    pipeline.on('onSpecial', (segment) => {
+      events.push(`special:${segment.special}`)
+      events.push(`turn:${segment.turnId}`)
+    })
+
+    const intent = pipeline.openIntent({ turnId: 'turn-1' })
+    intent.end()
+
+    await delay(0)
+    expect(scheduled.map(item => item.text)).toEqual(['before'])
+    expect(events).toEqual(['tts:before'])
+
+    end(scheduled[0]!)
+    await delay(0)
+
+    expect(events).toEqual([
+      'tts:before',
+      'special:<|CALL ["plugin.action"]|>',
+      'turn:turn-1',
+    ])
+  })
+
+  it('does not schedule queued timeline playback after the owning intent is cancelled', async () => {
+    const { scheduled, playback, end } = createPlaybackSpy({ autoEnd: false })
+
+    const pipeline = createSpeechPipeline<string>({
+      ttsMaxConcurrent: 2,
+      segmenter: (_tokens, meta) => {
+        return new ReadableStream<TextSegment>({
+          start(controller) {
+            controller.enqueue({
+              turnId: meta.turnId,
+              streamId: meta.streamId,
+              intentId: meta.intentId,
+              segmentId: 'segment:0',
+              text: 'first',
+              special: null,
+              reason: 'flush',
+              createdAt: Date.now(),
+            })
+            controller.enqueue({
+              turnId: meta.turnId,
+              streamId: meta.streamId,
+              intentId: meta.intentId,
+              segmentId: 'segment:1',
+              text: 'second',
+              special: null,
+              reason: 'flush',
+              createdAt: Date.now(),
+            })
+            controller.close()
+          },
+        })
+      },
+      playback,
+      async tts(request) {
+        return request.text
+      },
+    })
+
+    const intent = pipeline.openIntent({ intentId: 'intent-1', turnId: 'turn-1' })
+    intent.end()
+
+    await delay(0)
+    expect(scheduled.map(item => item.text)).toEqual(['first'])
+
+    intent.cancel('newer-intent')
+    end(scheduled[0]!)
+    await delay(0)
+
+    expect(scheduled.map(item => item.text)).toEqual(['first'])
+  })
 })
@@ -19,6 +19,7 @@ import { speechPipelineEventMap } from './eventa'
 import { createPriorityResolver } from './priority'
 import { createTtsSegmentStream } from './processors/tts-chunker'
 import { createPushStream } from './stream'
+import { createTimeline } from './timeline'

 export interface SpeechPipelineOptions<TAudio> {
  tts: (request: TtsRequest, signal: AbortSignal) => Promise<TAudio | null>
@@ -40,10 +41,11 @@ export interface SpeechPipelineOptions<TAudio> {
  }
  logger?: LoggerLike
  priority?: ReturnType<typeof createPriorityResolver>
-  segmenter?: (tokens: ReadableStream<TextToken>, meta: { streamId: string, intentId: string }) => ReadableStream<TextSegment>
+  segmenter?: (tokens: ReadableStream<TextToken>, meta: { streamId: string, intentId: string, turnId?: string }) => ReadableStream<TextSegment>
 }

 interface IntentState {
+  turnId?: string
  intentId: string
  streamId: string
  priority: number
@@ -66,15 +68,42 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
  const segmenter = options.segmenter ?? createTtsSegmentStream
  const ttsMaxConcurrent = Math.max(1, options.ttsMaxConcurrent ?? 4)
  const context = createContext()
+  const timeline = createTimeline()

  const intents = new Map<string, IntentState>()
  const pending: IntentState[] = []
  let activeIntent: IntentState | null = null
+  const playbackWaiters = new Map<string, () => void>()
+
+  function resolvePlayback(itemId: string) {
+    const resolve = playbackWaiters.get(itemId)
+    if (!resolve)
+      return
+
+    playbackWaiters.delete(itemId)
+    resolve()
+  }

  options.playback.onStart(event => context.emit(speechPipelineEventMap.onPlaybackStart, event))
-  options.playback.onEnd(event => context.emit(speechPipelineEventMap.onPlaybackEnd, event))
-  options.playback.onInterrupt(event => context.emit(speechPipelineEventMap.onPlaybackInterrupt, event))
-  options.playback.onReject(event => context.emit(speechPipelineEventMap.onPlaybackReject, event))
+  options.playback.onEnd((event) => {
+    context.emit(speechPipelineEventMap.onPlaybackEnd, event)
+    resolvePlayback(event.item.id)
+  })
+  options.playback.onInterrupt((event) => {
+    context.emit(speechPipelineEventMap.onPlaybackInterrupt, event)
+    resolvePlayback(event.item.id)
+  })
+  options.playback.onReject((event) => {
+    context.emit(speechPipelineEventMap.onPlaybackReject, event)
+    resolvePlayback(event.item.id)
+  })
+
+  function waitForPlayback(item: PlaybackItem<TAudio>) {
+    return new Promise<void>((resolve) => {
+      playbackWaiters.set(item.id, resolve)
+      options.playback.schedule(item)
+    })
+  }

  function enqueueIntent(intent: IntentState) {
    pending.push(intent)
@@ -90,22 +119,67 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
  async function runIntent(intent: IntentState) {
    activeIntent = intent
    context.emit(speechPipelineEventMap.onIntentStart, intent.intentId)
+    if (intent.turnId)
+      context.emit(speechPipelineEventMap.onTurnStart, intent.turnId)

    const tokenStream = intent.stream
-    const segmentStream = segmenter(tokenStream, { streamId: intent.streamId, intentId: intent.intentId })
+    const segmentStream = segmenter(tokenStream, { streamId: intent.streamId, intentId: intent.intentId, turnId: intent.turnId })
    const completedRequests = new Map<number, TtsResult<TAudio> | null>()
    const inFlightTasks = new Set<Promise<void>>()
    let nextRequestSequence = 0
    let nextSequenceToSchedule = 0

+    function enqueueSpecial(segment: TextSegment) {
+      timeline.enqueue({
+        id: `special:${segment.segmentId}`,
+        track: 'speech',
+        run() {
+          if (intent.canceled || intent.controller.signal.aborted)
+            return
+
+          context.emit(speechPipelineEventMap.onSpecial, segment)
+        },
+      })
+    }
+
+    function enqueuePlayback(item: PlaybackItem<TAudio>) {
+      timeline.enqueue({
+        id: `playback:${item.id}`,
+        track: 'speech',
+        async run() {
+          if (intent.canceled || intent.controller.signal.aborted)
+            return
+
+          await waitForPlayback(item)
+
+          if (intent.canceled || intent.controller.signal.aborted)
+            return
+
+          if (item.special) {
+            context.emit(speechPipelineEventMap.onSpecial, {
+              turnId: item.turnId,
+              streamId: item.streamId,
+              intentId: item.intentId,
+              segmentId: item.segmentId,
+              text: item.text,
+              special: item.special,
+              reason: 'special',
+              createdAt: item.createdAt,
+            })
+          }
+        },
+      })
+    }
+
    function scheduleCompletedRequests() {
      while (completedRequests.has(nextSequenceToSchedule)) {
        const completedRequest = completedRequests.get(nextSequenceToSchedule) ?? null
        completedRequests.delete(nextSequenceToSchedule)

        if (completedRequest) {
-          options.playback.schedule({
+          enqueuePlayback({
            id: createId('playback'),
+            turnId: completedRequest.turnId,
            streamId: completedRequest.streamId,
            intentId: completedRequest.intentId,
            segmentId: completedRequest.segmentId,
@@ -148,6 +222,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
        }

        const ttsResult: TtsResult<TAudio> = {
+          turnId: request.turnId,
          streamId: request.streamId,
          intentId: request.intentId,
          segmentId: request.segmentId,
@@ -191,11 +266,12 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
        context.emit(speechPipelineEventMap.onSegment, value)

        if (value.text === '' && value.special) {
-          context.emit(speechPipelineEventMap.onSpecial, value)
+          enqueueSpecial(value)
          continue
        }

        const request: TtsRequest = {
+          turnId: value.turnId,
          streamId: value.streamId,
          intentId: value.intentId,
          segmentId: value.segmentId,
@@ -212,6 +288,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud

      await Promise.allSettled(inFlightTasks)
      scheduleCompletedRequests()
+      await timeline.flush('speech')
      reader.releaseLock()
    }
    catch (err) {
@@ -220,9 +297,13 @@ 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 })
+        if (intent.turnId)
+          context.emit(speechPipelineEventMap.onTurnCancel, { turnId: intent.turnId, reason: intent.controller.signal.reason as string | undefined })
      }
      else {
        context.emit(speechPipelineEventMap.onIntentEnd, intent.intentId)
+        if (intent.turnId)
+          context.emit(speechPipelineEventMap.onTurnEnd, intent.turnId)
      }

      intents.delete(intent.intentId)
@@ -239,6 +320,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud

  function openIntent(optionsInput?: IntentOptions): IntentHandle {
    const intentId = optionsInput?.intentId ?? createId('intent')
+    const turnId = optionsInput?.turnId
    const streamId = optionsInput?.streamId ?? createId('stream')
    const priority = priorityResolver.resolve(optionsInput?.priority)
    const behavior = optionsInput?.behavior ?? 'queue'
@@ -249,6 +331,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
    let sequence = 0

    const intent: IntentState = {
+      turnId,
      intentId,
      streamId,
      priority,
@@ -264,6 +347,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
    intents.set(intentId, intent)

    const handle: IntentHandle = {
+      turnId,
      intentId,
      streamId,
      priority,
@@ -275,6 +359,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
        write({
          type: 'literal',
          value: text,
+          turnId,
          streamId,
          intentId,
          sequence: sequence++,
@@ -287,6 +372,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
        write({
          type: 'special',
          value: special,
+          turnId,
          streamId,
          intentId,
          sequence: sequence++,
@@ -298,6 +384,7 @@ export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAud
          return
        write({
          type: 'flush',
+          turnId,
          streamId,
          intentId,
          sequence: sequence++,
@@ -0,0 +1,102 @@
+export interface TimelineClock {
+  now: () => number
+  sleep: (ms: number) => Promise<void>
+}
+
+export interface TimelineRunContext {
+  clock: TimelineClock
+  signal: AbortSignal
+}
+
+export interface TimelineItem {
+  id?: string
+  track: string
+  run: (context: TimelineRunContext) => Promise<void> | void
+}
+
+export interface TimelineItemHandle {
+  id: string
+  track: string
+  done: Promise<void>
+  cancel: (reason?: string) => void
+}
+
+export interface TimelineOptions {
+  clock?: TimelineClock
+}
+
+function createDefaultClock(): TimelineClock {
+  return {
+    now: () => Date.now(),
+    sleep: ms => new Promise(resolve => setTimeout(resolve, ms)),
+  }
+}
+
+function createId(prefix: string) {
+  return `${prefix}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`
+}
+
+/**
+ * Creates a thin named-track timeline for ordering async media/control work.
+ *
+ * Use when:
+ * - Multiple async queues need a shared ordering clock
+ * - Items on the same named track must execute serially
+ *
+ * Expects:
+ * - Item handlers own their business side effects
+ * - Cancellation is cooperative through the provided signal
+ *
+ * Returns:
+ * - A scheduler that serializes each track while allowing different tracks to run concurrently
+ */
+export function createTimeline(options?: TimelineOptions) {
+  const clock = options?.clock ?? createDefaultClock()
+  const trackTails = new Map<string, Promise<void>>()
+
+  function enqueue(item: TimelineItem): TimelineItemHandle {
+    const id = item.id ?? createId('timeline')
+    const controller = new AbortController()
+    const previous = trackTails.get(item.track)
+
+    const run = async () => {
+      if (controller.signal.aborted)
+        return
+
+      await item.run({
+        clock,
+        signal: controller.signal,
+      })
+    }
+
+    const done = previous
+      ? previous.catch(() => undefined).then(run)
+      : run()
+
+    trackTails.set(item.track, done.catch(() => undefined))
+
+    return {
+      id,
+      track: item.track,
+      done,
+      cancel(reason?: string) {
+        controller.abort(reason)
+      },
+    }
+  }
+
+  async function flush(track?: string) {
+    if (track) {
+      await trackTails.get(track)
+      return
+    }
+
+    await Promise.all(trackTails.values())
+  }
+
+  return {
+    clock,
+    enqueue,
+    flush,
+  }
+}
@@ -7,6 +7,7 @@ export interface PriorityResolver {
 export interface TextToken {
  type: 'literal' | 'special' | 'flush'
  value?: string
+  turnId?: string
  streamId: string
  intentId: string
  sequence: number
@@ -14,6 +15,7 @@ export interface TextToken {
 }

 export interface TextSegment {
+  turnId?: string
  streamId: string
  intentId: string
  segmentId: string
@@ -24,6 +26,7 @@ export interface TextSegment {
 }

 export interface TtsRequest {
+  turnId?: string
  streamId: string
  intentId: string
  segmentId: string
@@ -35,6 +38,7 @@ export interface TtsRequest {
 }

 export interface TtsResult<TAudio> {
+  turnId?: string
  streamId: string
  intentId: string
  segmentId: string
@@ -47,6 +51,7 @@ export interface TtsResult<TAudio> {

 export interface PlaybackItem<TAudio> {
  id: string
+  turnId?: string
  streamId: string
  intentId: string
  segmentId: string
@@ -83,6 +88,7 @@ export interface PlaybackRejectEvent<TAudio> {
 export type IntentBehavior = 'queue' | 'interrupt' | 'replace'

 export interface IntentOptions {
+  turnId?: string
  intentId?: string
  streamId?: string
  priority?: PriorityLevel | number
@@ -91,6 +97,7 @@ export interface IntentOptions {
 }

 export interface IntentHandle {
+  turnId?: string
  intentId: string
  streamId: string
  priority: number
@@ -114,7 +121,10 @@ export interface SpeechPipelineEvents<TAudio> {
  onPlaybackReject: (event: PlaybackRejectEvent<TAudio>) => void
  onIntentStart: (intentId: string) => void
  onIntentEnd: (intentId: string) => void
-  onIntentCancel: (intentId: string, reason?: string) => void
+  onIntentCancel: (event: { intentId: string, reason?: string }) => void
+  onTurnStart: (turnId: string) => void
+  onTurnEnd: (turnId: string) => void
+  onTurnCancel: (event: { turnId: string, reason?: string }) => void
 }

 export interface LoggerLike {
@@ -3,6 +3,7 @@ import { createContext as createBroadcastChannelContext } from '@moeru/eventa/ad

 export interface SpeechIntentStartPayload {
  originId: string
+  turnId?: string
  intentId: string
  streamId: string
  ownerId?: string
@@ -12,6 +13,7 @@ export interface SpeechIntentStartPayload {

 export interface SpeechIntentTokenPayload {
  originId: string
+  turnId?: string
  intentId: string
  streamId: string
  sequence: number
@@ -20,12 +22,14 @@ export interface SpeechIntentTokenPayload {

 export interface SpeechIntentEndPayload {
  originId: string
+  turnId?: string
  intentId: string
  streamId: string
 }

 export interface SpeechIntentCancelPayload {
  originId: string
+  turnId?: string
  intentId: string
  streamId: string
  reason?: string
@@ -55,6 +55,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
        return

      const intent = hostPipeline.openIntent({
+        turnId: payload.turnId,
        intentId: payload.intentId,
        streamId: payload.streamId,
        ownerId: payload.ownerId,
@@ -72,7 +73,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
      if (!intent) {
        if (!hostPipeline)
          return
-        const fallback = hostPipeline.openIntent({ intentId: payload.intentId, streamId: payload.streamId })
+        const fallback = hostPipeline.openIntent({ turnId: payload.turnId, intentId: payload.intentId, streamId: payload.streamId })
        remoteIntentMap.set(payload.intentId, fallback)
        writer(fallback, payload.value)
        return
@@ -137,6 +138,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {

  function createRemoteIntent(options?: IntentOptions): IntentHandle {
    const intentId = options?.intentId ?? createId('intent')
+    const turnId = options?.turnId
    const streamId = options?.streamId ?? createId('stream')
    const priority = typeof options?.priority === 'number' ? options?.priority : undefined
    const behavior = options?.behavior
@@ -148,6 +150,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {

    context.emit(speechIntentStartEvent, {
      originId,
+      turnId,
      intentId,
      streamId,
      ownerId,
@@ -157,6 +160,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {

    const handle: IntentHandle = {
      intentId,
+      turnId,
      streamId,
      ownerId,
      priority: priority ?? 0,
@@ -164,9 +168,10 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
      writeLiteral(value: string) {
        if (closed)
          return
-        write({ type: 'literal', value, streamId, intentId, sequence, createdAt: Date.now() })
+        write({ type: 'literal', value, turnId, streamId, intentId, sequence, createdAt: Date.now() })
        context.emit(speechIntentLiteralEvent, {
          originId,
+          turnId,
          intentId,
          streamId,
          sequence: sequence++,
@@ -176,9 +181,10 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
      writeSpecial(value: string) {
        if (closed)
          return
-        write({ type: 'special', value, streamId, intentId, sequence, createdAt: Date.now() })
+        write({ type: 'special', value, turnId, streamId, intentId, sequence, createdAt: Date.now() })
        context.emit(speechIntentSpecialEvent, {
          originId,
+          turnId,
          intentId,
          streamId,
          sequence: sequence++,
@@ -188,9 +194,10 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
      writeFlush() {
        if (closed)
          return
-        write({ type: 'flush', streamId, intentId, sequence, createdAt: Date.now() })
+        write({ type: 'flush', turnId, streamId, intentId, sequence, createdAt: Date.now() })
        context.emit(speechIntentFlushEvent, {
          originId,
+          turnId,
          intentId,
          streamId,
          sequence: sequence++,
@@ -203,6 +210,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
        close()
        context.emit(speechIntentEndEvent, {
          originId,
+          turnId,
          intentId,
          streamId,
        })
@@ -214,6 +222,7 @@ export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
        close()
        context.emit(speechIntentCancelEvent, {
          originId,
+          turnId,
          intentId,
          streamId,
          reason,
@@ -79,6 +79,7 @@ export const useCharacterStore = defineStore('character', () => {
      }) satisfies CharacterSparkNotifyReaction

      const intent = speechRuntimeStore.openIntent({
+        turnId: `spark:${sparkEventId}`,
        intentId: `spark:${sparkEventId}`,
        ownerId: ownerId.value,
        priority: 'high',
@@ -80,12 +80,17 @@ vi.mock('../composables', () => ({
 }))

 vi.mock('../composables/llm-marker-parser', () => ({
-  useLlmmarkerParser: (options: { onLiteral?: (literal: string) => Promise<void>, onEnd?: (fullText: string) => Promise<void> }) => {
+  useLlmmarkerParser: (options: { onLiteral?: (literal: string) => Promise<void>, onSpecial?: (special: string) => Promise<void>, onEnd?: (fullText: string) => Promise<void> }) => {
    let fullText = ''
    return {
      consume: async (textPart: string) => {
        parserConsumeMock(textPart)
        fullText += textPart
+        if (textPart.startsWith('<|') && textPart.endsWith('|>')) {
+          await options.onSpecial?.(textPart)
+          return
+        }
+
        await options.onLiteral?.(textPart)
      },
      end: async () => {
@@ -152,6 +157,12 @@ vi.mock('./llm', () => ({
  }),
 }))

+vi.mock('./llm-toolset-prompts', () => ({
+  useLlmToolsetPromptsStore: () => ({
+    activeToolsetPrompt: 'Plugin toolset guidance.',
+  }),
+}))
+
 vi.mock('./modules/consciousness', () => ({
  useConsciousnessStore: () => ({
    activeProvider: ref('mock-provider'),
@@ -293,7 +304,8 @@ describe('chat orchestrator contract', () => {
    // instead of a system anchor).
    const systemContent = (composedMessages[0] as any).content
    const systemText = typeof systemContent === 'string' ? systemContent : systemContent.map((p: any) => p.text).join('')
-    expect(systemText).toBe('system prompt')
+    expect(systemText).toContain('system prompt')
+    expect(systemText).toContain('Plugin toolset guidance.')

    // The user turn is prefixed with [YYYY-MM-DD HH:MM]. Both historic and
    // current turns share the same shape so prefix-cache stays valid when a
@@ -310,6 +322,26 @@ describe('chat orchestrator contract', () => {
    expect(syntheticContextText).toContain('- system:weather: sunny')
  })

+  it('emits special tokens for speech timeline handling during chat streaming', async () => {
+    getContextsSnapshotMock.mockReturnValue({})
+    llmStreamMock.mockImplementationOnce(async (_model, _provider, _messages, options) => {
+      await options.onStreamEvent({ type: 'text-delta', text: '<|CALL ["plugin.action"]|>' })
+    })
+
+    const store = useChatOrchestratorStore()
+    const specialHook = vi.fn()
+    store.onTokenSpecial(specialHook)
+
+    await store.ingest('trigger special', {
+      chatProvider: provider,
+      model: 'mock-model',
+    })
+
+    expect(specialHook).toHaveBeenCalledWith('<|CALL ["plugin.action"]|>', expect.objectContaining({
+      contexts: {},
+    }))
+  })
+
  it('rejects cancelled queued sends before they start', async () => {
    llmStreamMock.mockImplementation(async () => {
      // keep pending
@@ -25,6 +25,7 @@ import { useChatSessionStore } from './chat/session-store'
 import { useChatStreamStore } from './chat/stream-store'
 import { useContextObservabilityStore } from './devtools/context-observability'
 import { useLLM } from './llm'
+import { useLlmToolsetPromptsStore } from './llm-toolset-prompts'
 import { useAiriCardStore } from './modules/airi-card'
 import { useAutonomousArtistryStore } from './modules/artistry-autonomous'
 import { useConsciousnessStore } from './modules/consciousness'
@@ -108,6 +109,7 @@ export interface QueuedSendSnapshot {

 export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => {
  const llmStore = useLLM()
+  const llmToolsetPromptsStore = useLlmToolsetPromptsStore()
  const consciousnessStore = useConsciousnessStore()
  const artistryAutonomousStore = useAutonomousArtistryStore()
  const { activeProvider } = storeToRefs(consciousnessStore)
@@ -377,6 +379,20 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => {
        return rawMessage
      })

+      const activeToolsetPrompt = llmToolsetPromptsStore.activeToolsetPrompt.trim()
+      if (activeToolsetPrompt) {
+        const systemMessage = newMessages.find(message => message.role === 'system')
+        if (systemMessage) {
+          systemMessage.content = `${systemMessage.content}\n\n${activeToolsetPrompt}`
+        }
+        else {
+          newMessages.unshift({
+            role: 'system',
+            content: activeToolsetPrompt,
+          })
+        }
+      }
+
      const contextsSnapshot = chatContext.getContextsSnapshot()
      const contextPromptText = formatContextPromptText(contextsSnapshot)
      if (contextPromptText) {
@@ -0,0 +1,98 @@
+import { IOAttributes, IOEvents, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared'
+import { createPinia, setActivePinia } from 'pinia'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { shallowRef } from 'vue'
+
+const spanMock = vi.hoisted(() => ({
+  addEvent: vi.fn(),
+  end: vi.fn(),
+  setAttribute: vi.fn(),
+}))
+
+const startSpanMock = vi.hoisted(() => vi.fn(() => spanMock))
+
+vi.mock('../composables/use-io-tracer', () => ({
+  activeTurnSpan: shallowRef(undefined),
+  startSpan: startSpanMock,
+}))
+
+describe('useLlmStreamingControlStore', () => {
+  beforeEach(() => {
+    setActivePinia(createPinia())
+    startSpanMock.mockClear()
+    spanMock.addEvent.mockClear()
+    spanMock.end.mockClear()
+    spanMock.setAttribute.mockClear()
+  })
+
+  /**
+   * @example
+   * await store.dispatchWith('<|CALL ["plugin.action"]|>')
+   * expect(startSpan).toHaveBeenCalledWith('Streaming control dispatch', ...)
+   */
+  it('records streaming control dispatch spans and call handler events', async () => {
+    const { useLlmStreamingControlStore } = await import('./llm-streaming-control')
+    const store = useLlmStreamingControlStore()
+    const handler = vi.fn()
+
+    store.on({
+      name: 'plugin.action',
+      prompt: 'Run the plugin action.',
+    }, handler)
+
+    await expect(store.dispatchWith('<|CALL ["plugin.action"]|>')).resolves.toBe(true)
+
+    expect(startSpanMock).toHaveBeenCalledWith(
+      IOSpanNames.StreamingControlDispatch,
+      undefined,
+      expect.objectContaining({
+        [IOAttributes.StreamingControlTokenLength]: '<|CALL ["plugin.action"]|>'.length,
+        [IOAttributes.Subsystem]: IOSubsystems.StreamingControl,
+      }),
+    )
+    expect(spanMock.setAttribute).toHaveBeenCalledWith(IOAttributes.StreamingControlParserName, 'CALL')
+    expect(spanMock.setAttribute).toHaveBeenCalledWith(IOAttributes.StreamingControlTokenType, 'call')
+    expect(spanMock.setAttribute).toHaveBeenCalledWith(IOAttributes.StreamingControlCallName, 'plugin.action')
+    expect(spanMock.setAttribute).toHaveBeenCalledWith(IOAttributes.StreamingControlHandlerCount, 1)
+    expect(spanMock.addEvent).toHaveBeenCalledWith(
+      IOEvents.StreamingControlParsed,
+      expect.objectContaining({
+        [IOAttributes.StreamingControlTokenType]: 'call',
+      }),
+    )
+    expect(spanMock.addEvent).toHaveBeenCalledWith(
+      IOEvents.StreamingControlHandlerStart,
+      expect.objectContaining({
+        [IOAttributes.StreamingControlCallName]: 'plugin.action',
+      }),
+    )
+    expect(spanMock.addEvent).toHaveBeenCalledWith(
+      IOEvents.StreamingControlHandlerEnd,
+      expect.objectContaining({
+        [IOAttributes.StreamingControlCallName]: 'plugin.action',
+      }),
+    )
+    expect(spanMock.end).toHaveBeenCalledTimes(1)
+    expect(handler).toHaveBeenCalledTimes(1)
+  })
+
+  /**
+   * @example
+   * await store.dispatchWith('<|CALL []|>')
+   * expect(span.addEvent).toHaveBeenCalledWith(IOEvents.StreamingControlRejected, ...)
+   */
+  it('records rejected streaming control dispatches', async () => {
+    const { useLlmStreamingControlStore } = await import('./llm-streaming-control')
+    const store = useLlmStreamingControlStore()
+
+    await expect(store.dispatchWith('<|CALL []|>')).resolves.toBe(false)
+
+    expect(spanMock.addEvent).toHaveBeenCalledWith(
+      IOEvents.StreamingControlRejected,
+      expect.objectContaining({
+        [IOAttributes.StreamingControlReason]: 'parse-failed',
+      }),
+    )
+    expect(spanMock.end).toHaveBeenCalledTimes(1)
+  })
+})
@@ -0,0 +1,150 @@
+import type { LlmStreamingControlDispatchContext, LlmStreamingControlDispatchEvent } from '@proj-airi/pipelines-audio'
+
+import { errorMessageFrom } from '@moeru/std'
+import { createStreamingControlParser } from '@proj-airi/pipelines-audio'
+import { IOAttributes, IOEvents, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared'
+import { useBroadcastChannel } from '@vueuse/core'
+import { nanoid } from 'nanoid'
+import { defineStore } from 'pinia'
+import { watch } from 'vue'
+
+import { activeTurnSpan, startSpan } from '../composables/use-io-tracer'
+
+interface RemoteCallMessage {
+  type: 'turn-call'
+  fromInstanceId: string
+  turnId: string
+  callName: string
+  payload?: Record<string, unknown>
+}
+
+export const useLlmStreamingControlStore = defineStore('llm-streaming-control', () => {
+  const controller = createStreamingControlParser()
+  const instanceId = `streaming-control-${nanoid()}`
+
+  const { post: postRemoteCall, data: incomingRemoteCall } = useBroadcastChannel<RemoteCallMessage, RemoteCallMessage>({ name: 'airi-streaming-control-turn-calls' })
+
+  const tooltipKeys = [
+    'token_type',
+    'call_name',
+    'parameter',
+    'handler_count',
+    'turn_id',
+    'reason',
+    'raw_token',
+  ]
+
+  watch(incomingRemoteCall, (message) => {
+    if (!message || message.type !== 'turn-call')
+      return
+    if (message.fromInstanceId === instanceId)
+      return
+
+    const callPayload = message.payload === undefined
+      ? [message.callName]
+      : [message.callName, message.payload]
+    void dispatchWith(`<|CALL ${JSON.stringify(callPayload)}|>`, {
+      turnId: message.turnId,
+      remote: true,
+    })
+  })
+
+  async function dispatchWith(special: string, context?: Partial<LlmStreamingControlDispatchContext>) {
+    const span = startSpan(IOSpanNames.StreamingControlDispatch, activeTurnSpan.value, {
+      [IOAttributes.StreamingControlMatched]: false,
+      [IOAttributes.StreamingControlParsed]: false,
+      [IOAttributes.StreamingControlTokenLength]: special.length,
+      [IOAttributes.Subsystem]: IOSubsystems.StreamingControl,
+      ...(context?.turnId ? { [IOAttributes.StreamingControlTurnId]: context.turnId } : {}),
+    })
+    span.setAttribute(IOAttributes.TooltipKeys, tooltipKeys)
+
+    function observe(event: LlmStreamingControlDispatchEvent) {
+      switch (event.type) {
+        case 'rejected':
+          span.setAttribute(IOAttributes.StreamingControlReason, event.reason)
+          span.setAttribute(IOAttributes.StreamingControlRawToken, special)
+          if (event.parserName) {
+            span.setAttribute(IOAttributes.StreamingControlMatched, true)
+            span.setAttribute(IOAttributes.StreamingControlParserName, event.parserName)
+          }
+          span.addEvent(IOEvents.StreamingControlRejected, {
+            [IOAttributes.StreamingControlReason]: event.reason,
+            [IOAttributes.StreamingControlRawToken]: special,
+          })
+          break
+        case 'parsed':
+          span.setAttribute(IOAttributes.StreamingControlMatched, true)
+          span.setAttribute(IOAttributes.StreamingControlParsed, true)
+          span.setAttribute(IOAttributes.StreamingControlParserName, event.parserName)
+          span.setAttribute(IOAttributes.StreamingControlTokenType, event.tokenType)
+          if (event.callName)
+            span.setAttribute(IOAttributes.StreamingControlCallName, event.callName)
+          if (event.parameter)
+            span.setAttribute(IOAttributes.StreamingControlParameter, event.parameter)
+          span.addEvent(IOEvents.StreamingControlParsed, {
+            [IOAttributes.StreamingControlTokenType]: event.tokenType,
+            ...(event.parameter ? { [IOAttributes.StreamingControlParameter]: event.parameter } : {}),
+          })
+          break
+        case 'call-handler-count':
+          span.setAttribute(IOAttributes.StreamingControlHandlerCount, event.count)
+          break
+        case 'call-handler-missing':
+          if (context?.turnId && !context.remote) {
+            postRemoteCall({
+              type: 'turn-call',
+              fromInstanceId: instanceId,
+              turnId: context.turnId,
+              callName: event.callName,
+              ...(event.payload ? { payload: event.payload } : {}),
+            })
+          }
+          break
+        case 'call-handler-start':
+          span.addEvent(IOEvents.StreamingControlHandlerStart, {
+            [IOAttributes.StreamingControlCallName]: event.callName,
+          })
+          break
+        case 'call-handler-end':
+          span.addEvent(IOEvents.StreamingControlHandlerEnd, {
+            [IOAttributes.StreamingControlCallName]: event.callName,
+          })
+          break
+        case 'call-handler-error':
+          span.addEvent(IOEvents.StreamingControlHandlerError, {
+            [IOAttributes.StreamingControlCallName]: event.callName,
+            [IOAttributes.StreamingControlReason]: errorMessageFrom(event.error) ?? 'Unknown error',
+          })
+          break
+        case 'signal-handler-error':
+          span.addEvent(IOEvents.StreamingControlSignalHandlerError, {
+            [IOAttributes.StreamingControlReason]: errorMessageFrom(event.error) ?? 'Unknown error',
+            [IOAttributes.StreamingControlTokenType]: event.tokenType,
+          })
+          break
+      }
+    }
+
+    try {
+      return await controller.dispatchWith(special, {
+        ...context,
+        observer: observe,
+      })
+    }
+    finally {
+      span.end()
+    }
+  }
+
+  return {
+    dispatchWith,
+    beginTurn: controller.beginTurn,
+    completeTurn: controller.completeTurn,
+    cancelTurn: controller.cancelTurn,
+    match: controller.match,
+    on: controller.on,
+    renderManifestPrompt: controller.renderManifestPrompt,
+    onSignal: controller.onSignal,
+  }
+})
@@ -0,0 +1,231 @@
+import { createPinia, setActivePinia } from 'pinia'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { ref } from 'vue'
+
+import { useLlmStreamingControlStore } from '../../llm-streaming-control'
+import { useContextBridgeStore } from './context-bridge'
+
+type SparkNotifyReactionMock = (event: {
+  data?: {
+    id?: string
+  }
+}, options?: unknown) => Promise<string>
+
+const handleSparkNotifyWithReaction = vi.fn<SparkNotifyReactionMock>(async () => 'reaction text')
+
+function getLastSparkEventId() {
+  return handleSparkNotifyWithReaction.mock.calls.at(-1)?.[0]?.data?.id
+}
+
+vi.mock('../../character', () => ({
+  useCharacterOrchestratorStore: () => ({
+    handleSparkNotifyWithReaction,
+  }),
+}))
+
+vi.mock('../../chat', () => ({
+  useChatOrchestratorStore: () => ({}),
+}))
+
+vi.mock('../../chat/session-store', () => ({
+  useChatSessionStore: () => ({}),
+}))
+
+vi.mock('../../chat/stream-store', () => ({
+  useChatStreamStore: () => ({}),
+}))
+
+vi.mock('../../chat/context-store', () => ({
+  useChatContextStore: () => ({
+    ingestContextMessage: vi.fn(),
+  }),
+}))
+
+vi.mock('../../devtools/context-observability', () => ({
+  useContextObservabilityStore: () => ({
+    recordLifecycle: vi.fn(),
+  }),
+}))
+
+vi.mock('../../modules/consciousness', () => ({
+  useConsciousnessStore: () => ({
+    activeProvider: ref(undefined),
+    activeModel: ref(undefined),
+  }),
+}))
+
+vi.mock('../../providers', () => ({
+  useProvidersStore: () => ({}),
+}))
+
+vi.mock('./channel-server', () => ({
+  useModsServerChannelStore: () => ({
+    ensureConnected: vi.fn(async () => undefined),
+    send: vi.fn(),
+    onReconnected: vi.fn(() => () => undefined),
+    onContextUpdate: vi.fn(() => () => undefined),
+    onEvent: vi.fn(() => () => undefined),
+  }),
+}))
+
+describe('dispatchSparkNotifyPerformance', () => {
+  beforeEach(() => {
+    setActivePinia(createPinia())
+    handleSparkNotifyWithReaction.mockClear()
+  })
+
+  /**
+   * @example
+   * const resultPromise = store.dispatchSparkNotifyPerformance({ headline: 'x', fallbackResponseText: '', calls: [{ manifest: { name: 'plugin.action', prompt: 'Run it.' }, handler }] })
+   * await streamingControl.dispatchWith('<|CALL ["plugin.action"]|>', { turnId })
+   * await expect(resultPromise).resolves.toMatchObject({ type: 'called', name: 'plugin.action' })
+   */
+  it('resolves when a registered generic performance call is emitted', async () => {
+    const store = useContextBridgeStore()
+    store.setSparkNotifyHostRole('main')
+    const handler = vi.fn()
+    const streamingControl = useLlmStreamingControlStore()
+
+    const resultPromise = store.dispatchSparkNotifyPerformance({
+      headline: 'Plugin performance',
+      fallbackResponseText: '',
+      calls: [
+        {
+          manifest: {
+            name: 'plugin.action',
+            prompt: 'Run the plugin action when the model is ready.',
+            examples: [
+              '<|CALL ["plugin.action"]|>',
+            ],
+          },
+          handler,
+        },
+      ],
+      timeoutMs: 1000,
+    })
+
+    const sparkEventId = getLastSparkEventId()
+    expect(sparkEventId).toEqual(expect.any(String))
+
+    await streamingControl.dispatchWith('<|CALL ["plugin.action"]|>', {
+      turnId: `spark:${sparkEventId}`,
+    })
+
+    await expect(resultPromise).resolves.toEqual({
+      type: 'called',
+      name: 'plugin.action',
+      reaction: 'reaction text',
+    })
+    expect(handler).toHaveBeenCalledTimes(1)
+    expect(handleSparkNotifyWithReaction).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        messageOverride: expect.objectContaining({
+          appendSystemInstructions: [
+            expect.stringContaining('<|CALL ["plugin.action"]|>'),
+          ],
+        }),
+      }),
+    )
+  })
+
+  it('includes the CALL payload in the performance result', async () => {
+    const store = useContextBridgeStore()
+    store.setSparkNotifyHostRole('main')
+    const handler = vi.fn()
+    const streamingControl = useLlmStreamingControlStore()
+
+    const resultPromise = store.dispatchSparkNotifyPerformance({
+      headline: 'Plugin performance',
+      fallbackResponseText: '',
+      calls: [
+        {
+          manifest: {
+            name: 'plugin.action',
+            prompt: 'Run the plugin action when the model is ready.',
+          },
+          handler,
+        },
+      ],
+    })
+    const sparkEventId = getLastSparkEventId()
+    expect(sparkEventId).toEqual(expect.any(String))
+
+    await streamingControl.dispatchWith('<|CALL ["plugin.action", {"move":"Nf3"}]|>', {
+      turnId: `spark:${sparkEventId}`,
+    })
+
+    await expect(resultPromise).resolves.toEqual({
+      type: 'called',
+      name: 'plugin.action',
+      payload: { move: 'Nf3' },
+      reaction: 'reaction text',
+    })
+    expect(handler).toHaveBeenCalledWith({ move: 'Nf3' })
+  })
+
+  it('falls back when a client-side performance bridge request is not answered', async () => {
+    vi.useFakeTimers()
+    const store = useContextBridgeStore()
+    store.setSparkNotifyHostRole('client')
+
+    const resultPromise = store.dispatchSparkNotifyPerformance({
+      headline: 'Plugin performance',
+      fallbackResponseText: 'fallback text',
+      calls: [
+        {
+          manifest: {
+            name: 'plugin.action',
+            prompt: 'Run the plugin action when the model is ready.',
+          },
+          handler: vi.fn(),
+        },
+      ],
+      timeoutMs: 10,
+    })
+
+    await vi.advanceTimersByTimeAsync(10)
+
+    await expect(resultPromise).resolves.toEqual({
+      type: 'timeout',
+      reaction: 'fallback text',
+    })
+    vi.useRealTimers()
+  })
+
+  /**
+   * @example
+   * const result = store.dispatchSparkNotifyPerformance({ calls: [{ manifest: { name: 'plugin.action', prompt: 'Run it.' }, handler }] })
+   * streamingControl.completeTurn(turnId)
+   * await expect(result).resolves.toMatchObject({ type: 'completed' })
+   */
+  it('resolves with completed when the turn ends without CALL', async () => {
+    const store = useContextBridgeStore()
+    store.setSparkNotifyHostRole('main')
+    const handler = vi.fn()
+    const streamingControl = useLlmStreamingControlStore()
+
+    const resultPromise = store.dispatchSparkNotifyPerformance({
+      headline: 'Plugin performance',
+      fallbackResponseText: '',
+      calls: [
+        {
+          manifest: {
+            name: 'plugin.action',
+            prompt: 'Run the plugin action when the model is ready.',
+          },
+          handler,
+        },
+      ],
+    })
+    const sparkEventId = getLastSparkEventId()
+    expect(sparkEventId).toEqual(expect.any(String))
+    streamingControl.completeTurn(`spark:${sparkEventId}`)
+
+    await expect(resultPromise).resolves.toEqual({
+      type: 'completed',
+      reaction: 'reaction text',
+    })
+    expect(handler).not.toHaveBeenCalled()
+  })
+})
@@ -1,9 +1,10 @@
+import type { LlmStreamingControlCallManifest } from '@proj-airi/pipelines-audio'
 import type { WebSocketEventOf } from '@proj-airi/server-sdk'
 import type { ChatProvider } from '@xsai-ext/providers/utils'
 import type { UserMessage } from '@xsai/shared-chat'

 import type { ChatStreamEvent, ChatStreamEventContext, ContextMessage } from '../../../types/chat'
-import type { SparkNotifyReactionOptions } from './spark-notify-reaction'
+import type { SparkNotifyPerformanceResult, SparkNotifyReactionOptions } from './spark-notify-reaction'

 import { errorMessageFrom } from '@moeru/std'
 import { isStageTamagotchi, isStageWeb } from '@proj-airi/stage-shared'
@@ -21,6 +22,7 @@ import { useChatContextStore } from '../../chat/context-store'
 import { useChatSessionStore } from '../../chat/session-store'
 import { useChatStreamStore } from '../../chat/stream-store'
 import { useContextObservabilityStore } from '../../devtools/context-observability'
+import { useLlmStreamingControlStore } from '../../llm-streaming-control'
 import { useConsciousnessStore } from '../../modules/consciousness'
 import { useProvidersStore } from '../../providers'
 import { useModsServerChannelStore } from './channel-server'
@@ -57,6 +59,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
  const consciousnessStore = useConsciousnessStore()
  const providersStore = useProvidersStore()
  const { activeProvider, activeModel } = storeToRefs(consciousnessStore)
+  const streamingControl = useLlmStreamingControlStore()

  const { post: broadcastContext, data: incomingContext } = useBroadcastChannel<ContextMessage, ContextMessage>({ name: CONTEXT_CHANNEL_NAME })
  const { post: broadcastStreamEvent, data: incomingStreamEvent } = useBroadcastChannel<ChatStreamEvent, ChatStreamEvent>({ name: CHAT_STREAM_CHANNEL_NAME })
@@ -66,19 +69,24 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
      requestId: string
      fromInstanceId: string
      payload: SparkNotifyReactionOptions
+      performance?: {
+        callManifests: LlmStreamingControlCallManifest[]
+        timeoutMs?: number
+      }
    }
    | {
      type: 'response'
      requestId: string
      toInstanceId: string
      reaction: string
+      performance?: SparkNotifyPerformanceResult
    }
  const SPARK_NOTIFY_BRIDGE_CHANNEL_NAME = 'airi-spark-notify-bridge'
  const sparkNotifyBridgeInstanceId = `spark-notify-${nanoid()}`
  const sparkNotifyHostRole = ref<'main' | 'client'>('client')
  const sparkNotifyBridgeWaiters = new Map<string, {
-    resolve: (reaction: string) => void
-    timeout: ReturnType<typeof setTimeout>
+    resolve: (result: { reaction: string, performance?: SparkNotifyPerformanceResult }) => Promise<void> | void
+    timeout?: ReturnType<typeof setTimeout>
  }>()
  const { post: postSparkNotifyBridgeMessage, data: incomingSparkNotifyBridgeMessage } = useBroadcastChannel<SparkNotifyBridgeMessage, SparkNotifyBridgeMessage>({ name: SPARK_NOTIFY_BRIDGE_CHANNEL_NAME })

@@ -133,13 +141,30 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
    }
  }

-  async function handleSparkNotifyReactionLocal(options: SparkNotifyReactionOptions) {
+  function withStreamingCallPrompt(options: SparkNotifyReactionOptions, callPrompt: string): SparkNotifyReactionOptions {
+    if (!callPrompt) {
+      return options
+    }
+
+    return {
+      ...options,
+      messageOverride: {
+        ...options.messageOverride,
+        appendSystemInstructions: [
+          ...(options.messageOverride?.appendSystemInstructions ?? []),
+          callPrompt,
+        ],
+      },
+    }
+  }
+
+  async function handleSparkNotifyReactionLocal(options: SparkNotifyReactionOptions, identity?: { id?: string, eventId?: string }) {
    const event: WebSocketEventOf<'spark:notify'> = {
      type: 'spark:notify',
      source: options.source ?? 'plugin-module-host',
      data: {
-        id: nanoid(),
-        eventId: nanoid(),
+        id: identity?.id ?? nanoid(),
+        eventId: identity?.eventId ?? nanoid(),
        lane: options.lane,
        kind: options.kind ?? 'ping',
        urgency: options.urgency ?? 'immediate',
@@ -185,7 +210,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
      }, 5000)

      sparkNotifyBridgeWaiters.set(requestId, {
-        resolve: (reaction) => {
+        resolve: ({ reaction }) => {
          clearTimeout(timeout)
          resolve(reaction || options.fallbackResponseText)
        },
@@ -201,6 +226,154 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
    })
  }

+  async function handleSparkNotifyPerformanceLocal(options: SparkNotifyReactionOptions): Promise<SparkNotifyPerformanceResult> {
+    const calls = options.calls ?? []
+
+    if (calls.length === 0) {
+      const reaction = await handleSparkNotifyReactionLocal(options)
+      return {
+        type: 'completed',
+        reaction,
+      }
+    }
+
+    const sparkNotifyId = nanoid()
+    const turn = streamingControl.beginTurn({ turnId: `spark:${sparkNotifyId}` })
+
+    let latestReaction = ''
+    let reactionPromise: Promise<string> | undefined
+    let dispose: (() => void) | undefined
+
+    const calledPromise = new Promise<SparkNotifyPerformanceResult>((resolve) => {
+      const disposers = calls.map(call => turn.on(call.manifest, async (payload) => {
+        await call.handler(payload)
+        const reaction = await (reactionPromise ?? Promise.resolve(latestReaction || options.fallbackResponseText))
+        resolve({
+          type: 'called',
+          name: call.manifest.name,
+          payload,
+          reaction,
+        })
+      }))
+      dispose = () => {
+        for (const item of disposers) {
+          item()
+        }
+      }
+    })
+
+    reactionPromise = handleSparkNotifyReactionLocal(withStreamingCallPrompt(
+      options,
+      turn.renderManifestPrompt(),
+    ), { id: sparkNotifyId })
+      .then((reaction) => {
+        latestReaction = reaction
+        return reaction
+      })
+      .catch(() => {
+        latestReaction = options.fallbackResponseText
+        return options.fallbackResponseText
+      })
+
+    const turnDonePromise = turn.done.then(async (result): Promise<SparkNotifyPerformanceResult> => {
+      const reaction = await (reactionPromise ?? Promise.resolve(latestReaction || options.fallbackResponseText))
+      return {
+        type: result.type === 'cancelled' ? 'cancelled' : 'completed',
+        reaction: reaction || options.fallbackResponseText,
+      }
+    })
+
+    const result = await Promise.race([calledPromise, turnDonePromise])
+    dispose?.()
+    return result
+  }
+
+  async function dispatchSparkNotifyPerformance(options: SparkNotifyReactionOptions): Promise<SparkNotifyPerformanceResult> {
+    const calls = options.calls ?? []
+
+    if (sparkNotifyHostRole.value === 'main') {
+      return await handleSparkNotifyPerformanceLocal(options)
+    }
+
+    if (calls.length === 0) {
+      const reaction = await dispatchSparkNotifyReaction(options)
+      return {
+        type: 'completed',
+        reaction,
+      }
+    }
+
+    const requestId = nanoid()
+    return await new Promise<SparkNotifyPerformanceResult>((resolve) => {
+      const timeout = setTimeout(() => {
+        sparkNotifyBridgeWaiters.delete(requestId)
+        resolve(createFallbackPerformanceResult(options, 'timeout'))
+      }, Math.max(1, options.timeoutMs ?? 5000))
+
+      sparkNotifyBridgeWaiters.set(requestId, {
+        resolve: async ({ reaction, performance }) => {
+          clearTimeout(timeout)
+          if (performance?.type === 'called' && performance.name) {
+            await findPerformanceCall(options, performance.name)?.handler(performance.payload)
+          }
+
+          resolve(performance ?? createFallbackPerformanceResult(options, 'completed', reaction))
+        },
+        timeout,
+      })
+
+      const { calls: _calls, timeoutMs: _timeoutMs, ...payload } = options
+      postSparkNotifyBridgeMessage({
+        type: 'request',
+        requestId,
+        fromInstanceId: sparkNotifyBridgeInstanceId,
+        payload,
+        performance: {
+          callManifests: calls.map(call => call.manifest),
+          timeoutMs: options.timeoutMs,
+        },
+      })
+    })
+  }
+
+  function createFallbackPerformanceResult(
+    options: SparkNotifyReactionOptions,
+    type: Extract<SparkNotifyPerformanceResult['type'], 'completed' | 'timeout'>,
+    reaction?: string,
+  ): SparkNotifyPerformanceResult {
+    return {
+      type,
+      reaction: reaction || options.fallbackResponseText,
+    }
+  }
+
+  function findPerformanceCall(options: SparkNotifyReactionOptions, name: string) {
+    return options.calls?.find(call => call.manifest.name === name)
+  }
+
+  function withContextBridgeLock<T>(key: string, callback: () => Promise<T>) {
+    if (typeof navigator !== 'undefined' && 'locks' in navigator && typeof navigator.locks.request === 'function') {
+      return navigator.locks.request(key, callback)
+    }
+    return callback()
+  }
+
+  async function withContextBridgeExclusiveLock<T>(key: string, callback: () => Promise<T>) {
+    if (typeof navigator !== 'undefined' && 'locks' in navigator && typeof navigator.locks.request === 'function') {
+      // BroadcastChannel delivers the same bridge request to every Stage window.
+      // `ifAvailable` makes non-owning windows skip instead of queueing and replaying
+      // the same spark reaction after the first window finishes.
+      return await navigator.locks.request(key, { ifAvailable: true }, async (lock) => {
+        if (!lock) {
+          return undefined
+        }
+        return await callback()
+      })
+    }
+
+    return await callback()
+  }
+
  async function initialize() {
    await mutex.acquire()

@@ -281,12 +454,25 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
            return
          }

-          const reaction = await handleSparkNotifyReactionLocal(event.payload)
-          postSparkNotifyBridgeMessage({
-            type: 'response',
-            requestId: event.requestId,
-            toInstanceId: event.fromInstanceId,
-            reaction,
+          await withContextBridgeExclusiveLock(`context-bridge:spark-notify:${event.requestId}`, async () => {
+            const performance = event.performance?.callManifests.length
+              ? await handleSparkNotifyPerformanceLocal({
+                  ...event.payload,
+                  calls: event.performance.callManifests.map(manifest => ({
+                    manifest,
+                    handler: async () => undefined,
+                  })),
+                  timeoutMs: event.performance.timeoutMs,
+                })
+              : undefined
+            const reaction = performance?.reaction ?? await handleSparkNotifyReactionLocal(event.payload)
+            postSparkNotifyBridgeMessage({
+              type: 'response',
+              requestId: event.requestId,
+              toInstanceId: event.fromInstanceId,
+              reaction,
+              ...(performance ? { performance } : {}),
+            })
          })
          return
        }
@@ -302,7 +488,10 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
          }

          sparkNotifyBridgeWaiters.delete(event.requestId)
-          waiter.resolve(event.reaction)
+          await waiter.resolve({
+            reaction: event.reaction,
+            performance: event.performance,
+          })
        }
      })
      disposeHookFns.value.push(stopSparkNotifyBridgeWatch)
@@ -367,13 +556,6 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
        })
      }))

-      function withContextBridgeLock<T>(key: string, callback: () => Promise<T>) {
-        if (typeof navigator !== 'undefined' && 'locks' in navigator && typeof navigator.locks.request === 'function') {
-          return navigator.locks.request(key, callback)
-        }
-        return callback()
-      }
-
      disposeHookFns.value.push(serverChannelStore.onEvent('input:text', async (event) => {
        const {
          text,
@@ -720,7 +902,8 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
      remoteStreamGuard = null

      for (const [requestId, waiter] of sparkNotifyBridgeWaiters) {
-        clearTimeout(waiter.timeout)
+        if (waiter.timeout)
+          clearTimeout(waiter.timeout)
        sparkNotifyBridgeWaiters.delete(requestId)
      }
    }
@@ -735,6 +918,7 @@ export const useContextBridgeStore = defineStore('mods:api:context-bridge', () =
    initialize,
    dispose,
    dispatchSparkNotifyReaction,
+    dispatchSparkNotifyPerformance,
    setSparkNotifyHostRole,
  }
 })
@@ -1,4 +1,5 @@
 import type { SparkNotifyResponseControl } from '@proj-airi/core-agent/agents/spark-notify'
+import type { LlmStreamingControlCallManifest } from '@proj-airi/pipelines-audio'
 import type { WebSocketEventOf } from '@proj-airi/server-sdk'

 import { array, boolean, finite, looseObject, nonEmpty, number, optional, picklist, pipe, record, string, trim, unknown } from 'valibot'
@@ -6,6 +7,32 @@ import { array, boolean, finite, looseObject, nonEmpty, number, optional, pickli
 type SparkNotifyProtocolEvent = WebSocketEventOf<'spark:notify'>
 type SparkNotifyProtocolData = SparkNotifyProtocolEvent['data']

+export type SparkNotifyReactionCallHandler = (payload?: Record<string, unknown>) => Promise<void> | void
+
+/**
+ * Registered performance call available during one spark notify reaction.
+ */
+export interface SparkNotifyReactionCallRegistration {
+  /** Prompt manifest rendered into the model instructions and used as the dispatch key. */
+  manifest: LlmStreamingControlCallManifest
+  /** Runtime callback executed when the matching CALL token is emitted. */
+  handler: SparkNotifyReactionCallHandler
+}
+
+/**
+ * Result returned by the call-aware spark notify reaction bridge.
+ */
+export interface SparkNotifyPerformanceResult {
+  /** Text reaction produced by the existing spark notify path. */
+  reaction: string
+  /** Terminal state for the performance request. */
+  type: 'called' | 'completed' | 'timeout' | 'cancelled'
+  /** Name of the generic performance call that resolved the request, when applicable. */
+  name?: string
+  /** Payload emitted by the matching CALL token, when applicable. */
+  payload?: Record<string, unknown>
+}
+
 /**
 * Caller-facing request used by the context bridge to turn one spark notification into a reaction string.
 */
@@ -47,6 +74,14 @@ export interface SparkNotifyReactionOptions
   * @default 'plugin-module-host'
   */
  source?: SparkNotifyProtocolEvent['source']
+  /** Generic performance calls allowed during this spark notify reaction request. */
+  calls?: SparkNotifyReactionCallRegistration[]
+  /**
+   * Maximum time to wait for a registered performance call after spark notify starts.
+   *
+   * @default 5000
+   */
+  timeoutMs?: number
 }

 export const sparkNotifyReactionOptionsSchema = looseObject({