From fe1c2198ac5ca7742e4a0e516521817abc710661 Mon Sep 17 00:00:00 2001 From: Dallin Romney Date: Fri, 7 Aug 2026 16:05:32 +0800 Subject: [PATCH] feat(plugin-sdk): open Gateway-managed Talk sessions --- docs/plugins/sdk-runtime.md | 20 ++ src/gateway/talk-plugin-session.test.ts | 149 ++++++++++++++ src/gateway/talk-plugin-session.ts | 190 ++++++++++++++++++ src/plugin-sdk/realtime-voice.ts | 6 + .../test-helpers/plugin-runtime-mock.test.ts | 6 + .../test-helpers/plugin-runtime-mock.ts | 3 + src/plugins/runtime/index.test.ts | 6 + src/plugins/runtime/index.ts | 13 ++ src/plugins/runtime/types.ts | 5 + src/talk/plugin-session.ts | 46 +++++ 10 files changed, 444 insertions(+) create mode 100644 src/gateway/talk-plugin-session.test.ts create mode 100644 src/gateway/talk-plugin-session.ts create mode 100644 src/talk/plugin-session.ts diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index 66e01cd15794..9de5dd75aceb 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -406,6 +406,26 @@ snapshots; OpenClaw owns all persistence and lifecycle coordination. `details`, retry metadata, and the Gateway error code for recovery flows. Use `isAvailable()` before choosing this path from tools that can also run in standalone agent processes. + + + Open a realtime voice conversation using the Gateway's configured Talk provider and agent. + OpenClaw handles turn timing, interruptions, and agent tool use; the plugin sends microphone + PCM and renders the returned events. + + ```typescript + const session = await api.runtime.talk.openSession({ + sessionKey: "agent:main:avatar", + onEvent: (event) => renderVoiceEvent(event), + }); + + session.sendAudio(pcm16le24kMono); + session.cancelOutput("barge-in"); + session.close(); + ``` + + This method is available to trusted plugin request routes that declare the + `gatewayMethodDispatch` contract. Output audio is 24 kHz mono PCM16. + Launch and manage background subagent runs. diff --git a/src/gateway/talk-plugin-session.test.ts b/src/gateway/talk-plugin-session.test.ts new file mode 100644 index 000000000000..66c40c9d2808 --- /dev/null +++ b/src/gateway/talk-plugin-session.test.ts @@ -0,0 +1,149 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + scope: vi.fn(), + createSession: vi.fn(), + sendAudio: vi.fn(), + cancelTurn: vi.fn(), + stopSession: vi.fn(), + warn: vi.fn(), +})); + +vi.mock("../plugins/runtime/gateway-request-scope.js", () => ({ + getPluginRuntimeGatewayRequestScope: mocks.scope, +})); +vi.mock("./talk-realtime-session-create.js", () => ({ + createGatewayRealtimeTalkSession: mocks.createSession, +})); +vi.mock("./talk-realtime-relay.js", () => ({ + sendTalkRealtimeRelayAudio: mocks.sendAudio, + cancelTalkRealtimeRelayTurn: mocks.cancelTurn, + stopTalkRealtimeRelaySession: mocks.stopSession, +})); + +import { openPluginTalkSession } from "./talk-plugin-session.js"; + +describe("plugin Talk session", () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.scope.mockReturnValue({ + pluginId: "avatar", + gatewayMethodDispatchAllowed: true, + context: { logGateway: { warn: mocks.warn } }, + }); + mocks.createSession.mockResolvedValue({ relaySessionId: "relay-1" }); + }); + + it("uses the shared Gateway session and maps owner-scoped media events", async () => { + const onEvent = vi.fn(); + const session = await openPluginTalkSession({ + sessionKey: "agent:main:avatar", + voice: "alloy", + onEvent, + }); + const createParams = mocks.createSession.mock.calls[0]?.[0]; + + expect(createParams).toMatchObject({ + context: { logGateway: { warn: mocks.warn } }, + request: { sessionKey: "agent:main:avatar", voice: "alloy" }, + }); + expect(createParams.ownerId).toMatch(/^plugin:avatar:/); + + createParams.eventSink({ relaySessionId: "relay-1", type: "ready" }); + createParams.eventSink({ relaySessionId: "relay-1", type: "audioStarted" }); + createParams.eventSink({ + relaySessionId: "relay-1", + type: "audio", + audioBase64: Buffer.from([1, 0]).toString("base64"), + }); + createParams.eventSink({ relaySessionId: "relay-1", type: "clear", reason: "barge-in" }); + + expect(onEvent.mock.calls.map(([event]) => event)).toEqual([ + { type: "state", generation: 0, ptsMs: 0, state: "listening" }, + { type: "state", generation: 0, ptsMs: 0, state: "speaking" }, + { + type: "audio", + generation: 0, + sequence: 0, + ptsMs: 0, + pcm: Buffer.from([1, 0]), + }, + { type: "clear", generation: 1, reason: "barge-in" }, + { type: "state", generation: 1, ptsMs: 0, state: "listening" }, + ]); + + session.sendAudio(new Uint8Array([2, 0]), { timestamp: 20 }); + session.cancelOutput("barge-in"); + session.close(); + + expect(mocks.sendAudio).toHaveBeenCalledWith({ + relaySessionId: "relay-1", + connId: createParams.ownerId, + audioBase64: "AgA=", + timestamp: 20, + }); + expect(mocks.cancelTurn).toHaveBeenCalledWith({ + relaySessionId: "relay-1", + connId: createParams.ownerId, + reason: "barge-in", + }); + expect(mocks.stopSession).toHaveBeenCalledWith({ + relaySessionId: "relay-1", + connId: createParams.ownerId, + }); + }); + + it("stops accepting media after the Gateway closes the session", async () => { + const onEvent = vi.fn(); + const session = await openPluginTalkSession({ + sessionKey: "agent:main:avatar", + onEvent, + }); + const eventSink = mocks.createSession.mock.calls[0]?.[0].eventSink; + + eventSink({ relaySessionId: "relay-1", type: "close", reason: "error" }); + eventSink({ relaySessionId: "relay-1", type: "close", reason: "error" }); + + expect(onEvent).toHaveBeenCalledOnce(); + expect(onEvent).toHaveBeenCalledWith({ + type: "closed", + generation: 0, + reason: "error", + }); + expect(() => session.sendAudio(new Uint8Array([1, 0]))).toThrow("Talk session is closed"); + session.cancelOutput(); + session.close(); + expect(mocks.cancelTurn).not.toHaveBeenCalled(); + expect(mocks.stopSession).not.toHaveBeenCalled(); + }); + + it("closes the relay when the plugin event callback fails", async () => { + await openPluginTalkSession({ + sessionKey: "agent:main:avatar", + onEvent: async () => { + throw new Error("renderer gone"); + }, + }); + const createParams = mocks.createSession.mock.calls[0]?.[0]; + + createParams.eventSink({ relaySessionId: "relay-1", type: "ready" }); + await vi.waitFor(() => expect(mocks.stopSession).toHaveBeenCalledOnce()); + + expect(mocks.warn).toHaveBeenCalledWith("plugin Talk event delivery failed: renderer gone"); + expect(mocks.stopSession).toHaveBeenCalledWith({ + relaySessionId: "relay-1", + connId: createParams.ownerId, + }); + }); + + it("requires an entitled request scope and a selected agent session", async () => { + mocks.scope.mockReturnValue(undefined); + await expect( + openPluginTalkSession({ sessionKey: "agent:main:avatar", onEvent: vi.fn() }), + ).rejects.toThrow("gatewayMethodDispatch contract"); + + await expect(openPluginTalkSession({ sessionKey: " ", onEvent: vi.fn() })).rejects.toThrow( + "intended agent and workspace", + ); + }); +}); diff --git a/src/gateway/talk-plugin-session.ts b/src/gateway/talk-plugin-session.ts new file mode 100644 index 000000000000..90116d2e8c1e --- /dev/null +++ b/src/gateway/talk-plugin-session.ts @@ -0,0 +1,190 @@ +import { randomUUID } from "node:crypto"; +import { formatErrorMessage } from "../infra/errors.js"; +import { getPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js"; +import { + PLUGIN_TALK_AUDIO_FORMAT, + type OpenPluginTalkSessionParams, + type PluginTalkSession, + type PluginTalkSessionEvent, +} from "../talk/plugin-session.js"; +import type { TalkRealtimeRelayEvent } from "./talk-realtime-relay-state.js"; +import { + cancelTalkRealtimeRelayTurn, + sendTalkRealtimeRelayAudio, + stopTalkRealtimeRelaySession, +} from "./talk-realtime-relay.js"; +import { createGatewayRealtimeTalkSession } from "./talk-realtime-session-create.js"; + +const PCM16_24KHZ_MONO_BYTES_PER_MS = 48; + +function requirePluginTalkScope() { + const scope = getPluginRuntimeGatewayRequestScope(); + if (!scope?.context || !scope.pluginId || scope.gatewayMethodDispatchAllowed !== true) { + throw new Error( + "Interactive Talk sessions require a plugin request route that declares the gatewayMethodDispatch contract.", + ); + } + return { context: scope.context, pluginId: scope.pluginId }; +} + +function createPluginTalkEventSink( + params: OpenPluginTalkSessionParams, + onDeliveryError: (error: unknown) => void, +) { + let generation = 0; + let sequence = 0; + let ptsMs = 0; + let state: Extract["state"] = "idle"; + let closed = false; + + const deliver = (event: PluginTalkSessionEvent): void => { + try { + void Promise.resolve(params.onEvent(event)).catch(onDeliveryError); + } catch (error) { + onDeliveryError(error); + } + }; + const setState = (next: typeof state): void => { + if (state === next || closed) { + return; + } + state = next; + deliver({ type: "state", generation, ptsMs, state }); + }; + + return { + get closed() { + return closed; + }, + eventSink(event: TalkRealtimeRelayEvent): void { + switch (event.type) { + case "ready": + case "inputAudio": + case "audioDone": + setState("listening"); + return; + case "audioStarted": + setState("speaking"); + return; + case "audio": { + setState("speaking"); + const pcm = Buffer.from(event.audioBase64, "base64"); + deliver({ type: "audio", generation, sequence, ptsMs, pcm }); + sequence += 1; + ptsMs += pcm.byteLength / PCM16_24KHZ_MONO_BYTES_PER_MS; + return; + } + case "transcript": + if (event.role === "user" && event.final && event.text.trim()) { + setState("thinking"); + } + return; + case "toolCall": + setState("thinking"); + return; + case "clear": + generation += 1; + sequence = 0; + ptsMs = 0; + deliver({ + type: "clear", + generation, + reason: event.reason === "barge-in" ? "barge-in" : "cancel", + }); + setState("listening"); + return; + case "error": + setState("error"); + return; + case "close": + if (closed) { + return; + } + closed = true; + deliver({ type: "closed", generation, reason: event.reason }); + break; + case "mark": + case "toolCallCancelled": + case "toolProgress": + case "toolResult": + break; + } + }, + }; +} + +export async function openPluginTalkSession( + params: OpenPluginTalkSessionParams, +): Promise { + const sessionKey = params.sessionKey.trim(); + if (!sessionKey) { + throw new Error( + "Choose an OpenClaw session before starting voice so the conversation uses the intended agent and workspace.", + ); + } + const { context, pluginId } = requirePluginTalkScope(); + const ownerId = `plugin:${pluginId}:${randomUUID()}`; + const lifecycle: { relaySessionId?: string } = {}; + let deliveryError: unknown; + const events = createPluginTalkEventSink(params, (error) => { + deliveryError ??= error; + context.logGateway.warn(`plugin Talk event delivery failed: ${formatErrorMessage(error)}`); + if (lifecycle.relaySessionId && !events.closed) { + try { + stopTalkRealtimeRelaySession({ relaySessionId: lifecycle.relaySessionId, connId: ownerId }); + } catch (closeError) { + context.logGateway.warn( + `plugin Talk session cleanup failed: ${formatErrorMessage(closeError)}`, + ); + } + } + }); + const session = await createGatewayRealtimeTalkSession({ + context, + ownerId, + request: { + sessionKey, + ...(params.provider ? { provider: params.provider } : {}), + ...(params.model ? { model: params.model } : {}), + ...(params.voice ? { voice: params.voice } : {}), + ...(params.language ? { language: params.language } : {}), + }, + eventSink: events.eventSink, + }); + lifecycle.relaySessionId = session.relaySessionId; + if (deliveryError) { + stopTalkRealtimeRelaySession({ relaySessionId, connId: ownerId }); + throw deliveryError; + } + + return { + audio: PLUGIN_TALK_AUDIO_FORMAT, + sendAudio(pcm, options) { + if (events.closed) { + throw new Error("Talk session is closed"); + } + sendTalkRealtimeRelayAudio({ + relaySessionId: session.relaySessionId, + connId: ownerId, + audioBase64: Buffer.from(pcm.buffer, pcm.byteOffset, pcm.byteLength).toString("base64"), + timestamp: options?.timestamp, + }); + }, + cancelOutput(reason) { + if (events.closed) { + return; + } + cancelTalkRealtimeRelayTurn({ + relaySessionId: session.relaySessionId, + connId: ownerId, + reason: reason?.trim() || "plugin-cancelled", + }); + }, + close() { + if (events.closed) { + return; + } + stopTalkRealtimeRelaySession({ relaySessionId: session.relaySessionId, connId: ownerId }); + }, + }; +} diff --git a/src/plugin-sdk/realtime-voice.ts b/src/plugin-sdk/realtime-voice.ts index 55234c1e53cb..5302342dd8ed 100644 --- a/src/plugin-sdk/realtime-voice.ts +++ b/src/plugin-sdk/realtime-voice.ts @@ -1,5 +1,11 @@ /** Production-private runtime seam for bundled and separately published official plugins. */ export type { RealtimeVoiceProviderPlugin } from "../plugins/types.js"; +export { + PLUGIN_TALK_AUDIO_FORMAT, + type OpenPluginTalkSessionParams, + type PluginTalkSession, + type PluginTalkSessionEvent, +} from "../talk/plugin-session.js"; export type { OpenAICompatibleRealtimeAudioFormat, RealtimeVoiceAudioFormat, diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.test.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.test.ts index 8b2493cbf0eb..f459cc209b2a 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.test.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.test.ts @@ -5,6 +5,12 @@ import { createPluginRuntimeMock } from "openclaw/plugin-sdk/plugin-test-runtime import { describe, expect, it, vi } from "vitest"; describe("createPluginRuntimeMock", () => { + it("includes the interactive Talk session runtime", () => { + const runtime = createPluginRuntimeMock(); + + expect(vi.isMockFunction(runtime.talk.openSession)).toBe(true); + }); + it("clones the initializer callback input and applies its final extension patch", async () => { const runtime = createPluginRuntimeMock(); const pluginExtensions = { codex: { marker: "original" } }; diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts index 86f9221d9563..743c399ea7d7 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts @@ -509,6 +509,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial = isAvailable: vi.fn(async () => false), request: vi.fn(), }, + talk: { + openSession: vi.fn(), + }, config: { current: vi.fn(() => ({})), mutateConfigFile: createGenericMock( diff --git a/src/plugins/runtime/index.test.ts b/src/plugins/runtime/index.test.ts index 33980d806280..6be571d9f913 100644 --- a/src/plugins/runtime/index.test.ts +++ b/src/plugins/runtime/index.test.ts @@ -166,6 +166,12 @@ describe("plugin runtime command execution", () => { readValue: (runtime: ReturnType) => runtime.version, expected: VERSION, }, + { + name: "exposes runtime.talk.openSession", + readValue: (runtime: ReturnType) => + typeof runtime.talk.openSession, + expected: "function", + }, ] as const)("$name", ({ readValue, expected }) => { expectRuntimeValue(readValue, expected); }); diff --git a/src/plugins/runtime/index.ts b/src/plugins/runtime/index.ts index 04d7aab4c351..9c668d0afbf0 100644 --- a/src/plugins/runtime/index.ts +++ b/src/plugins/runtime/index.ts @@ -46,6 +46,9 @@ const loadModelAuthRuntime = createLazyRuntimeModule( const loadGatewayPluginRuntime = createLazyRuntimeModule( () => import("../../gateway/server-plugins.js"), ); +const loadTalkPluginRuntime = createLazyRuntimeModule( + () => import("../../gateway/talk-plugin-session.js"), +); function createRuntimeGateway(): PluginRuntime["gateway"] { return { @@ -60,6 +63,15 @@ function createRuntimeGateway(): PluginRuntime["gateway"] { }; } +function createRuntimeTalk(): PluginRuntime["talk"] { + return { + openSession: async (params) => { + const runtime = await loadTalkPluginRuntime(); + return await runtime.openPluginTalkSession(params); + }, + }; +} + function createRuntimeTts(): PluginRuntime["tts"] { const bindTtsRuntime = createLazyRuntimeMethodBinder(loadTtsRuntime); const bindTtsRequestRuntime = createLazyRuntimeMethodBinder(loadTtsRequestRuntime); @@ -267,6 +279,7 @@ export function createPluginRuntime(_options: CreatePluginRuntimeOptions = {}): // always see the same version the CLI reports, avoiding API-version drift. version: VERSION, gateway: _options.gateway ?? createRuntimeGateway(), + talk: createRuntimeTalk(), config: createRuntimeConfig(), agent, subagent: _options.subagent ?? createUnavailableSubagentRuntime(), diff --git a/src/plugins/runtime/types.ts b/src/plugins/runtime/types.ts index 228d499dd2bf..9cf3a65e7a8f 100644 --- a/src/plugins/runtime/types.ts +++ b/src/plugins/runtime/types.ts @@ -121,6 +121,11 @@ export type PluginRuntime = PluginRuntimeCore & { options?: RuntimeGatewayRequestOptions, ) => Promise; }; + talk: { + openSession: ( + params: import("../../talk/plugin-session.js").OpenPluginTalkSessionParams, + ) => Promise; + }; subagent: { run: (params: SubagentRunParams) => Promise; waitForRun: (params: SubagentWaitParams) => Promise; diff --git a/src/talk/plugin-session.ts b/src/talk/plugin-session.ts new file mode 100644 index 000000000000..72668e33aa5f --- /dev/null +++ b/src/talk/plugin-session.ts @@ -0,0 +1,46 @@ +export const PLUGIN_TALK_AUDIO_FORMAT = { + encoding: "pcm16le", + sampleRateHz: 24_000, + channels: 1, +} as const; + +export type PluginTalkSessionEvent = + | { + type: "state"; + generation: number; + ptsMs: number; + state: "idle" | "listening" | "thinking" | "speaking" | "error"; + } + | { + type: "audio"; + generation: number; + sequence: number; + ptsMs: number; + pcm: Uint8Array; + } + | { + type: "clear"; + generation: number; + reason: "barge-in" | "cancel" | "replace" | "hangup" | "error"; + } + | { + type: "closed"; + generation: number; + reason: "completed" | "error" | "replaced"; + }; + +export type OpenPluginTalkSessionParams = { + sessionKey: string; + provider?: string; + model?: string; + voice?: string; + language?: string; + onEvent: (event: PluginTalkSessionEvent) => void | Promise; +}; + +export type PluginTalkSession = { + readonly audio: typeof PLUGIN_TALK_AUDIO_FORMAT; + sendAudio(pcm: Uint8Array, options?: { timestamp?: number }): void; + cancelOutput(reason?: string): void; + close(): void; +};