From bf609ba9563196a3fe66dd4e009c95717cf5536a Mon Sep 17 00:00:00 2001
From: Dallin Romney <dallinromney@gmail.com>
Date: Tue, 11 Aug 2026 22:31:07 +0700
Subject: [PATCH] fix(plugin-sdk): bind Talk sessions to plugin lifetimes

---
 docs/plugins/sdk-runtime.md                   |   5 +
 src/gateway/server-methods/talk-session.ts    | 100 ++++------------
 src/gateway/talk-plugin-session.test.ts       | 102 +++++++++++++---
 src/gateway/talk-plugin-session.ts            | 110 +++++++++++++-----
 .../talk-realtime-relay-forced-consults.ts    |   6 +-
 src/gateway/talk-realtime-relay-operations.ts |  20 ++--
 .../talk-realtime-relay-provider-results.ts   |   4 +-
 .../talk-realtime-relay-session-create.ts     |  22 ++--
 src/gateway/talk-realtime-relay-state.ts      |  20 +++-
 src/gateway/talk-realtime-relay.test.ts       | 109 +++++++++++++++++
 src/gateway/talk-realtime-session-create.ts   | 102 ++++++++++++++++
 src/talk/plugin-session.ts                    |   6 +-
 12 files changed, 454 insertions(+), 152 deletions(-)
 create mode 100644 src/gateway/talk-realtime-session-create.ts

diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md
index 970bbec667e1..5ed039fe806c 100644
--- a/docs/plugins/sdk-runtime.md
+++ b/docs/plugins/sdk-runtime.md
@@ -413,8 +413,10 @@ snapshots; OpenClaw owns all persistence and lifecycle coordination.
     PCM and renders the returned events.
 
     ```typescript
+    const connection = new AbortController();
     const session = await api.runtime.talk.openSession({
       sessionKey: "agent:main:avatar",
+      signal: connection.signal,
       onEvent: (event) => renderVoiceEvent(event),
     });
 
@@ -423,6 +425,9 @@ snapshots; OpenClaw owns all persistence and lifecycle coordination.
     session.close();
     ```
 
+    Bind `signal` to the browser or socket displaying the session. Aborting it releases the Talk
+    session if that connection disappears.
+
     `sessionKey` selects the agent conversation and workspace. `provider`, `model`, `voice`, and
     `language` optionally override its configured Talk defaults. Input and output use signed
     PCM16 little-endian audio at 24 kHz, mono (`session.audio.encoding` is `"pcm16"`).
diff --git a/src/gateway/server-methods/talk-session.ts b/src/gateway/server-methods/talk-session.ts
index a753128fdd6e..f8a19aa80543 100644
--- a/src/gateway/server-methods/talk-session.ts
+++ b/src/gateway/server-methods/talk-session.ts
@@ -1,7 +1,4 @@
-import {
-  normalizeOptionalLowercaseString,
-  normalizeOptionalString,
-} from "@openclaw/normalization-core/string-coerce";
+import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
 import {
   ErrorCodes,
   errorShape,
@@ -13,13 +10,9 @@ import {
   validateTalkSessionSubmitToolResultParams,
 } from "../../../packages/gateway-protocol/src/index.js";
 import { AgentSelectionRequiredError } from "../../agents/agent-scope.js";
-import { buildAgentMainSessionKey, parseAgentSessionKey } from "../../routing/session-key.js";
-import { REALTIME_VOICE_AGENT_CONSULT_TOOL } from "../../talk/agent-consult-tool.js";
-import { REALTIME_VOICE_AGENT_CONTROL_TOOL } from "../../talk/agent-run-control-shared.js";
+import { parseAgentSessionKey } from "../../routing/session-key.js";
 import { controlRealtimeVoiceAgentRun } from "../../talk/agent-run-control.js";
 import { resolveTalkSessionAgentId } from "../../talk/agent-target.js";
-import { ensureClientVoiceAgentSessionEntry } from "../../talk/client-voice-session.js";
-import { resolveConfiguredRealtimeVoiceProvider } from "../../talk/provider-resolver.js";
 import { ADMIN_SCOPE } from "../operator-scopes.js";
 import { resolveRequestedSessionAgentId } from "../session-request-agent.js";
 import { resolveSessionKeyFromResolveParams } from "../sessions-resolve.js";
@@ -27,12 +20,15 @@ import { resolveTalkAgentConsultAuthority } from "../talk-client-gateway-control
 import { createTalkHandoff, getTalkHandoff, revokeTalkHandoff } from "../talk-handoff.js";
 import {
   cancelTalkRealtimeRelayTurn,
-  createTalkRealtimeRelaySession,
   sendTalkRealtimeRelayAudio,
   steerTalkRealtimeRelayAgentRun,
   stopTalkRealtimeRelaySession,
   submitTalkRealtimeRelayToolResult,
 } from "../talk-realtime-relay.js";
+import {
+  createGatewayRealtimeTalkSession,
+  TalkRealtimeSessionRequestError,
+} from "../talk-realtime-session-create.js";
 import {
   forgetUnifiedTalkSession,
   getUnifiedTalkSession,
@@ -48,17 +44,12 @@ import { formatForLog } from "../ws-log.js";
 import { acknowledgeTalkSessionMark } from "./talk-session-mark.js";
 import {
   broadcastTalkRoomEvents,
-  buildRealtimeInstructions,
-  buildRealtimeVoiceLaunchOptions,
-  buildTalkRealtimeConfig,
   buildTalkTranscriptionConfig,
   canUseTalkDirectTools,
   normalizeTalkSessionBrain,
   normalizeTalkSessionMode,
   normalizeTalkSessionTransport,
   resolveConfiguredRealtimeTranscriptionProvider,
-  resolveTalkRealtimeProviderInstructions,
-  resolveTalkRealtimeGatewayRelayLaunch,
 } from "./talk-shared.js";
 import type { GatewayRequestHandlers, RespondFn } from "./types.js";
 import { assertValidParams } from "./validation.js";
@@ -239,11 +230,6 @@ export const talkSessionHandlers: GatewayRequestHandlers = {
           );
         }
         const runtimeConfig = context.getRuntimeConfig();
-        const realtimeConfig = buildTalkRealtimeConfig(runtimeConfig, params.provider);
-        const launchOptions = buildRealtimeVoiceLaunchOptions({
-          requested: params,
-          defaults: realtimeConfig,
-        });
         const requestedSessionKey = normalizeOptionalString(params.sessionKey);
         const bareTalkAgentId =
           requestedSessionKey && !parseAgentSessionKey(requestedSessionKey)
@@ -260,67 +246,21 @@ export const talkSessionHandlers: GatewayRequestHandlers = {
           requestedOwner?.agentId ??
           bareTalkAgentId ??
           resolveTalkSessionAgentId(runtimeConfig, requestedSessionKey);
-        const resolution = resolveConfiguredRealtimeVoiceProvider({
-          configuredProviderId: realtimeConfig.provider,
-          providerConfigs: realtimeConfig.providers,
-          providerConfigOverrides: launchOptions.model ? { model: launchOptions.model } : {},
-          cfg: runtimeConfig,
-          agentId,
-          defaultModel: realtimeConfig.model,
-          surface: "gateway-relay",
-        });
-        const relayLaunch = resolveTalkRealtimeGatewayRelayLaunch({
-          ...resolution,
-          cfg: runtimeConfig,
-          launchOptions,
-          consultRouting: realtimeConfig.consultRouting,
-        });
-        if (relayLaunch.error) {
-          // GPT-Live delegates natively; forced transcript consults are a GA-model mode.
-          return respondInvalidRequest(respond, relayLaunch.error);
+        try {
+          const session = await createGatewayRealtimeTalkSession({
+            context,
+            ownerId: connId,
+            agentId,
+            request: params,
+          });
+          return respondOk(respond, session);
+        } catch (error) {
+          if (error instanceof TalkRealtimeSessionRequestError) {
+            respondInvalidRequest(respond, error.message);
+            return;
+          }
+          throw error;
         }
-        const realtimeContext = await resolveTalkRealtimeProviderInstructions({
-          config: runtimeConfig,
-          agentId,
-          configuredInstructions: realtimeConfig.instructions,
-          sessionKey: params.sessionKey,
-          requireSessionKeyForProfile: true,
-          warn: (message) => context.logGateway.warn(`talk realtime context: ${message}`),
-        });
-        const sessionKey =
-          realtimeContext.requestedSessionKey ??
-          buildAgentMainSessionKey({ agentId: realtimeContext.agentId });
-        await ensureClientVoiceAgentSessionEntry({
-          agentId: realtimeContext.agentId,
-          sessionKey,
-        });
-        const session = createTalkRealtimeRelaySession({
-          context,
-          connId,
-          cfg: runtimeConfig,
-          consultAuthority: resolveTalkAgentConsultAuthority(client?.connect?.scopes),
-          provider: resolution.provider,
-          providerConfig: relayLaunch.providerConfig,
-          instructions: buildRealtimeInstructions(realtimeContext.instructions),
-          tools: [REALTIME_VOICE_AGENT_CONSULT_TOOL, REALTIME_VOICE_AGENT_CONTROL_TOOL],
-          model: launchOptions.model,
-          sessionKey,
-          voice: launchOptions.voice,
-          language: normalizeOptionalLowercaseString(params.language),
-          forceAgentConsultOnFinalTranscript: relayLaunch.forceAgentConsultOnFinalTranscript,
-        });
-        rememberUnifiedTalkSession(session.relaySessionId, {
-          kind: "realtime-relay",
-          connId,
-          relaySessionId: session.relaySessionId,
-        });
-        return respondOk(respond, {
-          ...session,
-          sessionId: session.relaySessionId,
-          voiceSessionId: session.relaySessionId,
-          mode,
-          brain,
-        });
       }
 
       if (mode === "transcription") {
diff --git a/src/gateway/talk-plugin-session.test.ts b/src/gateway/talk-plugin-session.test.ts
index d2adcce8e68a..4100701814fd 100644
--- a/src/gateway/talk-plugin-session.test.ts
+++ b/src/gateway/talk-plugin-session.test.ts
@@ -4,7 +4,7 @@ const mocks = vi.hoisted(() => ({
   scope: vi.fn(),
   createSession: vi.fn(),
   sendAudio: vi.fn(),
-  cancelTurn: vi.fn(),
+  cancelOutput: vi.fn(),
   stopSession: vi.fn(),
   warn: vi.fn(),
 }));
@@ -17,15 +17,18 @@ vi.mock("./talk-realtime-session-create.js", () => ({
 }));
 vi.mock("./talk-realtime-relay.js", () => ({
   sendTalkRealtimeRelayAudio: mocks.sendAudio,
-  cancelTalkRealtimeRelayTurn: mocks.cancelTurn,
+  cancelTalkRealtimeRelayOutput: mocks.cancelOutput,
   stopTalkRealtimeRelaySession: mocks.stopSession,
 }));
 
 import { openPluginTalkSession } from "./talk-plugin-session.js";
 
 describe("plugin Talk session", () => {
+  let controller: AbortController;
+
   beforeEach(() => {
     vi.clearAllMocks();
+    controller = new AbortController();
     mocks.scope.mockReturnValue({
       pluginId: "avatar",
       gatewayMethodDispatchAllowed: true,
@@ -42,6 +45,7 @@ describe("plugin Talk session", () => {
     const onEvent = vi.fn();
     const session = await openPluginTalkSession({
       sessionKey: "agent:main:avatar",
+      signal: controller.signal,
       voice: "alloy",
       onEvent,
     });
@@ -51,14 +55,20 @@ describe("plugin Talk session", () => {
       context: { logGateway: { warn: mocks.warn } },
       request: { sessionKey: "agent:main:avatar", voice: "alloy" },
     });
-    expect(createParams.ownerId).toBe("plugin:avatar:plugin-http:127.0.0.1");
+    expect(createParams.ownerId).toMatch(/^plugin:avatar:[0-9a-f-]+$/);
+    expect(createParams.quotaOwnerId).toBe("plugin:avatar:plugin-http:127.0.0.1");
 
     createParams.eventSink({ relaySessionId: "relay-1", type: "ready" });
-    createParams.eventSink({ relaySessionId: "relay-1", type: "audioStarted" });
+    createParams.eventSink({
+      relaySessionId: "relay-1",
+      type: "audioStarted",
+      outputGeneration: 1,
+    });
     createParams.eventSink({
       relaySessionId: "relay-1",
       type: "audio",
       audioBase64: Buffer.from([1, 0]).toString("base64"),
+      outputGeneration: 1,
     });
     createParams.eventSink({ relaySessionId: "relay-1", type: "clear", reason: "barge-in" });
 
@@ -86,9 +96,10 @@ describe("plugin Talk session", () => {
       audioBase64: "AgA=",
       timestamp: 20,
     });
-    expect(mocks.cancelTurn).toHaveBeenCalledWith({
+    expect(mocks.cancelOutput).toHaveBeenCalledWith({
       relaySessionId: "relay-1",
       connId: createParams.ownerId,
+      outputGeneration: 1,
       reason: "barge-in",
     });
     expect(mocks.stopSession).toHaveBeenCalledWith({
@@ -97,11 +108,21 @@ describe("plugin Talk session", () => {
     });
   });
 
-  it("shares the route owner across opens so relay session limits apply", async () => {
-    await openPluginTalkSession({ sessionKey: "agent:main:first", onEvent: vi.fn() });
-    await openPluginTalkSession({ sessionKey: "agent:main:second", onEvent: vi.fn() });
+  it("separates per-session cleanup ownership from route quotas", async () => {
+    await openPluginTalkSession({
+      sessionKey: "agent:main:first",
+      signal: controller.signal,
+      onEvent: vi.fn(),
+    });
+    await openPluginTalkSession({
+      sessionKey: "agent:main:second",
+      signal: controller.signal,
+      onEvent: vi.fn(),
+    });
 
-    expect(mocks.createSession.mock.calls.map(([params]) => params.ownerId)).toEqual([
+    const createParams = mocks.createSession.mock.calls.map(([params]) => params);
+    expect(createParams[0].ownerId).not.toBe(createParams[1].ownerId);
+    expect(createParams.map((params) => params.quotaOwnerId)).toEqual([
       "plugin:avatar:plugin-http:127.0.0.1",
       "plugin:avatar:plugin-http:127.0.0.1",
     ]);
@@ -111,6 +132,7 @@ describe("plugin Talk session", () => {
     const onEvent = vi.fn();
     const session = await openPluginTalkSession({
       sessionKey: "agent:main:avatar",
+      signal: controller.signal,
       onEvent,
     });
     const eventSink = mocks.createSession.mock.calls[0]?.[0].eventSink;
@@ -127,13 +149,14 @@ describe("plugin Talk session", () => {
     expect(() => session.sendAudio(new Uint8Array([1, 0]))).toThrow("Talk session is closed");
     session.cancelOutput();
     session.close();
-    expect(mocks.cancelTurn).not.toHaveBeenCalled();
+    expect(mocks.cancelOutput).not.toHaveBeenCalled();
     expect(mocks.stopSession).not.toHaveBeenCalled();
   });
 
   it("closes the relay when the plugin event callback fails", async () => {
     await openPluginTalkSession({
       sessionKey: "agent:main:avatar",
+      signal: controller.signal,
       onEvent: async () => {
         throw new Error("renderer gone");
       },
@@ -159,6 +182,7 @@ describe("plugin Talk session", () => {
     await expect(
       openPluginTalkSession({
         sessionKey: "agent:main:avatar",
+        signal: controller.signal,
         onEvent: () => {
           throw new Error("renderer gone");
         },
@@ -167,7 +191,7 @@ describe("plugin Talk session", () => {
 
     expect(mocks.stopSession).toHaveBeenCalledWith({
       relaySessionId: "relay-1",
-      connId: "plugin:avatar:plugin-http:127.0.0.1",
+      connId: mocks.createSession.mock.calls[0]?.[0].ownerId,
     });
   });
 
@@ -183,7 +207,11 @@ describe("plugin Talk session", () => {
     });
 
     await expect(
-      openPluginTalkSession({ sessionKey: "agent:main:avatar", onEvent: vi.fn() }),
+      openPluginTalkSession({
+        sessionKey: "agent:main:avatar",
+        signal: controller.signal,
+        onEvent: vi.fn(),
+      }),
     ).rejects.toThrow("authenticated plugin request with Talk access");
     expect(mocks.createSession).not.toHaveBeenCalled();
   });
@@ -191,11 +219,55 @@ describe("plugin Talk session", () => {
   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() }),
+      openPluginTalkSession({
+        sessionKey: "agent:main:avatar",
+        signal: controller.signal,
+        onEvent: vi.fn(),
+      }),
     ).rejects.toThrow("gatewayMethodDispatch contract");
 
-    await expect(openPluginTalkSession({ sessionKey: " ", onEvent: vi.fn() })).rejects.toThrow(
-      "intended agent and workspace",
+    await expect(
+      openPluginTalkSession({ sessionKey: " ", signal: controller.signal, onEvent: vi.fn() }),
+    ).rejects.toThrow("intended agent and workspace");
+  });
+
+  it("closes the relay when its consuming connection ends", async () => {
+    await openPluginTalkSession({
+      sessionKey: "agent:main:avatar",
+      signal: controller.signal,
+      onEvent: vi.fn(),
+    });
+    const createParams = mocks.createSession.mock.calls[0]?.[0];
+
+    controller.abort();
+
+    expect(mocks.stopSession).toHaveBeenCalledWith({
+      relaySessionId: "relay-1",
+      connId: createParams.ownerId,
+    });
+  });
+
+  it("closes a relay that finishes opening after its connection ends", async () => {
+    let resolveSession: ((session: { relaySessionId: string }) => void) | undefined;
+    mocks.createSession.mockReturnValueOnce(
+      new Promise((resolve) => {
+        resolveSession = resolve;
+      }),
     );
+    const opening = openPluginTalkSession({
+      sessionKey: "agent:main:avatar",
+      signal: controller.signal,
+      onEvent: vi.fn(),
+    });
+    const createParams = mocks.createSession.mock.calls[0]?.[0];
+
+    controller.abort(new Error("browser disconnected"));
+    resolveSession?.({ relaySessionId: "relay-late" });
+
+    await expect(opening).rejects.toThrow("browser disconnected");
+    expect(mocks.stopSession).toHaveBeenCalledWith({
+      relaySessionId: "relay-late",
+      connId: createParams.ownerId,
+    });
   });
 });
diff --git a/src/gateway/talk-plugin-session.ts b/src/gateway/talk-plugin-session.ts
index 7e21e4f1be73..cb266d2dbb3a 100644
--- a/src/gateway/talk-plugin-session.ts
+++ b/src/gateway/talk-plugin-session.ts
@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import { formatErrorMessage } from "../infra/errors.js";
 import { getPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js";
 import {
@@ -9,7 +10,7 @@ import {
 import { authorizeOperatorScopesForMethod } from "./method-scopes.js";
 import type { TalkRealtimeRelayEvent } from "./talk-realtime-relay-state.js";
 import {
-  cancelTalkRealtimeRelayTurn,
+  cancelTalkRealtimeRelayOutput,
   sendTalkRealtimeRelayAudio,
   stopTalkRealtimeRelaySession,
 } from "./talk-realtime-relay.js";
@@ -17,6 +18,10 @@ import { createGatewayRealtimeTalkSession } from "./talk-realtime-session-create
 
 const PCM16_24KHZ_MONO_BYTES_PER_MS = 48;
 
+function talkSessionAbortError(signal: AbortSignal, fallback: string): Error {
+  return signal.reason instanceof Error ? signal.reason : new Error(fallback);
+}
+
 function requirePluginTalkScope() {
   const scope = getPluginRuntimeGatewayRequestScope();
   if (
@@ -37,7 +42,8 @@ function requirePluginTalkScope() {
   }
   return {
     context: scope.context,
-    ownerId: `plugin:${scope.pluginId}:${scope.client.connId}`,
+    pluginId: scope.pluginId,
+    quotaOwnerId: `plugin:${scope.pluginId}:${scope.client.connId}`,
   };
 }
 
@@ -50,6 +56,7 @@ function createPluginTalkEventSink(
   let ptsMs = 0;
   let state: Extract<PluginTalkSessionEvent, { type: "state" }>["state"] = "idle";
   let closed = false;
+  let outputGeneration: number | undefined;
 
   const deliver = (event: PluginTalkSessionEvent): void => {
     try {
@@ -70,6 +77,9 @@ function createPluginTalkEventSink(
     get closed() {
       return closed;
     },
+    get outputGeneration() {
+      return outputGeneration;
+    },
     eventSink(event: TalkRealtimeRelayEvent): void {
       switch (event.type) {
         case "ready":
@@ -78,9 +88,11 @@ function createPluginTalkEventSink(
           setState("listening");
           return;
         case "audioStarted":
+          outputGeneration = event.outputGeneration;
           setState("speaking");
           return;
         case "audio": {
+          outputGeneration = event.outputGeneration;
           setState("speaking");
           const pcm = Buffer.from(event.audioBase64, "base64");
           deliver({ type: "audio", generation, sequence, ptsMs, pcm });
@@ -136,38 +148,72 @@ export async function openPluginTalkSession(
       "Choose an OpenClaw session before starting voice so the conversation uses the intended agent and workspace.",
     );
   }
-  const { context, ownerId } = requirePluginTalkScope();
-  const lifecycle: { relaySessionId?: string } = {};
+  if (params.signal.aborted) {
+    throw talkSessionAbortError(params.signal, "Talk session was cancelled before it opened");
+  }
+  const { context, pluginId, quotaOwnerId } = requirePluginTalkScope();
+  const ownerId = `plugin:${pluginId}:${randomUUID()}`;
+  const lifecycle: { relaySessionId?: string; aborted: boolean; removeAbortListener?: () => void } =
+    {
+      aborted: false,
+    };
+  const stopRelay = (): void => {
+    const relaySessionId = lifecycle.relaySessionId;
+    if (!relaySessionId || events.closed) {
+      return;
+    }
+    try {
+      stopTalkRealtimeRelaySession({ relaySessionId, connId: ownerId });
+    } catch (error) {
+      context.logGateway.warn(`plugin Talk session cleanup failed: ${formatErrorMessage(error)}`);
+    }
+  };
   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,
+    stopRelay();
   });
+  const abort = (): void => {
+    lifecycle.aborted = true;
+    stopRelay();
+  };
+  params.signal.addEventListener("abort", abort, { once: true });
+  lifecycle.removeAbortListener = () => params.signal.removeEventListener("abort", abort);
+  let session: Awaited<ReturnType<typeof createGatewayRealtimeTalkSession>>;
+  try {
+    session = await createGatewayRealtimeTalkSession({
+      context,
+      ownerId,
+      quotaOwnerId,
+      request: {
+        sessionKey,
+        ...(params.provider ? { provider: params.provider } : {}),
+        ...(params.model ? { model: params.model } : {}),
+        ...(params.voice ? { voice: params.voice } : {}),
+        ...(params.language ? { language: params.language } : {}),
+      },
+      eventSink: (event) => {
+        events.eventSink(event);
+        if (event.type === "close") {
+          lifecycle.removeAbortListener?.();
+        }
+      },
+    });
+  } catch (error) {
+    lifecycle.removeAbortListener();
+    throw error;
+  }
   lifecycle.relaySessionId = session.relaySessionId;
-  if (deliveryError) {
-    stopTalkRealtimeRelaySession({ relaySessionId: session.relaySessionId, connId: ownerId });
-    throw deliveryError;
+  if (lifecycle.aborted || deliveryError) {
+    stopRelay();
+    lifecycle.removeAbortListener();
+    if (deliveryError) {
+      throw deliveryError instanceof Error
+        ? deliveryError
+        : new Error(`Plugin Talk event delivery failed: ${formatErrorMessage(deliveryError)}`);
+    }
+    throw talkSessionAbortError(params.signal, "Talk session was cancelled while opening");
   }
 
   return {
@@ -187,9 +233,14 @@ export async function openPluginTalkSession(
       if (events.closed) {
         return;
       }
-      cancelTalkRealtimeRelayTurn({
+      const outputGeneration = events.outputGeneration;
+      if (outputGeneration === undefined) {
+        return;
+      }
+      cancelTalkRealtimeRelayOutput({
         relaySessionId: session.relaySessionId,
         connId: ownerId,
+        outputGeneration,
         reason: reason?.trim() || "plugin-cancelled",
       });
     },
@@ -197,6 +248,7 @@ export async function openPluginTalkSession(
       if (events.closed) {
         return;
       }
+      lifecycle.removeAbortListener?.();
       stopTalkRealtimeRelaySession({ relaySessionId: session.relaySessionId, connId: ownerId });
     },
   };
diff --git a/src/gateway/talk-realtime-relay-forced-consults.ts b/src/gateway/talk-realtime-relay-forced-consults.ts
index ee8be0aeeb91..0929afa55ebc 100644
--- a/src/gateway/talk-realtime-relay-forced-consults.ts
+++ b/src/gateway/talk-realtime-relay-forced-consults.ts
@@ -18,7 +18,7 @@ import {
   trackPendingWorkingToolResult,
 } from "./talk-realtime-relay-provider-results.js";
 import {
-  broadcastToOwner,
+  publishTalkRealtimeRelayEvent,
   ensureRelayTurn,
   noFallbackRelayOutputFlush,
   relaySessions,
@@ -181,7 +181,7 @@ export function scheduleForcedAgentConsult(
       { audioPlaybackActive: true, force: true },
       noFallbackRelayOutputFlush,
     );
-    broadcastToOwner(session.context, session.connId, {
+    publishTalkRealtimeRelayEvent(session, {
       relaySessionId: session.id,
       type: "toolCall",
       itemId,
@@ -287,7 +287,7 @@ export function submitRealtimeAgentConsultWorkingResponse(
     if (session.toolResultEpoch !== epoch) {
       return;
     }
-    broadcastToOwner(session.context, session.connId, {
+    publishTalkRealtimeRelayEvent(session, {
       relaySessionId: session.id,
       type: "toolResult",
       callId,
diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts
index ba9999645102..0cc7111c5d8d 100644
--- a/src/gateway/talk-realtime-relay-operations.ts
+++ b/src/gateway/talk-realtime-relay-operations.ts
@@ -29,10 +29,10 @@ import {
   MAX_AUDIO_BASE64_BYTES,
   MAX_RELAY_SESSIONS_GLOBAL,
   MAX_RELAY_SESSIONS_PER_CONN,
-  broadcastToOwner,
   drainingRelaySessions,
   ensureRelayTurn,
   noFallbackRelayOutputFlush,
+  publishTalkRealtimeRelayEvent,
   relaySessions,
   resolveRelayProviderToolCallId,
   type RelaySession,
@@ -122,7 +122,7 @@ export function closeRelaySession(
     // Provider teardown may throw, but the relay must still reach its durable
     // voice and owner-visible terminal state before that error is surfaced.
     void closeRelayVoiceSession(session);
-    broadcastToOwner(session.context, session.connId, {
+    publishTalkRealtimeRelayEvent(session, {
       relaySessionId: session.id,
       type: "close",
       reason,
@@ -157,27 +157,27 @@ function pruneExpiredRelaySessions(nowMs = Date.now()): void {
   });
 }
 
-function countRelaySessionsForConn(connId: string): number {
+function countRelaySessionsForQuotaOwner(quotaOwnerId: string): number {
   let count = 0;
   for (const session of relaySessions.values()) {
-    if (session.connId === connId) {
+    if (session.quotaOwnerId === quotaOwnerId) {
       count += 1;
     }
   }
   for (const session of drainingRelaySessions.values()) {
-    if (session.connId === connId) {
+    if (session.quotaOwnerId === quotaOwnerId) {
       count += 1;
     }
   }
   return count;
 }
 
-export function enforceRelaySessionLimits(connId: string): void {
+export function enforceRelaySessionLimits(quotaOwnerId: string): void {
   pruneExpiredRelaySessions();
   if (relaySessions.size + drainingRelaySessions.size >= MAX_RELAY_SESSIONS_GLOBAL) {
     throw new Error("Too many active realtime relay sessions");
   }
-  if (countRelaySessionsForConn(connId) >= MAX_RELAY_SESSIONS_PER_CONN) {
+  if (countRelaySessionsForQuotaOwner(quotaOwnerId) >= MAX_RELAY_SESSIONS_PER_CONN) {
     throw new Error("Too many active realtime relay sessions for this connection");
   }
 }
@@ -206,7 +206,7 @@ export function sendTalkRealtimeRelayAudio(params: {
   const audio = decodeTalkRelayAudioBase64(params.audioBase64, "Realtime relay");
   const turnId = ensureRelayTurn(session);
   session.bridge.sendAudio(audio);
-  broadcastToOwner(session.context, session.connId, {
+  publishTalkRealtimeRelayEvent(session, {
     relaySessionId: session.id,
     type: "inputAudio",
     byteLength: audio.byteLength,
@@ -494,7 +494,7 @@ export async function steerTalkRealtimeRelayAgentRun(params: {
   if (relaySessions.get(session.id) !== session) {
     return finalResult;
   }
-  broadcastToOwner(session.context, session.connId, {
+  publishTalkRealtimeRelayEvent(session, {
     relaySessionId: session.id,
     type: "toolProgress",
     result: finalResult,
@@ -547,7 +547,7 @@ export function cancelTalkRealtimeRelayTurn(params: {
     turnId,
     payload: { reason },
   });
-  broadcastToOwner(session.context, session.connId, {
+  publishTalkRealtimeRelayEvent(session, {
     relaySessionId: session.id,
     type: "clear",
     talkEvent: cancelled.ok ? cancelled.event : undefined,
diff --git a/src/gateway/talk-realtime-relay-provider-results.ts b/src/gateway/talk-realtime-relay-provider-results.ts
index 399fa4c3141f..0d4117dc50b4 100644
--- a/src/gateway/talk-realtime-relay-provider-results.ts
+++ b/src/gateway/talk-realtime-relay-provider-results.ts
@@ -1,7 +1,7 @@
 import { buildRealtimeVoiceAgentCancelProviderResult } from "../talk/agent-run-control-shared.js";
 import type { RealtimeVoiceToolResultOptions } from "../talk/provider-types.js";
 import {
-  broadcastToOwner,
+  publishTalkRealtimeRelayEvent,
   relaySessions,
   resolveRelayProviderToolCallId,
   type RelaySession,
@@ -27,7 +27,7 @@ export function broadcastToolResultToOwner(
 ): void {
   const payload =
     params.forced === true ? { result: params.result, forced: true } : { result: params.result };
-  broadcastToOwner(session.context, session.connId, {
+  publishTalkRealtimeRelayEvent(session, {
     relaySessionId: session.id,
     type: "toolResult",
     callId: params.callId,
diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts
index 9f6089a97d47..e368dc1acf30 100644
--- a/src/gateway/talk-realtime-relay-session-create.ts
+++ b/src/gateway/talk-realtime-relay-session-create.ts
@@ -41,8 +41,8 @@ import {
   RELAY_SESSION_TTL_MS,
   RELAY_TRANSCRIPT_ECHO_LOOKBACK_MS,
   adoptRelayProviderToolCallId,
-  broadcastToOwner,
   ensureRelayTurn,
+  publishTalkRealtimeRelayEvent,
   relaySessions,
   type CreateTalkRealtimeRelaySessionParams,
   type RelaySession,
@@ -71,7 +71,8 @@ function isRelayAssistantEchoTranscript(session: RelaySession | undefined, text:
 export function createTalkRealtimeRelaySession(
   params: CreateTalkRealtimeRelaySessionParams,
 ): TalkRealtimeRelaySessionResult {
-  enforceRelaySessionLimits(params.connId);
+  const quotaOwnerId = params.quotaOwnerId ?? params.connId;
+  enforceRelaySessionLimits(quotaOwnerId);
   const forceAgentConsultOnFinalTranscript = params.forceAgentConsultOnFinalTranscript === true;
   const relaySessionId = randomUUID();
   const expiresAtMs = resolveExpiresAtMsFromDurationMs(RELAY_SESSION_TTL_MS);
@@ -99,8 +100,13 @@ export function createTalkRealtimeRelaySession(
     transcriptLookbackMs: RELAY_TRANSCRIPT_ECHO_LOOKBACK_MS,
     captureBridgeEvents: false,
   });
+  const eventOwner = {
+    context: params.context,
+    connId: params.connId,
+    ...(params.eventSink ? { eventSink: params.eventSink } : {}),
+  };
   const emit = (event: TalkRealtimeRelayEventPayload, talkEvent?: TalkEventInput) =>
-    broadcastToOwner(params.context, params.connId, {
+    publishTalkRealtimeRelayEvent(eventOwner, {
       ...event,
       ...(talkEvent ? { talkEvent: harness.emit(talkEvent) } : {}),
     });
@@ -285,7 +291,7 @@ export function createTalkRealtimeRelaySession(
           return;
         }
         const clearEvent = { relaySessionId, type: "clear" as const };
-        broadcastToOwner(params.context, params.connId, {
+        publishTalkRealtimeRelayEvent(eventOwner, {
           ...clearEvent,
           ...(talkEvent ? { talkEvent } : {}),
         });
@@ -305,7 +311,7 @@ export function createTalkRealtimeRelaySession(
             type: "toolCallCancelled" as const,
             callId: relayCallId,
           };
-          broadcastToOwner(params.context, params.connId, cancelledEvent);
+          publishTalkRealtimeRelayEvent(eventOwner, cancelledEvent);
         }
         return;
       }
@@ -324,7 +330,7 @@ export function createTalkRealtimeRelaySession(
         return;
       }
       const terminalTalkEvent = harness.talk.recentEvents.at(-1);
-      broadcastToOwner(params.context, params.connId, {
+      publishTalkRealtimeRelayEvent(eventOwner, {
         relaySessionId,
         type: "audioDone",
         ...(currentOutputItemId ? { itemId: currentOutputItemId } : {}),
@@ -348,7 +354,7 @@ export function createTalkRealtimeRelaySession(
         const errorTalkEvent = harness.talk.recentEvents.findLast(
           (event) => event.type === "session.error" && event.payload === outcome,
         );
-        broadcastToOwner(params.context, params.connId, {
+        publishTalkRealtimeRelayEvent(eventOwner, {
           ...relayIssuePayload(relaySessionId, issue),
           ...(errorTalkEvent ? { talkEvent: errorTalkEvent } : {}),
         });
@@ -555,7 +561,9 @@ export function createTalkRealtimeRelaySession(
   const relay: RelaySession = {
     id: relaySessionId,
     connId: params.connId,
+    quotaOwnerId,
     context: params.context,
+    ...(params.eventSink ? { eventSink: params.eventSink } : {}),
     bridge,
     harness,
     sessionKey: initialSessionKey,
diff --git a/src/gateway/talk-realtime-relay-state.ts b/src/gateway/talk-realtime-relay-state.ts
index ae16e1439cab..0805dae9d2da 100644
--- a/src/gateway/talk-realtime-relay-state.ts
+++ b/src/gateway/talk-realtime-relay-state.ts
@@ -1,5 +1,6 @@
 import { randomUUID } from "node:crypto";
 import type { OpenClawConfig } from "../config/types.js";
+import { formatErrorMessage } from "../infra/errors.js";
 import type { RealtimeVoiceProviderPlugin } from "../plugins/types.js";
 import type { BoundedSerialQueue } from "../shared/bounded-serial-queue.js";
 import type { RealtimeVoiceAgentControlResult } from "../talk/agent-run-control.js";
@@ -70,7 +71,8 @@ export type TalkRealtimeRelayEventPayload =
     }
   | { relaySessionId: string; type: "close"; reason: "completed" | "error" };
 
-type TalkRealtimeRelayEvent = TalkRealtimeRelayEventPayload & { talkEvent?: TalkEvent };
+export type TalkRealtimeRelayEvent = TalkRealtimeRelayEventPayload & { talkEvent?: TalkEvent };
+export type TalkRealtimeRelayEventSink = (event: TalkRealtimeRelayEvent) => void;
 
 export type ForcedTerminalProviderResult = {
   result: unknown;
@@ -87,7 +89,9 @@ export type RelayAgentControlProviderSubmission = {
 export type RelaySession = {
   id: string;
   connId: string;
+  quotaOwnerId: string;
   context: GatewayRequestContext;
+  eventSink?: TalkRealtimeRelayEventSink;
   bridge: RealtimeVoiceBridgeSession;
   harness: RealtimeVoiceSessionHarness;
   sessionKey?: string;
@@ -122,6 +126,8 @@ export type RelaySession = {
 export type CreateTalkRealtimeRelaySessionParams = {
   context: GatewayRequestContext;
   connId: string;
+  quotaOwnerId?: string;
+  eventSink?: TalkRealtimeRelayEventSink;
   cfg?: OpenClawConfig;
   consultAuthority?: TalkAgentConsultAuthority;
   provider: RealtimeVoiceProviderPlugin;
@@ -184,15 +190,19 @@ export function resolveRelayProviderToolCallId(session: RelaySession, relayCallI
   return session.providerToolCallIds.get(relayCallId) ?? relayCallId;
 }
 
-export function broadcastToOwner(
-  context: GatewayRequestContext,
-  connId: string,
+export function publishTalkRealtimeRelayEvent(
+  owner: Pick<RelaySession, "connId" | "context" | "eventSink">,
   event: TalkRealtimeRelayEvent,
 ): void {
   // Classify the materialized Talk event so final results cannot be mistaken
   // for transient tool progress by individual provider callback paths.
   const delivery = relayEventDeliveryOptions(event, event.talkEvent);
-  context.broadcastToConnIds(RELAY_EVENT, event, new Set([connId]), delivery);
+  try {
+    owner.eventSink?.(event);
+  } catch (error) {
+    owner.context.logGateway.warn(`talk realtime event sink failed: ${formatErrorMessage(error)}`);
+  }
+  owner.context.broadcastToConnIds(RELAY_EVENT, event, new Set([owner.connId]), delivery);
 }
 
 function relayEventDeliveryOptions(
diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts
index 9e138a71d637..7e2d39d00437 100644
--- a/src/gateway/talk-realtime-relay.test.ts
+++ b/src/gateway/talk-realtime-relay.test.ts
@@ -215,6 +215,82 @@ describe("talk realtime gateway relay", () => {
     };
   }
 
+  it("delivers owner events to the local sink and Gateway connection", () => {
+    const eventSink = vi.fn();
+    const broadcastToConnIds = vi.fn();
+    const session = createTalkRealtimeRelaySession({
+      context: {
+        broadcastToConnIds,
+        getRuntimeConfig: () => ({}),
+        logGateway: { warn: vi.fn() },
+      } as never,
+      connId: "conn-local-sink",
+      eventSink,
+      provider: createIdleRelayProvider(),
+      providerConfig: {},
+      instructions: "brief",
+      tools: [],
+    });
+
+    sendTalkRealtimeRelayAudio({
+      relaySessionId: session.relaySessionId,
+      connId: "conn-local-sink",
+      audioBase64: Buffer.from([1, 2]).toString("base64"),
+    });
+
+    expect(eventSink).toHaveBeenCalledWith(
+      expect.objectContaining({
+        relaySessionId: session.relaySessionId,
+        type: "inputAudio",
+        byteLength: 2,
+      }),
+    );
+    expect(broadcastToConnIds).toHaveBeenCalledWith(
+      "talk.event",
+      expect.objectContaining({ relaySessionId: session.relaySessionId, type: "inputAudio" }),
+      new Set(["conn-local-sink"]),
+      { dropIfSlow: true },
+    );
+  });
+
+  it("keeps Gateway and provider delivery alive when the local sink throws", () => {
+    const sendAudio = vi.fn();
+    const provider = createIdleRelayProvider();
+    provider.createBridge = () => makeRelayTransport({ sendAudio });
+    const broadcastToConnIds = vi.fn();
+    const warn = vi.fn();
+    const session = createTalkRealtimeRelaySession({
+      context: {
+        broadcastToConnIds,
+        getRuntimeConfig: () => ({}),
+        logGateway: { warn },
+      } as never,
+      connId: "conn-throwing-sink",
+      eventSink: () => {
+        throw new Error("renderer gone");
+      },
+      provider,
+      providerConfig: {},
+      instructions: "brief",
+      tools: [],
+    });
+
+    sendTalkRealtimeRelayAudio({
+      relaySessionId: session.relaySessionId,
+      connId: "conn-throwing-sink",
+      audioBase64: Buffer.from([1, 2]).toString("base64"),
+    });
+
+    expect(warn).toHaveBeenCalledWith("talk realtime event sink failed: renderer gone");
+    expect(broadcastToConnIds).toHaveBeenCalledWith(
+      "talk.event",
+      expect.objectContaining({ relaySessionId: session.relaySessionId, type: "inputAudio" }),
+      new Set(["conn-throwing-sink"]),
+      { dropIfSlow: true },
+    );
+    expect(sendAudio).toHaveBeenCalledWith(Buffer.from([1, 2]));
+  });
+
   it("closes only realtime relays owned by the disconnected connection", async () => {
     const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]);
     const tempDir = tempDirs.make("openclaw-relay-disconnect-");
@@ -4184,5 +4260,38 @@ describe("talk realtime gateway relay", () => {
       outputEncoding: "pcm16",
     });
   });
+
+  it("keeps plugin quotas stable while cleaning up each consuming connection", () => {
+    const provider: RealtimeVoiceProviderPlugin = {
+      id: "relay-test",
+      label: "Relay Test",
+      isConfigured: () => true,
+      createBridge: () => makeRelayTransport(),
+    };
+    const context = {
+      broadcastToConnIds: vi.fn(),
+      logGateway: { warn: vi.fn() },
+    } as never;
+    const createSession = (connId: string) =>
+      createTalkRealtimeRelaySession({
+        context,
+        connId,
+        quotaOwnerId: "plugin:avatar:plugin-http:127.0.0.1",
+        provider,
+        providerConfig: {},
+        instructions: "brief",
+        tools: [],
+      });
+
+    const first = createSession("plugin:avatar:lifecycle-1");
+    createSession("plugin:avatar:lifecycle-2");
+    expect(() => createSession("plugin:avatar:lifecycle-3")).toThrow(
+      "Too many active realtime relay sessions for this connection",
+    );
+
+    cleanupTalkConnection("plugin:avatar:lifecycle-1", context.logGateway);
+    expect(relaySessions.has(first.relaySessionId)).toBe(false);
+    expect(() => createSession("plugin:avatar:lifecycle-3")).not.toThrow();
+  });
 });
 /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
diff --git a/src/gateway/talk-realtime-session-create.ts b/src/gateway/talk-realtime-session-create.ts
new file mode 100644
index 000000000000..03c7e64ef320
--- /dev/null
+++ b/src/gateway/talk-realtime-session-create.ts
@@ -0,0 +1,102 @@
+import { normalizeOptionalLowercaseString } from "@openclaw/normalization-core/string-coerce";
+import type { TalkSessionCreateParams } from "../../packages/gateway-protocol/src/index.js";
+import { buildAgentMainSessionKey } from "../routing/session-key.js";
+import { REALTIME_VOICE_AGENT_CONSULT_TOOL } from "../talk/agent-consult-tool.js";
+import { REALTIME_VOICE_AGENT_CONTROL_TOOL } from "../talk/agent-run-control-shared.js";
+import { resolveTalkSessionAgentId } from "../talk/agent-target.js";
+import { ensureClientVoiceAgentSessionEntry } from "../talk/client-voice-session.js";
+import { resolveConfiguredRealtimeVoiceProvider } from "../talk/provider-resolver.js";
+import {
+  buildRealtimeInstructions,
+  buildRealtimeVoiceLaunchOptions,
+  buildTalkRealtimeConfig,
+  resolveTalkRealtimeGatewayRelayLaunch,
+  resolveTalkRealtimeProviderInstructions,
+} from "./server-methods/talk-shared.js";
+import type { GatewayRequestContext } from "./server-methods/types.js";
+import type { TalkRealtimeRelayEventSink } from "./talk-realtime-relay-state.js";
+import { createTalkRealtimeRelaySession } from "./talk-realtime-relay.js";
+import { rememberUnifiedTalkSession } from "./talk-session-registry.js";
+
+type RealtimeTalkSessionRequest = Pick<
+  TalkSessionCreateParams,
+  "language" | "model" | "provider" | "sessionKey" | "voice"
+>;
+
+export class TalkRealtimeSessionRequestError extends Error {}
+
+export async function createGatewayRealtimeTalkSession(params: {
+  context: GatewayRequestContext;
+  ownerId: string;
+  agentId?: string;
+  quotaOwnerId?: string;
+  request: RealtimeTalkSessionRequest;
+  eventSink?: TalkRealtimeRelayEventSink;
+}) {
+  const runtimeConfig = params.context.getRuntimeConfig();
+  const realtimeConfig = buildTalkRealtimeConfig(runtimeConfig, params.request.provider);
+  const launchOptions = buildRealtimeVoiceLaunchOptions({
+    requested: params.request,
+    defaults: realtimeConfig,
+  });
+  const agentId =
+    params.agentId ?? resolveTalkSessionAgentId(runtimeConfig, params.request.sessionKey);
+  const resolution = resolveConfiguredRealtimeVoiceProvider({
+    configuredProviderId: realtimeConfig.provider,
+    providerConfigs: realtimeConfig.providers,
+    providerConfigOverrides: launchOptions.model ? { model: launchOptions.model } : {},
+    cfg: runtimeConfig,
+    agentId,
+    defaultModel: realtimeConfig.model,
+    surface: "gateway-relay",
+  });
+  const relayLaunch = resolveTalkRealtimeGatewayRelayLaunch({
+    ...resolution,
+    cfg: runtimeConfig,
+    launchOptions,
+    consultRouting: realtimeConfig.consultRouting,
+  });
+  if (relayLaunch.error) {
+    throw new TalkRealtimeSessionRequestError(relayLaunch.error);
+  }
+  const realtimeContext = await resolveTalkRealtimeProviderInstructions({
+    config: runtimeConfig,
+    agentId,
+    configuredInstructions: realtimeConfig.instructions,
+    sessionKey: params.request.sessionKey,
+    requireSessionKeyForProfile: true,
+    warn: (message) => params.context.logGateway.warn(`talk realtime context: ${message}`),
+  });
+  const sessionKey =
+    realtimeContext.requestedSessionKey ??
+    buildAgentMainSessionKey({ agentId: realtimeContext.agentId });
+  await ensureClientVoiceAgentSessionEntry({ agentId: realtimeContext.agentId, sessionKey });
+  const session = createTalkRealtimeRelaySession({
+    context: params.context,
+    connId: params.ownerId,
+    ...(params.quotaOwnerId ? { quotaOwnerId: params.quotaOwnerId } : {}),
+    ...(params.eventSink ? { eventSink: params.eventSink } : {}),
+    cfg: runtimeConfig,
+    provider: resolution.provider,
+    providerConfig: relayLaunch.providerConfig,
+    instructions: buildRealtimeInstructions(realtimeContext.instructions),
+    tools: [REALTIME_VOICE_AGENT_CONSULT_TOOL, REALTIME_VOICE_AGENT_CONTROL_TOOL],
+    model: launchOptions.model,
+    sessionKey,
+    voice: launchOptions.voice,
+    language: normalizeOptionalLowercaseString(params.request.language),
+    forceAgentConsultOnFinalTranscript: relayLaunch.forceAgentConsultOnFinalTranscript,
+  });
+  rememberUnifiedTalkSession(session.relaySessionId, {
+    kind: "realtime-relay",
+    connId: params.ownerId,
+    relaySessionId: session.relaySessionId,
+  });
+  return {
+    ...session,
+    sessionId: session.relaySessionId,
+    voiceSessionId: session.relaySessionId,
+    mode: "realtime" as const,
+    brain: "agent-consult" as const,
+  };
+}
diff --git a/src/talk/plugin-session.ts b/src/talk/plugin-session.ts
index 6ad5bb114285..b679c7c2ab02 100644
--- a/src/talk/plugin-session.ts
+++ b/src/talk/plugin-session.ts
@@ -5,7 +5,10 @@ import {
 
 export const PLUGIN_TALK_AUDIO_FORMAT: Readonly<
   Extract<RealtimeVoiceAudioFormat, { encoding: "pcm16" }>
-> = REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ;
+> = REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ as Extract<
+  RealtimeVoiceAudioFormat,
+  { encoding: "pcm16" }
+>;
 
 export type PluginTalkSessionEvent =
   | {
@@ -34,6 +37,7 @@ export type PluginTalkSessionEvent =
 
 export type OpenPluginTalkSessionParams = {
   sessionKey: string;
+  signal: AbortSignal;
   provider?: string;
   model?: string;
   voice?: string;