feat(plugin-sdk): open Gateway-managed Talk sessions

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.
 
+  </Accordion>
+  <Accordion title="api.runtime.talk">
+    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.
+
   </Accordion>
   <Accordion title="api.runtime.subagent">
     Launch and manage background subagent runs.

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",
+    );
+  });
+});

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<PluginTalkSessionEvent, { type: "state" }>["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<PluginTalkSession> {
+  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 });
+    },
+  };
+}

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,

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" } };

src/plugin-sdk/test-helpers/plugin-runtime-mock.ts

@@ -509,6 +509,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial<PluginRuntime> =
       isAvailable: vi.fn(async () => false),
       request: vi.fn(),
     },
+    talk: {
+      openSession: vi.fn(),
+    },
     config: {
       current: vi.fn<PluginRuntime["config"]["current"]>(() => ({})),
       mutateConfigFile: createGenericMock<PluginRuntime["config"]["mutateConfigFile"]>(

src/plugins/runtime/index.test.ts

@@ -166,6 +166,12 @@ describe("plugin runtime command execution", () => {
       readValue: (runtime: ReturnType<typeof createPluginRuntime>) => runtime.version,
       expected: VERSION,
     },
+    {
+      name: "exposes runtime.talk.openSession",
+      readValue: (runtime: ReturnType<typeof createPluginRuntime>) =>
+        typeof runtime.talk.openSession,
+      expected: "function",
+    },
   ] as const)("$name", ({ readValue, expected }) => {
     expectRuntimeValue(readValue, expected);
   });

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(),

src/plugins/runtime/types.ts

@@ -121,6 +121,11 @@ export type PluginRuntime = PluginRuntimeCore & {
       options?: RuntimeGatewayRequestOptions,
     ) => Promise<T>;
   };
+  talk: {
+    openSession: (
+      params: import("../../talk/plugin-session.js").OpenPluginTalkSessionParams,
+    ) => Promise<import("../../talk/plugin-session.js").PluginTalkSession>;
+  };
   subagent: {
     run: (params: SubagentRunParams) => Promise<SubagentRunResult>;
     waitForRun: (params: SubagentWaitParams) => Promise<AgentWaitResult>;

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<void>;
+};
+
+export type PluginTalkSession = {
+  readonly audio: typeof PLUGIN_TALK_AUDIO_FORMAT;
+  sendAudio(pcm: Uint8Array, options?: { timestamp?: number }): void;
+  cancelOutput(reason?: string): void;
+  close(): void;
+};