From 4622b99e2a13714e37cff9420371cc17fc96e3a8 Mon Sep 17 00:00:00 2001 From: Neko Ayaka Date: Mon, 18 May 2026 01:20:21 +0800 Subject: [PATCH] refactor(pipelines-audio): better token processing pipeline, ACT, DELAY, CALL, unified --- .../components/extension-ui-host.vue | 1 + .../composables/use-bridge-spark.test.ts | 76 +++++ .../composables/use-bridge-spark.ts | 37 ++- apps/stage-tamagotchi/vitest.config.ts | 5 + packages/i18n/src/locales/en/base.yaml | 29 +- packages/i18n/src/locales/zh-Hans/base.yaml | 8 +- packages/i18n/src/locales/zh-Hant/base.yaml | 8 +- packages/pipelines-audio/src/eventa.ts | 10 + packages/pipelines-audio/src/index.ts | 2 + .../src/llm-streaming-control/controller.ts | 264 +++++++++++++++ .../src/llm-streaming-control/index.test.ts | 313 ++++++++++++++++++ .../src/llm-streaming-control/index.ts | 24 ++ .../src/llm-streaming-control/parsers/act.ts | 51 +++ .../parsers/call.test.ts | 25 ++ .../src/llm-streaming-control/parsers/call.ts | 113 +++++++ .../llm-streaming-control/parsers/delay.ts | 43 +++ .../llm-streaming-control/parsers/index.ts | 3 + .../llm-streaming-control/payloads.test.ts | 48 +++ .../src/llm-streaming-control/payloads.ts | 100 ++++++ .../src/llm-streaming-control/types.ts | 192 +++++++++++ .../src/processors/tts-chunker.ts | 3 +- .../src/speech-pipeline.test.ts | 127 ++++++- .../pipelines-audio/src/speech-pipeline.ts | 101 +++++- packages/pipelines-audio/src/timeline.ts | 102 ++++++ packages/pipelines-audio/src/types.ts | 12 +- packages/stage-ui/src/services/speech/bus.ts | 4 + .../src/services/speech/pipeline-runtime.ts | 17 +- .../stage-ui/src/stores/character/index.ts | 1 + .../stage-ui/src/stores/chat.contract.test.ts | 36 +- packages/stage-ui/src/stores/chat.ts | 16 + .../src/stores/llm-streaming-control.test.ts | 98 ++++++ .../src/stores/llm-streaming-control.ts | 150 +++++++++ .../context-bridge-performance-call.test.ts | 231 +++++++++++++ .../src/stores/mods/api/context-bridge.ts | 228 +++++++++++-- .../stores/mods/api/spark-notify-reaction.ts | 35 ++ 35 files changed, 2454 insertions(+), 59 deletions(-) create mode 100644 packages/pipelines-audio/src/llm-streaming-control/controller.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/index.test.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/index.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/parsers/act.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/parsers/call.test.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/parsers/call.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/parsers/delay.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/parsers/index.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/payloads.test.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/payloads.ts create mode 100644 packages/pipelines-audio/src/llm-streaming-control/types.ts create mode 100644 packages/pipelines-audio/src/timeline.ts create mode 100644 packages/stage-ui/src/stores/llm-streaming-control.test.ts create mode 100644 packages/stage-ui/src/stores/llm-streaming-control.ts create mode 100644 packages/stage-ui/src/stores/mods/api/context-bridge-performance-call.test.ts diff --git a/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/components/extension-ui-host.vue b/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/components/extension-ui-host.vue index 1af711bdd..4c8693fc6 100644 --- a/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/components/extension-ui-host.vue +++ b/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/components/extension-ui-host.vue @@ -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) { diff --git a/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.test.ts b/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.test.ts index cd5e33493..a636274d2 100644 --- a/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.test.ts +++ b/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.test.ts @@ -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', + }, + }), + })) + }) }) diff --git a/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.ts b/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.ts index a5e24b51d..65cfa2ca1 100644 --- a/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.ts +++ b/apps/stage-tamagotchi/src/renderer/widgets/extension-ui/composables/use-bridge-spark.ts @@ -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 + dispatchSparkNotifyPerformance?: (options: SparkNotifyReactionOptions) => Promise emit: (event: typeof widgetsIframeBroadcastEvent, payload: Record) => 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) { 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, + }, + } + : {}), }, }) diff --git a/apps/stage-tamagotchi/vitest.config.ts b/apps/stage-tamagotchi/vitest.config.ts index 075601487..e7abe03e3 100644 --- a/apps/stage-tamagotchi/vitest.config.ts +++ b/apps/stage-tamagotchi/vitest.config.ts @@ -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'], diff --git a/packages/i18n/src/locales/en/base.yaml b/packages/i18n/src/locales/en/base.yaml index bebf5bc92..30f75b196 100644 --- a/packages/i18n/src/locales/en/base.yaml +++ b/packages/i18n/src/locales/en/base.yaml @@ -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: , "motion": '} + ACT {'{"emotion": <{ "name": emotion, "intensity": 0-1 } or emotion string>, "motion": }'} 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: diff --git a/packages/i18n/src/locales/zh-Hans/base.yaml b/packages/i18n/src/locales/zh-Hans/base.yaml index 327bfd13e..e3ff3bb3a 100644 --- a/packages/i18n/src/locales/zh-Hans/base.yaml +++ b/packages/i18n/src/locales/zh-Hans/base.yaml @@ -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: , "motion": '} 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": '}'} 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: diff --git a/packages/i18n/src/locales/zh-Hant/base.yaml b/packages/i18n/src/locales/zh-Hant/base.yaml index 325eb83ae..b9151b67f 100644 --- a/packages/i18n/src/locales/zh-Hant/base.yaml +++ b/packages/i18n/src/locales/zh-Hant/base.yaml @@ -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: , "motion": '} 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": '}'} 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: diff --git a/packages/pipelines-audio/src/eventa.ts b/packages/pipelines-audio/src/eventa.ts index cd634f5ea..634b7cc6a 100644 --- a/packages/pipelines-audio/src/eventa.ts +++ b/packages/pipelines-audio/src/eventa.ts @@ -12,16 +12,23 @@ import { defineEventa } from '@moeru/eventa' export const speechSegmentEvent = defineEventa('proj-airi:pipelines:output:speech:segment') export const speechSpecialEvent = defineEventa('proj-airi:pipelines:output:speech:special') + export const speechTtsRequestEvent = defineEventa('proj-airi:pipelines:output:speech:tts-request') export const speechTtsResultEvent = defineEventa>('proj-airi:pipelines:output:speech:tts-result') + export const speechPlaybackStartEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-start') export const speechPlaybackEndEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-end') export const speechPlaybackInterruptEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-interrupt') export const speechPlaybackRejectEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-reject') + export const speechIntentStartEvent = defineEventa('proj-airi:pipelines:output:speech:intent-start') export const speechIntentEndEvent = defineEventa('proj-airi:pipelines:output:speech:intent-end') export const speechIntentCancelEvent = defineEventa<{ intentId: string, reason?: string }>('proj-airi:pipelines:output:speech:intent-cancel') +export const speechTurnStartEvent = defineEventa('proj-airi:pipelines:output:speech:turn-start') +export const speechTurnEndEvent = defineEventa('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 diff --git a/packages/pipelines-audio/src/index.ts b/packages/pipelines-audio/src/index.ts index e39356c99..fbb9ed6f8 100644 --- a/packages/pipelines-audio/src/index.ts +++ b/packages/pipelines-audio/src/index.ts @@ -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' diff --git a/packages/pipelines-audio/src/llm-streaming-control/controller.ts b/packages/pipelines-audio/src/llm-streaming-control/controller.ts new file mode 100644 index 000000000..ca9145556 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/controller.ts @@ -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> + callManifests: Map + settle: (result: LlmStreamingControlTurnDone) => void + done: Promise +} + +/** + * 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>() + const callManifests = new Map() + const turns = new Map() + const signalHandlers = new Set() + 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() + 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((resolve) => { + settle = (result) => { + if (settled) + return + settled = true + resolve(result) + } + }) + const turn = { + handlers: new Map>(), + callManifests: new Map(), + 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, + manifest: LlmStreamingControlCallManifest, + handler: LlmStreamingControlCallHandler, + ) { + const normalizedName = manifest.name.trim() + const normalizedPrompt = manifest.prompt.trim() + if (!normalizedName || !normalizedPrompt) { + return () => undefined + } + + turn.callManifests.set(normalizedName, { ...manifest, name: normalizedName, prompt: normalizedPrompt }) + const registeredHandlers = turn.handlers.get(normalizedName) ?? new Set() + registeredHandlers.add(handler) + turn.handlers.set(normalizedName, registeredHandlers) + + return () => { + registeredHandlers.delete(handler) + if (registeredHandlers.size === 0) { + turn.handlers.delete(normalizedName) + turn.callManifests.delete(normalizedName) + } + } + } +} diff --git a/packages/pipelines-audio/src/llm-streaming-control/index.test.ts b/packages/pipelines-audio/src/llm-streaming-control/index.test.ts new file mode 100644 index 000000000..25feee7c1 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/index.test.ts @@ -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((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() + }) +}) diff --git a/packages/pipelines-audio/src/llm-streaming-control/index.ts b/packages/pipelines-audio/src/llm-streaming-control/index.ts new file mode 100644 index 000000000..76a06d73a --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/index.ts @@ -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' diff --git a/packages/pipelines-audio/src/llm-streaming-control/parsers/act.ts b/packages/pipelines-audio/src/llm-streaming-control/parsers/act.ts new file mode 100644 index 000000000..d81f34327 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/parsers/act.ts @@ -0,0 +1,51 @@ +import type { LlmStreamingControlParser, LlmStreamingControlTokenAct } from '../types' + +const actTokenPrefix = '<|ACT ' +const markerSuffix = '|>' + +function isPlainObject(value: unknown): value is Record { + 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 { + 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, + } + }, + } +} diff --git a/packages/pipelines-audio/src/llm-streaming-control/parsers/call.test.ts b/packages/pipelines-audio/src/llm-streaming-control/parsers/call.test.ts new file mode 100644 index 000000000..397b8e8ad --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/parsers/call.test.ts @@ -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"]|>') + }) +}) diff --git a/packages/pipelines-audio/src/llm-streaming-control/parsers/call.ts b/packages/pipelines-audio/src/llm-streaming-control/parsers/call.ts new file mode 100644 index 000000000..cf37db4d0 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/parsers/call.ts @@ -0,0 +1,113 @@ +import type { + LlmStreamingControlCallManifest, + LlmStreamingControlParser, + LlmStreamingControlTokenCall, +} from '../types' + +const callTokenPrefix = '<|CALL ' +const markerSuffix = '|>' + +function isPlainObject(value: unknown): value is Record { + 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 { + 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, + } + }, + } +} diff --git a/packages/pipelines-audio/src/llm-streaming-control/parsers/delay.ts b/packages/pipelines-audio/src/llm-streaming-control/parsers/delay.ts new file mode 100644 index 000000000..ca11deaf0 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/parsers/delay.ts @@ -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 { + 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, + } + }, + } +} diff --git a/packages/pipelines-audio/src/llm-streaming-control/parsers/index.ts b/packages/pipelines-audio/src/llm-streaming-control/parsers/index.ts new file mode 100644 index 000000000..153d2af71 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/parsers/index.ts @@ -0,0 +1,3 @@ +export { tokenAct } from './act' +export { tokenCall } from './call' +export { tokenDelay } from './delay' diff --git a/packages/pipelines-audio/src/llm-streaming-control/payloads.test.ts b/packages/pipelines-audio/src/llm-streaming-control/payloads.test.ts new file mode 100644 index 000000000..00d43ea74 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/payloads.test.ts @@ -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 }, + }) + }) +}) diff --git a/packages/pipelines-audio/src/llm-streaming-control/payloads.ts b/packages/pipelines-audio/src/llm-streaming-control/payloads.ts new file mode 100644 index 000000000..3143eed14 --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/payloads.ts @@ -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): 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 +} diff --git a/packages/pipelines-audio/src/llm-streaming-control/types.ts b/packages/pipelines-audio/src/llm-streaming-control/types.ts new file mode 100644 index 000000000..e77ae8aea --- /dev/null +++ b/packages/pipelines-audio/src/llm-streaming-control/types.ts @@ -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 } + | { 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 = Record, +> = ( + payload: TPayload | undefined, + context: LlmStreamingControlCallContext, +) => void | Promise + +export type LlmStreamingControlSignalHandler = ( + signal: LlmStreamingControlSignal, + context: LlmStreamingControlSignalContext, +) => void | Promise + +/** + * 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: = Record>( + manifest: LlmStreamingControlCallManifest, + handler: LlmStreamingControlCallHandler, + ) => () => void + renderManifestPrompt: () => string + complete: () => void + cancel: () => void + done: Promise +} + +/** + * Parser for one LLM streaming control syntax. + * + * @param TParsed Parsed token payload owned by the matching control. + */ +export interface LlmStreamingControlParser { + /** 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) => Promise + /** + * Registers a callback for one plugin-owned CALL name. + * + * Returns: + * - A disposer that unregisters the callback + */ + on: = Record>( + manifest: LlmStreamingControlCallManifest, + handler: LlmStreamingControlCallHandler, + ) => () => 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[] +} + +/** + * Parsed CALL token payload. + * + * @param TPayload Payload object shape expected by one CALL name. + */ +export interface LlmStreamingControlTokenCall< + TPayload extends Record = Record, +> { + /** 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 +} + +/** + * Parsed DELAY token payload. + */ +export interface LlmStreamingControlTokenDelay { + /** Parsed signal kind. */ + type: 'delay' + /** Delay length in seconds. */ + seconds: number +} + +export type LlmStreamingControlSignal + = | LlmStreamingControlTokenAct + | LlmStreamingControlTokenCall + | LlmStreamingControlTokenDelay diff --git a/packages/pipelines-audio/src/processors/tts-chunker.ts b/packages/pipelines-audio/src/processors/tts-chunker.ts index 3957cd2ec..41cce54f6 100644 --- a/packages/pipelines-audio/src/processors/tts-chunker.ts +++ b/packages/pipelines-audio/src/processors/tts-chunker.ts @@ -384,7 +384,7 @@ export function processNarrative(text: string, options?: TtsInputChunkOptions): export function createTtsSegmentStream( tokens: ReadableStream, - meta: { streamId: string, intentId: string }, + meta: { streamId: string, intentId: string, turnId?: string }, options?: TtsInputChunkOptions, ) { const { stream, write, close, error } = createPushStream() @@ -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)}`, diff --git a/packages/pipelines-audio/src/speech-pipeline.test.ts b/packages/pipelines-audio/src/speech-pipeline.test.ts index c54218c63..a6af1e348 100644 --- a/packages/pipelines-audio/src/speech-pipeline.test.ts +++ b/packages/pipelines-audio/src/speech-pipeline.test.ts @@ -51,20 +51,30 @@ function createSegmenter(texts: string[]) { } } -function createPlaybackSpy() { +function createPlaybackSpy(options?: { autoEnd?: boolean }) { const scheduled: Array> = [] + const endListeners: Array<(event: { item: PlaybackItem, endedAt: number }) => void> = [] + const autoEnd = options?.autoEnd ?? true return { scheduled, + end(item: PlaybackItem) { + for (const listener of endListeners) + listener({ item, endedAt: Date.now() }) + }, playback: { schedule(item: PlaybackItem) { 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, 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({ + ttsMaxConcurrent: 2, + segmenter: (_tokens, meta) => { + return new ReadableStream({ + 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({ + ttsMaxConcurrent: 2, + segmenter: (_tokens, meta) => { + return new ReadableStream({ + 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']) + }) }) diff --git a/packages/pipelines-audio/src/speech-pipeline.ts b/packages/pipelines-audio/src/speech-pipeline.ts index 0627be8a7..8352967ab 100644 --- a/packages/pipelines-audio/src/speech-pipeline.ts +++ b/packages/pipelines-audio/src/speech-pipeline.ts @@ -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 { tts: (request: TtsRequest, signal: AbortSignal) => Promise @@ -40,10 +41,11 @@ export interface SpeechPipelineOptions { } logger?: LoggerLike priority?: ReturnType - segmenter?: (tokens: ReadableStream, meta: { streamId: string, intentId: string }) => ReadableStream + segmenter?: (tokens: ReadableStream, meta: { streamId: string, intentId: string, turnId?: string }) => ReadableStream } interface IntentState { + turnId?: string intentId: string streamId: string priority: number @@ -66,15 +68,42 @@ export function createSpeechPipeline(options: SpeechPipelineOptions() const pending: IntentState[] = [] let activeIntent: IntentState | null = null + const playbackWaiters = new Map 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) { + return new Promise((resolve) => { + playbackWaiters.set(item.id, resolve) + options.playback.schedule(item) + }) + } function enqueueIntent(intent: IntentState) { pending.push(intent) @@ -90,22 +119,67 @@ export function createSpeechPipeline(options: SpeechPipelineOptions | null>() const inFlightTasks = new Set>() 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) { + 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(options: SpeechPipelineOptions = { + turnId: request.turnId, streamId: request.streamId, intentId: request.intentId, segmentId: request.segmentId, @@ -191,11 +266,12 @@ export function createSpeechPipeline(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions(options: SpeechPipelineOptions number + sleep: (ms: number) => Promise +} + +export interface TimelineRunContext { + clock: TimelineClock + signal: AbortSignal +} + +export interface TimelineItem { + id?: string + track: string + run: (context: TimelineRunContext) => Promise | void +} + +export interface TimelineItemHandle { + id: string + track: string + done: Promise + 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>() + + 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, + } +} diff --git a/packages/pipelines-audio/src/types.ts b/packages/pipelines-audio/src/types.ts index 118baecd9..758ff15bd 100644 --- a/packages/pipelines-audio/src/types.ts +++ b/packages/pipelines-audio/src/types.ts @@ -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 { + turnId?: string streamId: string intentId: string segmentId: string @@ -47,6 +51,7 @@ export interface TtsResult { export interface PlaybackItem { id: string + turnId?: string streamId: string intentId: string segmentId: string @@ -83,6 +88,7 @@ export interface PlaybackRejectEvent { 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 { onPlaybackReject: (event: PlaybackRejectEvent) => 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 { diff --git a/packages/stage-ui/src/services/speech/bus.ts b/packages/stage-ui/src/services/speech/bus.ts index 74c4d1800..769d7b0c3 100644 --- a/packages/stage-ui/src/services/speech/bus.ts +++ b/packages/stage-ui/src/services/speech/bus.ts @@ -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 diff --git a/packages/stage-ui/src/services/speech/pipeline-runtime.ts b/packages/stage-ui/src/services/speech/pipeline-runtime.ts index be2a75af0..5efff3cbf 100644 --- a/packages/stage-ui/src/services/speech/pipeline-runtime.ts +++ b/packages/stage-ui/src/services/speech/pipeline-runtime.ts @@ -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, diff --git a/packages/stage-ui/src/stores/character/index.ts b/packages/stage-ui/src/stores/character/index.ts index b6f34c3e2..2187086d4 100644 --- a/packages/stage-ui/src/stores/character/index.ts +++ b/packages/stage-ui/src/stores/character/index.ts @@ -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', diff --git a/packages/stage-ui/src/stores/chat.contract.test.ts b/packages/stage-ui/src/stores/chat.contract.test.ts index 13bee6b05..0b1da3cf5 100644 --- a/packages/stage-ui/src/stores/chat.contract.test.ts +++ b/packages/stage-ui/src/stores/chat.contract.test.ts @@ -80,12 +80,17 @@ vi.mock('../composables', () => ({ })) vi.mock('../composables/llm-marker-parser', () => ({ - useLlmmarkerParser: (options: { onLiteral?: (literal: string) => Promise, onEnd?: (fullText: string) => Promise }) => { + useLlmmarkerParser: (options: { onLiteral?: (literal: string) => Promise, onSpecial?: (special: string) => Promise, onEnd?: (fullText: string) => Promise }) => { 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 diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts index 925027128..2a94ba347 100644 --- a/packages/stage-ui/src/stores/chat.ts +++ b/packages/stage-ui/src/stores/chat.ts @@ -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) { diff --git a/packages/stage-ui/src/stores/llm-streaming-control.test.ts b/packages/stage-ui/src/stores/llm-streaming-control.test.ts new file mode 100644 index 000000000..570bf817e --- /dev/null +++ b/packages/stage-ui/src/stores/llm-streaming-control.test.ts @@ -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) + }) +}) diff --git a/packages/stage-ui/src/stores/llm-streaming-control.ts b/packages/stage-ui/src/stores/llm-streaming-control.ts new file mode 100644 index 000000000..cb69a154a --- /dev/null +++ b/packages/stage-ui/src/stores/llm-streaming-control.ts @@ -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 +} + +export const useLlmStreamingControlStore = defineStore('llm-streaming-control', () => { + const controller = createStreamingControlParser() + const instanceId = `streaming-control-${nanoid()}` + + const { post: postRemoteCall, data: incomingRemoteCall } = useBroadcastChannel({ 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) { + 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, + } +}) diff --git a/packages/stage-ui/src/stores/mods/api/context-bridge-performance-call.test.ts b/packages/stage-ui/src/stores/mods/api/context-bridge-performance-call.test.ts new file mode 100644 index 000000000..0ca2b8387 --- /dev/null +++ b/packages/stage-ui/src/stores/mods/api/context-bridge-performance-call.test.ts @@ -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 + +const handleSparkNotifyWithReaction = vi.fn(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() + }) +}) diff --git a/packages/stage-ui/src/stores/mods/api/context-bridge.ts b/packages/stage-ui/src/stores/mods/api/context-bridge.ts index ff89e303d..c4b7e86b4 100644 --- a/packages/stage-ui/src/stores/mods/api/context-bridge.ts +++ b/packages/stage-ui/src/stores/mods/api/context-bridge.ts @@ -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({ name: CONTEXT_CHANNEL_NAME }) const { post: broadcastStreamEvent, data: incomingStreamEvent } = useBroadcastChannel({ 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 void - timeout: ReturnType + resolve: (result: { reaction: string, performance?: SparkNotifyPerformanceResult }) => Promise | void + timeout?: ReturnType }>() const { post: postSparkNotifyBridgeMessage, data: incomingSparkNotifyBridgeMessage } = useBroadcastChannel({ 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 { + 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 | undefined + let dispose: (() => void) | undefined + + const calledPromise = new Promise((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 => { + 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 { + 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((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, + 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(key: string, callback: () => Promise) { + if (typeof navigator !== 'undefined' && 'locks' in navigator && typeof navigator.locks.request === 'function') { + return navigator.locks.request(key, callback) + } + return callback() + } + + async function withContextBridgeExclusiveLock(key: string, callback: () => Promise) { + 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(key: string, callback: () => Promise) { - 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, } }) diff --git a/packages/stage-ui/src/stores/mods/api/spark-notify-reaction.ts b/packages/stage-ui/src/stores/mods/api/spark-notify-reaction.ts index 50d5e54a1..a7d9217dc 100644 --- a/packages/stage-ui/src/stores/mods/api/spark-notify-reaction.ts +++ b/packages/stage-ui/src/stores/mods/api/spark-notify-reaction.ts @@ -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) => Promise | 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 +} + /** * 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({