ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-24 19:35:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(plugin-sdk): bind Talk sessions to plugin lifetimes

Browse Source
This commit is contained in:
Dallin Romney
2026-08-11 22:31:07 +07:00
parent 3187e17696
commit bf609ba956
12 changed files with 454 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plugins/sdk-runtime.md
+5
View File
@@ -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"`).
src/gateway/server-methods/talk-session.ts
+20 -80
View File
@@ -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") {
src/gateway/talk-plugin-session.test.ts
+87 -15
View File
@@ -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,
});
});
});
src/gateway/talk-plugin-session.ts
+81 -29
View File
@@ -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 });
},
};
src/gateway/talk-realtime-relay-forced-consults.ts
+3 -3
View File
@@ -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,
src/gateway/talk-realtime-relay-operations.ts
+10 -10
View File
@@ -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,
src/gateway/talk-realtime-relay-provider-results.ts
+2 -2
View File
@@ -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,
src/gateway/talk-realtime-relay-session-create.ts
+15 -7
View File
@@ -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,
src/gateway/talk-realtime-relay-state.ts
+15 -5
View File
@@ -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(
src/gateway/talk-realtime-relay.test.ts
+109
View File
@@ -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. */
src/gateway/talk-realtime-session-create.ts
+102
View File
@@ -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,
};
}
src/talk/plugin-session.ts
+5 -1
View File
@@ -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;