mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix(openai): surface failed realtime response outcomes (#121177)
* fix(talk): type realtime response outcomes * fix(ci): align realtime outcome contracts * test(ui): identify completed realtime responses * chore(plugin-sdk): refresh realtime voice baselines
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
9dc5a9a152
commit
42a4d0a9b8
@@ -1 +1 @@
|
||||
{"contentHash":"607bfb12755ddfb5ade112771d24126d09aa47f5a93df3da6ee7a77e60bcebf2","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"}
|
||||
{"contentHash":"00b95749d23bbf126f7081ee8da4a303b2c984f181be7b68f8bf48b12b67e901","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"02d7c4b0ccf053d70fcdae12cd41e6028a025ac4464d9ad01996d825f5558c6f","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"}
|
||||
{"contentHash":"a806b806c516e82741cf2731d0d0375736eda65dcf130f0a66f3a631b6b1b482","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"78ece577895a6a5cc4a5bbe4e2791d72f28ce7f53a349b6faa6eaf1e3bc90c72","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"}
|
||||
{"contentHash":"60730746ea4ad9d8aef60a7f4d4aecd7cd28ae5d28b21e44646b3fe1398e6c95","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"dfaac94d7b6044e6b78ea3dd6d93113f2220bebbbbb094837618f1fb065f99ca","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"}
|
||||
{"contentHash":"3d0d58f29870cbb7889517e24b874ea5b63dfde6c15433c13414e8484b74d266","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"f7460f72ff844417e3ba3ca35fedb6b9d9f6f9d2cd1c4bb692bf3c7a8cd986fe","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"}
|
||||
{"contentHash":"f185748f1394f1d64653b21889954229bf0049e174526554e52c8e0bfc2c7dc5","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"6ddb91ab448c9471d4b96e86a687927ec237aa1fda9ac8d242caecf2863c12d4","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"}
|
||||
{"contentHash":"59cfee1296e172c295a550d4adbfff8776f10b73e07902394a68aa143efecabc","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"34f980205f902462ff8068e76142295460b47c6a57242d4355694a0b3b8d6244","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"}
|
||||
{"contentHash":"400ffb2b574275188fcbaeb2bc4fd6c5dd6cc1007c76223b1d63d2056b3ddd99","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"b06e4402306dbb88968f7f6e03ea05ac04d9b07be18c3d6bd17657ce50db7543","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"}
|
||||
{"contentHash":"589c2e34ab4910b95df172434a94e7a74838280c31f0955319fd696be3f3bc78","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"34565f9ed0312937609380d0f4b1ebfb4955a79d0b1fc3644d7c25f143bec5f4","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"}
|
||||
{"contentHash":"55b33df13c3362c01f13580f9903900075a2bb50b05cae56533a3ce9a5170639","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"141343dce0928e23c1863cc5a4305c1b281adea003beaf2e1853c10b553deb09","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"}
|
||||
{"contentHash":"61c67e6ebab5fc4d69226afa0d94f794e8d6798658dfaef60184506551b97a47","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"9c9caf0ad1b5ac92dd564c9a98b1f302855a5506f347a38da114804589bac187","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"}
|
||||
{"contentHash":"8d9f9721597b619189cbaa30747e10d2fb8adcf8e8404880b85683fb8fcebd47","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"79c5b931cceb7c1be3c95b9ca6e0362bdae5c6d6ee5e5aed97fcbdd9ea1062c1","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"}
|
||||
{"contentHash":"b25cfc5913de6cc44719995ec258226c7cb9fd1623afc277d75a5bc2e9870ccf","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"bc0d9ea4642d666045fe90464c8e5e34c0afc049d20d05823a5ddb4b52ae1c10","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"}
|
||||
{"contentHash":"0a01c88224f2d5dfd489f7478940032838e8246c9b00ff94fb6bcaf8d6bf84f5","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"9b6353909b35ab9bc36feb31037a3702e943e0c1749214799a928bf3c1a48035","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"}
|
||||
{"contentHash":"54687a6ab40ae684e91ee657a2188685a46df434232982807f5ebcadd941d3b8","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"contentHash":"b6c16456c26e0be9f02d0e420ee6054a48f529a9288c66e0e31e16b1367b6d5e","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"}
|
||||
{"contentHash":"92132ddbd26e3d81621f59004521f3875158b4704ee0b9e39f4798766f229702","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"contentHash":"1a2a81855860352fe59cf3162489fa1a93907c16b75d68460b610777e29120a0","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"}
|
||||
{"contentHash":"50ffb05c64c0f2898b3543900132261e152eda3999462154360efc3db7194359","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"}
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
import type {
|
||||
RealtimeVoiceBridgeEvent,
|
||||
RealtimeVoiceResponseOutcome,
|
||||
} from "openclaw/plugin-sdk/realtime-voice";
|
||||
import { vi } from "vitest";
|
||||
import { ChannelType } from "../internal/discord.js";
|
||||
import { createVoiceCaptureState } from "./capture-state.js";
|
||||
@@ -39,13 +43,14 @@ export type TestRealtimeSessionEntry = {
|
||||
};
|
||||
|
||||
export type TestRealtimeBridgeParams = {
|
||||
audioSink?: { sendAudio: (audio: Buffer) => void };
|
||||
audioSink: { sendAudio: (audio: Buffer) => void };
|
||||
autoRespondToAudio?: boolean;
|
||||
cfg?: unknown;
|
||||
instructions?: string;
|
||||
interruptResponseOnInputAudio?: boolean;
|
||||
onEvent?: (event: { detail?: string; direction: "client" | "server"; type: string }) => void;
|
||||
onEvent?: (event: RealtimeVoiceBridgeEvent) => void;
|
||||
onReady?: () => void;
|
||||
onResponseDone?: (outcome: RealtimeVoiceResponseOutcome) => void;
|
||||
onToolCall?: (
|
||||
event: { args: unknown; callId: string; itemId: string; name: string },
|
||||
session: unknown,
|
||||
|
||||
@@ -286,7 +286,48 @@ vi.mock("openclaw/plugin-sdk/realtime-voice", async () => {
|
||||
return {
|
||||
...harness,
|
||||
createBridge: (bridgeParams: Parameters<typeof harness.createBridge>[0]) =>
|
||||
createRealtimeVoiceBridgeSessionMock(bridgeParams),
|
||||
harness.createBridge({
|
||||
...bridgeParams,
|
||||
provider: {
|
||||
...bridgeParams.provider,
|
||||
label: bridgeParams.provider.label ?? "Test realtime provider",
|
||||
isConfigured: bridgeParams.provider.isConfigured ?? (() => true),
|
||||
createBridge: (request) => {
|
||||
createRealtimeVoiceBridgeSessionMock({
|
||||
...bridgeParams,
|
||||
audioSink: {
|
||||
...bridgeParams.audioSink,
|
||||
sendAudio: request.onAudio,
|
||||
clearAudio: request.onClearAudio,
|
||||
},
|
||||
onEvent: request.onEvent,
|
||||
onReady: request.onReady,
|
||||
onResponseDone: request.onResponseDone,
|
||||
onToolCall: bridgeParams.onToolCall,
|
||||
onTranscript: request.onTranscript,
|
||||
});
|
||||
return {
|
||||
supportsToolResultContinuation:
|
||||
realtimeSessionMock.bridge.supportsToolResultContinuation,
|
||||
supportsToolResultSuppression:
|
||||
realtimeSessionMock.bridge.supportsToolResultSuppression,
|
||||
acknowledgeMark: realtimeSessionMock.acknowledgeMark,
|
||||
close: realtimeSessionMock.close,
|
||||
connect: realtimeSessionMock.connect,
|
||||
handleBargeIn: realtimeSessionMock.handleBargeIn,
|
||||
isConnected: () => true,
|
||||
sendAudio: realtimeSessionMock.sendAudio,
|
||||
sendUserMessage: realtimeSessionMock.sendUserMessage,
|
||||
setMediaTimestamp: realtimeSessionMock.setMediaTimestamp,
|
||||
submitToolResult: (callId, result, options) =>
|
||||
options === undefined
|
||||
? realtimeSessionMock.submitToolResult(callId, result)
|
||||
: realtimeSessionMock.submitToolResult(callId, result, options),
|
||||
triggerGreeting: realtimeSessionMock.triggerGreeting,
|
||||
};
|
||||
},
|
||||
},
|
||||
}),
|
||||
flushOutput: (flush: () => void) => flush(),
|
||||
handleBargeIn: (
|
||||
options: Parameters<typeof harness.handleBargeIn>[0],
|
||||
@@ -2944,6 +2985,71 @@ describe("DiscordVoiceManager", () => {
|
||||
expect(player.play).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it.each([
|
||||
[
|
||||
{ status: "failed" as const, responseId: "response-1", message: "provider failed" },
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{
|
||||
status: "incomplete" as const,
|
||||
responseId: "response-1",
|
||||
reason: "max_output_tokens",
|
||||
message: "provider response incomplete",
|
||||
},
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{ status: "cancelled" as const, responseId: "response-1", reason: "client_cancelled" },
|
||||
"turn.cancelled",
|
||||
],
|
||||
])("retires each response once and plays a later response", async (outcome, terminalType) => {
|
||||
const { bridgeParams, entry, manager, player } = await createJoinedAgentProxyFixture();
|
||||
const realtime = entry.realtime as unknown as { harness: RealtimeVoiceSessionHarness };
|
||||
|
||||
bridgeParams.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.created",
|
||||
responseId: outcome.responseId,
|
||||
});
|
||||
bridgeParams.audioSink.sendAudio(Buffer.alloc(480));
|
||||
bridgeParams.onResponseDone?.(outcome);
|
||||
bridgeParams.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: outcome.responseId,
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
expect(
|
||||
realtime.harness.talk.recentEvents.filter((event) => event.type === terminalType),
|
||||
).toHaveLength(1);
|
||||
expect(manager.status()).toHaveLength(1);
|
||||
expect(realtimeSessionMock.close).not.toHaveBeenCalled();
|
||||
expect(player.stop).toHaveBeenCalledTimes(1);
|
||||
|
||||
bridgeParams.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.created",
|
||||
responseId: "response-2",
|
||||
});
|
||||
bridgeParams.audioSink.sendAudio(Buffer.alloc(480));
|
||||
bridgeParams.onResponseDone?.({ status: "completed", responseId: "response-2" });
|
||||
bridgeParams.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: "response-2",
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
expect(
|
||||
realtime.harness.talk.recentEvents.filter(
|
||||
(event) => event.type === "turn.ended" || event.type === "turn.cancelled",
|
||||
),
|
||||
).toHaveLength(2);
|
||||
expect(createAudioResourceMock).toHaveBeenCalledOnce();
|
||||
expect(player.play).toHaveBeenCalledOnce();
|
||||
expect(manager.status()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("discards prebuffered realtime output when the response is cancelled", async () => {
|
||||
const { bridgeParams, player } = await createJoinedAgentProxyFixture();
|
||||
|
||||
@@ -2955,10 +3061,9 @@ describe("DiscordVoiceManager", () => {
|
||||
expect(player.stop).toHaveBeenCalledWith(true);
|
||||
|
||||
bridgeParams?.audioSink?.sendAudio(Buffer.alloc(480));
|
||||
bridgeParams?.onEvent?.({
|
||||
detail: "response completed with status=cancelled",
|
||||
direction: "server",
|
||||
type: "response.done",
|
||||
bridgeParams?.onResponseDone?.({
|
||||
status: "cancelled",
|
||||
reason: "client_cancelled",
|
||||
});
|
||||
|
||||
expect(createAudioResourceMock).not.toHaveBeenCalled();
|
||||
|
||||
@@ -164,9 +164,6 @@ function formatRealtimeInterruptionLog(event: RealtimeVoiceBridgeEvent): string
|
||||
if (event.type === "response.cancelled") {
|
||||
return `discord voice: realtime model interrupt confirmed ${event.direction}:${event.type}${detail}`;
|
||||
}
|
||||
if (event.type === "response.done" && event.detail?.includes("status=cancelled")) {
|
||||
return `discord voice: realtime model interrupt confirmed ${event.direction}:${event.type}${detail}`;
|
||||
}
|
||||
if (event.type === "error" && event.detail === DISCORD_REALTIME_CANCELLATION_RACE_DETAIL) {
|
||||
return `discord voice: realtime model interrupt raced ${event.direction}:${event.type}${detail}`;
|
||||
}
|
||||
@@ -182,14 +179,6 @@ function formatRealtimeLifecycleLog(event: RealtimeVoiceBridgeEvent): string | u
|
||||
return `discord voice: realtime lifecycle ${event.direction}:${event.type}${detail}`;
|
||||
}
|
||||
|
||||
function isRealtimeResponseCancelled(event: RealtimeVoiceBridgeEvent): boolean {
|
||||
return (
|
||||
event.direction === "server" &&
|
||||
(event.type === "response.cancelled" ||
|
||||
(event.type === "response.done" && event.detail?.includes("status=cancelled") === true))
|
||||
);
|
||||
}
|
||||
|
||||
function isRealtimeResponseCancellationRace(event: RealtimeVoiceBridgeEvent): boolean {
|
||||
return (
|
||||
event.direction === "server" &&
|
||||
@@ -562,12 +551,9 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
|
||||
if (shouldLogRealtimeVerboseEvent(event)) {
|
||||
logVoiceVerbose(`realtime ${event.direction}:${event.type}${detail}`);
|
||||
}
|
||||
const responseEnded =
|
||||
event.direction === "server" &&
|
||||
(event.type === "response.done" || event.type === "response.cancelled");
|
||||
const responseCancellationRaced =
|
||||
this.outputBackpressure !== undefined && isRealtimeResponseCancellationRace(event);
|
||||
if (responseEnded || responseCancellationRaced) {
|
||||
if (responseCancellationRaced) {
|
||||
const outputBackpressured = this.outputBackpressure !== undefined;
|
||||
this.outputBackpressure = undefined;
|
||||
if (
|
||||
@@ -577,7 +563,7 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
|
||||
this.completeExactSpeechResponse(event.type);
|
||||
}
|
||||
this.finishOutputAudioStream(event.type, {
|
||||
playBuffered: responseEnded && !isRealtimeResponseCancelled(event),
|
||||
playBuffered: false,
|
||||
});
|
||||
}
|
||||
const interruptionLog = formatRealtimeInterruptionLog(event);
|
||||
@@ -589,6 +575,27 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
|
||||
logger.info(lifecycleLog);
|
||||
}
|
||||
},
|
||||
onResponseDone: (outcome) => {
|
||||
this.markProviderGenerationObserved();
|
||||
const outputBackpressured = this.outputBackpressure !== undefined;
|
||||
this.outputBackpressure = undefined;
|
||||
if (
|
||||
this.exactSpeechResponseActive &&
|
||||
(outputBackpressured || !this.exactSpeechAudioStarted)
|
||||
) {
|
||||
this.completeExactSpeechResponse(outcome.status);
|
||||
}
|
||||
this.finishOutputAudioStream(outcome.status, {
|
||||
playBuffered: outcome.status === "completed",
|
||||
});
|
||||
if (outcome.status === "cancelled") {
|
||||
logger.info(
|
||||
`discord voice: realtime model interrupt confirmed server:response.done status=cancelled${outcome.reason ? ` reason=${outcome.reason}` : ""}`,
|
||||
);
|
||||
} else if (outcome.status === "failed" || outcome.status === "incomplete") {
|
||||
this.logRealtimeError(outcome.message);
|
||||
}
|
||||
},
|
||||
onError: (error) => this.logRealtimeError(formatErrorMessage(error)),
|
||||
onClose: (reason) => {
|
||||
this.flushSuppressedRealtimeErrors();
|
||||
@@ -715,7 +722,9 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
|
||||
);
|
||||
this.handleBargeIn("active-speaker-audio");
|
||||
}
|
||||
this.bridge.sendAudio(realtimePcm);
|
||||
if (this.harness.recordInputAudio(realtimePcm)) {
|
||||
this.bridge.sendAudio(realtimePcm);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -800,7 +809,7 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
|
||||
if (this.exactSpeechResponseActive) {
|
||||
this.exactSpeechAudioStarted = true;
|
||||
}
|
||||
this.harness.outputActivity.markAudio({
|
||||
this.harness.recordOutputAudio(realtimePcm24kMono, {
|
||||
audioMs: pcm16MonoDurationMs(
|
||||
realtimePcm24kMono,
|
||||
REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ.sampleRateHz,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
// OpenAI tests cover the native realtime voice bridge against the live API.
|
||||
import { describe, expect, it } from "vitest";
|
||||
import WebSocket from "ws";
|
||||
import { buildOpenAIRealtimeVoiceProvider } from "./realtime-voice-provider.js";
|
||||
|
||||
const OPENAI_API_KEY = process.env.OPENAI_API_KEY?.trim() ?? "";
|
||||
@@ -7,6 +8,72 @@ const LIVE_ENABLED = OPENAI_API_KEY.length > 0 && process.env.OPENCLAW_LIVE_TEST
|
||||
const describeLive = LIVE_ENABLED ? describe : describe.skip;
|
||||
|
||||
describeLive("OpenAI realtime voice lifecycle live", () => {
|
||||
it("emits an incomplete response and then reuses the same session", async () => {
|
||||
const socket = new WebSocket("wss://api.openai.com/v1/realtime?model=gpt-realtime-2.1", {
|
||||
headers: { Authorization: `Bearer ${OPENAI_API_KEY}` },
|
||||
});
|
||||
const outcomes: Array<{ status?: string; reason?: string }> = [];
|
||||
const sendTurn = (text: string, maxOutputTokens: number) => {
|
||||
socket.send(
|
||||
JSON.stringify({
|
||||
type: "conversation.item.create",
|
||||
item: { type: "message", role: "user", content: [{ type: "input_text", text }] },
|
||||
}),
|
||||
);
|
||||
socket.send(
|
||||
JSON.stringify({
|
||||
type: "response.create",
|
||||
response: { output_modalities: ["text"], max_output_tokens: maxOutputTokens },
|
||||
}),
|
||||
);
|
||||
};
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timeout = setTimeout(
|
||||
() => reject(new Error("Realtime live probe timed out")),
|
||||
45_000,
|
||||
);
|
||||
socket.on("message", (data) => {
|
||||
const payload = Buffer.isBuffer(data)
|
||||
? data
|
||||
: Array.isArray(data)
|
||||
? Buffer.concat(data)
|
||||
: Buffer.from(data);
|
||||
const event = JSON.parse(payload.toString("utf8")) as {
|
||||
type?: string;
|
||||
response?: { status?: string; status_details?: { reason?: string } | null };
|
||||
error?: { message?: string };
|
||||
};
|
||||
if (event.type === "error") {
|
||||
clearTimeout(timeout);
|
||||
reject(new Error(event.error?.message ?? "Realtime API error"));
|
||||
} else if (event.type === "session.created") {
|
||||
sendTurn("Write a detailed paragraph about ocean tides.", 1);
|
||||
} else if (event.type === "response.done") {
|
||||
outcomes.push({
|
||||
status: event.response?.status,
|
||||
reason: event.response?.status_details?.reason,
|
||||
});
|
||||
if (outcomes.length === 1) {
|
||||
sendTurn("Reply with exactly one word: ok", 100);
|
||||
} else {
|
||||
clearTimeout(timeout);
|
||||
resolve();
|
||||
}
|
||||
}
|
||||
});
|
||||
socket.on("error", reject);
|
||||
});
|
||||
} finally {
|
||||
socket.close();
|
||||
}
|
||||
|
||||
expect(outcomes).toEqual([
|
||||
{ status: "incomplete", reason: "max_output_tokens" },
|
||||
{ status: "completed", reason: undefined },
|
||||
]);
|
||||
}, 60_000);
|
||||
|
||||
it("reuses a bridge after a terminal close", async () => {
|
||||
let closeCount = 0;
|
||||
let readyCount = 0;
|
||||
|
||||
@@ -31,6 +31,7 @@ import type {
|
||||
import {
|
||||
REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ,
|
||||
REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ,
|
||||
normalizeRealtimeVoiceResponseOutcome,
|
||||
RealtimeVoiceSessionLifecycle,
|
||||
} from "openclaw/plugin-sdk/realtime-voice";
|
||||
import { sleepWithAbort, warn } from "openclaw/plugin-sdk/runtime-env";
|
||||
@@ -1430,6 +1431,18 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge {
|
||||
this.ws?.close(1000, "max-duration rotation");
|
||||
return;
|
||||
}
|
||||
if (event.type === "response.done") {
|
||||
this.handleResponseDone(event, connection, emitServerEvent);
|
||||
return;
|
||||
}
|
||||
if (event.type === "response.cancelled") {
|
||||
try {
|
||||
emitServerEvent();
|
||||
} finally {
|
||||
this.releaseResponseState();
|
||||
}
|
||||
return;
|
||||
}
|
||||
emitServerEvent();
|
||||
switch (event.type) {
|
||||
case "session.created":
|
||||
@@ -1521,29 +1534,6 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge {
|
||||
// is the sole execution boundary.
|
||||
return;
|
||||
|
||||
case "response.cancelled":
|
||||
case "response.done":
|
||||
if (this.handleCompletedResponse(event, connection)) {
|
||||
return;
|
||||
}
|
||||
this.responseActive = false;
|
||||
this.responseCreateInFlight = false;
|
||||
this.manualResponseCreateEventId = null;
|
||||
this.responseCancelInFlight = false;
|
||||
this.manualResponseCancelEventId = null;
|
||||
if (this.standaloneSpeechActive) {
|
||||
this.standaloneSpeechActive = false;
|
||||
this.standaloneSpeechEventId = null;
|
||||
}
|
||||
if (this.standaloneSpeechQueue.length > 0) {
|
||||
this.flushStandaloneSpeech();
|
||||
} else if (this.responseCreatePending) {
|
||||
this.flushPendingResponseCreate();
|
||||
} else {
|
||||
this.restoreAutoRespondAfterManualResponse();
|
||||
}
|
||||
return;
|
||||
|
||||
case "error": {
|
||||
const detail = readRealtimeErrorDetail(event.error);
|
||||
const rejectedEventId = readRealtimeErrorEventId(event.error);
|
||||
@@ -1605,6 +1595,28 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge {
|
||||
}
|
||||
}
|
||||
|
||||
private releaseResponseState(options: { drain?: boolean } = {}): void {
|
||||
this.responseActive = false;
|
||||
this.responseCreateInFlight = false;
|
||||
this.manualResponseCreateEventId = null;
|
||||
this.responseCancelInFlight = false;
|
||||
this.manualResponseCancelEventId = null;
|
||||
if (this.standaloneSpeechActive) {
|
||||
this.standaloneSpeechActive = false;
|
||||
this.standaloneSpeechEventId = null;
|
||||
}
|
||||
if (options.drain === false) {
|
||||
return;
|
||||
}
|
||||
if (this.standaloneSpeechQueue.length > 0) {
|
||||
this.flushStandaloneSpeech();
|
||||
} else if (this.responseCreatePending) {
|
||||
this.flushPendingResponseCreate();
|
||||
} else {
|
||||
this.restoreAutoRespondAfterManualResponse();
|
||||
}
|
||||
}
|
||||
|
||||
handleBargeIn(options?: RealtimeVoiceBargeInOptions): void {
|
||||
const assistantItemId = this.lastAssistantItemId;
|
||||
const responseStartTimestamp = this.responseStartTimestamp;
|
||||
@@ -1751,6 +1763,47 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge {
|
||||
return false;
|
||||
}
|
||||
|
||||
private handleResponseDone(
|
||||
event: RealtimeEvent,
|
||||
connection: RealtimeVoiceSessionConnection,
|
||||
emitServerEvent: () => void,
|
||||
): void {
|
||||
const outcome = normalizeRealtimeVoiceResponseOutcome({
|
||||
providerLabel: "OpenAI realtime voice",
|
||||
response: event.response,
|
||||
responseId: event.response_id,
|
||||
});
|
||||
let callbackError: unknown;
|
||||
let providerTerminated = false;
|
||||
const invoke = (callback: () => void) => {
|
||||
try {
|
||||
callback();
|
||||
} catch (error) {
|
||||
callbackError ??= error;
|
||||
}
|
||||
};
|
||||
try {
|
||||
invoke(() => this.config.onResponseDone?.(outcome));
|
||||
invoke(emitServerEvent);
|
||||
invoke(() => {
|
||||
providerTerminated = this.handleCompletedResponse(event, connection);
|
||||
});
|
||||
} finally {
|
||||
// response.done owns response state regardless of observer success. A fatal tool
|
||||
// boundary still clears state, but must not start queued work on a closing socket.
|
||||
const canDrain =
|
||||
!providerTerminated &&
|
||||
this.lifecycle.acceptsEvents(connection) &&
|
||||
this.ws?.readyState === WebSocket.OPEN;
|
||||
this.releaseResponseState({ drain: canDrain });
|
||||
}
|
||||
if (callbackError) {
|
||||
throw callbackError instanceof Error
|
||||
? callbackError
|
||||
: new Error("OpenAI realtime response callback failed", { cause: callbackError });
|
||||
}
|
||||
}
|
||||
|
||||
private rejectToolCallArguments(params: {
|
||||
itemId?: string;
|
||||
callId: string;
|
||||
@@ -2185,6 +2238,7 @@ async function createOpenAIRealtimeBrowserSession(
|
||||
onAudio: () => undefined,
|
||||
onClearAudio: () => undefined,
|
||||
onEvent: gatewayControl.onEvent,
|
||||
onResponseDone: gatewayControl.onResponseDone,
|
||||
onTranscript: gatewayControl.onTranscript,
|
||||
onToolCall: gatewayControl.onToolCall,
|
||||
onReady: gatewayControl.onReady,
|
||||
|
||||
@@ -0,0 +1,275 @@
|
||||
import { createServer } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import type { RealtimeVoiceResponseOutcome } from "openclaw/plugin-sdk/realtime-voice";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type WebSocket from "ws";
|
||||
import { WebSocketServer } from "ws";
|
||||
import { buildOpenAIRealtimeVoiceProvider } from "./realtime-voice-provider.js";
|
||||
|
||||
type CapturedOutcome = {
|
||||
clientEvents: string[];
|
||||
errors: string[];
|
||||
events: string[];
|
||||
outcomes: RealtimeVoiceResponseOutcome[];
|
||||
tools: Array<{ itemId: string; callId: string; name: string; args: unknown }>;
|
||||
connected: boolean;
|
||||
};
|
||||
|
||||
function signal() {
|
||||
let resolve = () => {};
|
||||
const promise = new Promise<void>((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
async function waitFor(promise: Promise<void>, label: string): Promise<void> {
|
||||
let timeout: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
promise,
|
||||
new Promise<never>((_, reject) => {
|
||||
timeout = setTimeout(() => reject(new Error(`timed out waiting for ${label}`)), 2_000);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
async function capture(
|
||||
terminalEvent: Record<string, unknown>,
|
||||
options: { completeFollowup?: boolean; queueFollowup?: boolean; throwCallback?: boolean } = {},
|
||||
): Promise<CapturedOutcome> {
|
||||
const captured: CapturedOutcome = {
|
||||
clientEvents: [],
|
||||
errors: [],
|
||||
events: [],
|
||||
outcomes: [],
|
||||
tools: [],
|
||||
connected: false,
|
||||
};
|
||||
const responseCreated = signal();
|
||||
const terminalProcessed = signal();
|
||||
const followupCreated = signal();
|
||||
const followupCompleted = signal();
|
||||
const server = createServer();
|
||||
const sockets = new Set<WebSocket>();
|
||||
const wss = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 });
|
||||
server.on("upgrade", (request, socket, head) => {
|
||||
wss.handleUpgrade(request, socket, head, (ws) => {
|
||||
sockets.add(ws);
|
||||
ws.on("message", (message) => {
|
||||
const event = JSON.parse(Buffer.from(message as Buffer).toString("utf8")) as {
|
||||
type?: string;
|
||||
};
|
||||
if (!event.type) {
|
||||
return;
|
||||
}
|
||||
captured.clientEvents.push(event.type);
|
||||
if (event.type === "session.update" && captured.clientEvents.length === 1) {
|
||||
ws.send(JSON.stringify({ type: "session.updated" }));
|
||||
}
|
||||
if (event.type === "response.create") {
|
||||
followupCreated.resolve();
|
||||
if (options.completeFollowup) {
|
||||
ws.send(JSON.stringify({ type: "response.created", response: { id: "response_2" } }));
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: "response.done",
|
||||
response: { id: "response_2", status: "completed", output: [] },
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const port = (server.address() as AddressInfo).port;
|
||||
const bridge = buildOpenAIRealtimeVoiceProvider().createBridge({
|
||||
providerConfig: { apiKey: "fixture-value", azureEndpoint: `http://127.0.0.1:${port}` },
|
||||
onAudio() {},
|
||||
onClearAudio() {},
|
||||
onError: (error) => captured.errors.push(error.message),
|
||||
onResponseDone: (outcome) => {
|
||||
captured.outcomes.push(outcome);
|
||||
captured.events.push(`outcome:${outcome.status}`);
|
||||
if (outcome.responseId === "response_2") {
|
||||
followupCompleted.resolve();
|
||||
}
|
||||
if (options.throwCallback && outcome.responseId === "response_1") {
|
||||
throw new Error("consumer callback failed");
|
||||
}
|
||||
},
|
||||
onToolCall: (tool) => captured.tools.push(tool),
|
||||
onEvent: (event) => {
|
||||
captured.events.push(`${event.direction}:${event.type}`);
|
||||
if (event.direction === "server" && event.type === "response.created") {
|
||||
responseCreated.resolve();
|
||||
}
|
||||
if (event.direction === "server" && event.type === terminalEvent.type) {
|
||||
queueMicrotask(terminalProcessed.resolve);
|
||||
}
|
||||
},
|
||||
});
|
||||
try {
|
||||
await bridge.connect();
|
||||
const socket = [...sockets][0];
|
||||
if (!socket) {
|
||||
throw new Error("expected a connected fixture socket");
|
||||
}
|
||||
socket.send(JSON.stringify({ type: "response.created", response: { id: "response_1" } }));
|
||||
await waitFor(responseCreated.promise, "response.created");
|
||||
if (options.queueFollowup) {
|
||||
bridge.sendUserMessage?.("Continue after the terminal response.");
|
||||
}
|
||||
socket.send(JSON.stringify(terminalEvent));
|
||||
await waitFor(terminalProcessed.promise, "terminal response");
|
||||
if (options.queueFollowup) {
|
||||
await waitFor(followupCreated.promise, "queued response.create");
|
||||
}
|
||||
if (options.completeFollowup) {
|
||||
await waitFor(followupCompleted.promise, "completed follow-up");
|
||||
}
|
||||
captured.connected = bridge.isConnected();
|
||||
return captured;
|
||||
} finally {
|
||||
bridge.close();
|
||||
for (const socket of sockets) {
|
||||
socket.terminate();
|
||||
}
|
||||
await new Promise<void>((resolve) => {
|
||||
wss.close(() => resolve());
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const completedTool = {
|
||||
id: "item_tool",
|
||||
type: "function_call",
|
||||
status: "completed",
|
||||
call_id: "call_tool",
|
||||
name: "lookup_weather",
|
||||
arguments: JSON.stringify({ city: "Paris" }),
|
||||
};
|
||||
|
||||
describe("OpenAI realtime terminal response ownership", () => {
|
||||
it.each([
|
||||
{
|
||||
response: { status: "completed", output: [] },
|
||||
expected: { responseId: "response_1", status: "completed" },
|
||||
},
|
||||
{
|
||||
response: {
|
||||
status: "cancelled",
|
||||
status_details: { reason: "client_cancelled" },
|
||||
output: [completedTool],
|
||||
},
|
||||
expected: { responseId: "response_1", status: "cancelled", reason: "client_cancelled" },
|
||||
},
|
||||
{
|
||||
response: {
|
||||
status: "failed",
|
||||
status_details: { error: { type: "server_error", code: "rate_limit_exceeded" } },
|
||||
output: [completedTool],
|
||||
},
|
||||
expected: {
|
||||
responseId: "response_1",
|
||||
status: "failed",
|
||||
error: { type: "server_error", code: "rate_limit_exceeded" },
|
||||
message: "OpenAI realtime voice response failed: rate_limit_exceeded",
|
||||
},
|
||||
},
|
||||
{
|
||||
response: {
|
||||
status: "incomplete",
|
||||
status_details: { reason: "max_output_tokens" },
|
||||
output: [completedTool],
|
||||
},
|
||||
expected: {
|
||||
responseId: "response_1",
|
||||
status: "incomplete",
|
||||
reason: "max_output_tokens",
|
||||
message: "OpenAI realtime voice response incomplete: max_output_tokens",
|
||||
},
|
||||
},
|
||||
{
|
||||
response: { output: [completedTool] },
|
||||
expected: {
|
||||
responseId: "response_1",
|
||||
status: "failed",
|
||||
reason: "invalid_response_status",
|
||||
error: { type: "invalid_response_status", message: "missing terminal status" },
|
||||
message: "OpenAI realtime voice response failed: missing terminal status",
|
||||
},
|
||||
},
|
||||
{
|
||||
response: { status: "in_progress", output: [completedTool] },
|
||||
expected: {
|
||||
responseId: "response_1",
|
||||
status: "failed",
|
||||
reason: "invalid_response_status",
|
||||
error: { type: "invalid_response_status", message: "invalid status in_progress" },
|
||||
message: "OpenAI realtime voice response failed: invalid status in_progress",
|
||||
},
|
||||
},
|
||||
])(
|
||||
"normalizes $response.status without closing the reusable socket",
|
||||
async ({ response, expected }) => {
|
||||
const captured = await capture(
|
||||
{ type: "response.done", response: { id: "response_1", ...response } },
|
||||
{ queueFollowup: true },
|
||||
);
|
||||
|
||||
expect(captured.errors).toEqual([]);
|
||||
expect(captured.outcomes).toEqual([expected]);
|
||||
expect(captured.tools).toEqual([]);
|
||||
expect(captured.clientEvents.filter((type) => type === "response.create")).toHaveLength(1);
|
||||
expect(captured.connected).toBe(true);
|
||||
expect(captured.events.indexOf(`outcome:${expected.status}`)).toBeLessThan(
|
||||
captured.events.indexOf("server:response.done"),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("executes terminal tool calls only for completed responses", async () => {
|
||||
const captured = await capture({
|
||||
type: "response.done",
|
||||
response: { id: "response_1", status: "completed", output: [completedTool] },
|
||||
});
|
||||
|
||||
expect(captured.tools).toEqual([
|
||||
{
|
||||
itemId: "item_tool",
|
||||
callId: "call_tool",
|
||||
name: "lookup_weather",
|
||||
args: { city: "Paris" },
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("drains a queued follow-up when the terminal consumer throws", async () => {
|
||||
const captured = await capture(
|
||||
{ type: "response.done", response: { id: "response_1", status: "failed", output: [] } },
|
||||
{ completeFollowup: true, queueFollowup: true, throwCallback: true },
|
||||
);
|
||||
|
||||
expect(captured.errors).toEqual([]);
|
||||
expect(captured.outcomes).toEqual([
|
||||
{
|
||||
responseId: "response_1",
|
||||
status: "failed",
|
||||
message: "OpenAI realtime voice response failed",
|
||||
},
|
||||
{ responseId: "response_2", status: "completed" },
|
||||
]);
|
||||
expect(captured.clientEvents.filter((type) => type === "response.create")).toHaveLength(1);
|
||||
expect(captured.connected).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -333,6 +333,68 @@ function requireCancelledTurn(call: CallRecord): RecentTalkEvent & { turnId: str
|
||||
}
|
||||
|
||||
describe("RealtimeCallHandler path routing", () => {
|
||||
it.each([
|
||||
[{ status: "completed" as const, responseId: "response-1" }, "turn.ended"],
|
||||
[
|
||||
{ status: "failed" as const, responseId: "response-1", message: "provider failed" },
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{
|
||||
status: "incomplete" as const,
|
||||
responseId: "response-1",
|
||||
reason: "max_output_tokens",
|
||||
message: "provider response incomplete",
|
||||
},
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{ status: "cancelled" as const, responseId: "response-1", reason: "client_cancelled" },
|
||||
"turn.cancelled",
|
||||
],
|
||||
])("finishes each telephony turn without closing the call", async (outcome, terminalType) => {
|
||||
await withBargeInHarness(
|
||||
{ providerCallId: `CA-response-${outcome.status}` },
|
||||
async ({ callbacks, call, ws }) => {
|
||||
callbacks.onTranscript?.("user", "first turn", true);
|
||||
callbacks.onAudio(Buffer.from([1]));
|
||||
callbacks.onResponseDone?.(outcome);
|
||||
callbacks.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: outcome.responseId,
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
const firstEvents = recentTalkEvents(call);
|
||||
expect(firstEvents.filter((event) => event.type === terminalType)).toHaveLength(1);
|
||||
expect(firstEvents.filter((event) => event.type === "output.audio.done")).toHaveLength(1);
|
||||
expect(firstEvents.filter((event) => event.type === "session.error")).toHaveLength(
|
||||
outcome.status === "failed" || outcome.status === "incomplete" ? 1 : 0,
|
||||
);
|
||||
expect(ws.readyState).toBe(WebSocket.OPEN);
|
||||
|
||||
callbacks.onEvent?.({ direction: "server", type: "input_audio_buffer.speech_started" });
|
||||
callbacks.onTranscript?.("user", "later turn", true);
|
||||
callbacks.onAudio(Buffer.from([2]));
|
||||
callbacks.onResponseDone?.({ status: "completed", responseId: "response-2" });
|
||||
callbacks.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: "response-2",
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
const finalEvents = recentTalkEvents(call);
|
||||
expect(
|
||||
finalEvents.filter(
|
||||
(event) => event.type === "turn.ended" || event.type === "turn.cancelled",
|
||||
),
|
||||
).toHaveLength(2);
|
||||
expect(finalEvents.filter((event) => event.type === "output.audio.done")).toHaveLength(2);
|
||||
expect(ws.readyState).toBe(WebSocket.OPEN);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("uses the request host and stream path in TwiML", () => {
|
||||
const handler = makeHandler();
|
||||
const payload = handler.buildTwiMLPayload(makeRequest("/voice/webhook", "gateway.ts.net"));
|
||||
|
||||
@@ -977,11 +977,6 @@ export class RealtimeCallHandler {
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (event.type === "response.done") {
|
||||
harness.finishOutputAudio("response.done");
|
||||
harness.endTurn("response.done");
|
||||
return;
|
||||
}
|
||||
if (event.type === "error") {
|
||||
harness.emit({
|
||||
type: "session.error",
|
||||
@@ -990,6 +985,11 @@ export class RealtimeCallHandler {
|
||||
});
|
||||
}
|
||||
},
|
||||
onResponseDone: (outcome) => {
|
||||
if (outcome.status === "failed" || outcome.status === "incomplete") {
|
||||
console.warn(`[voice-call] realtime response ${outcome.status}: ${outcome.message}`);
|
||||
}
|
||||
},
|
||||
onReady: () => {
|
||||
harness.emit({
|
||||
type: "session.ready",
|
||||
|
||||
@@ -308,6 +308,9 @@ function createLazyXaiRealtimeVoiceBridge(
|
||||
...(req.onEvent
|
||||
? { onEvent: guardProviderCallback(loadGeneration, req.onEvent) }
|
||||
: {}),
|
||||
...(req.onResponseDone
|
||||
? { onResponseDone: guardProviderCallback(loadGeneration, req.onResponseDone) }
|
||||
: {}),
|
||||
...(req.onToolCall
|
||||
? { onToolCall: guardProviderCallback(loadGeneration, req.onToolCall) }
|
||||
: {}),
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import { canonicalizeBase64 } from "openclaw/plugin-sdk/media-runtime";
|
||||
import type { RealtimeVoiceSessionConnection } from "openclaw/plugin-sdk/realtime-voice";
|
||||
import {
|
||||
normalizeRealtimeVoiceResponseOutcome,
|
||||
type RealtimeVoiceSessionConnection,
|
||||
} from "openclaw/plugin-sdk/realtime-voice";
|
||||
import { isRecord, normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import {
|
||||
XAI_REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX,
|
||||
@@ -14,13 +17,14 @@ export class XaiRealtimeMalformedAudioError extends Error {}
|
||||
export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
private assistantTranscriptBuffer = "";
|
||||
private assistantTranscriptFinalized = false;
|
||||
private finalizedToolCallItems = new Set<string>();
|
||||
private inputTranscriptReplacements = new Map<string, string>();
|
||||
|
||||
protected abstract acceptsEvent(connection: RealtimeVoiceSessionConnection): boolean;
|
||||
protected abstract onSessionUpdated(connection: RealtimeVoiceSessionConnection): void;
|
||||
|
||||
protected handleEvent(event: XaiRealtimeEvent, connection: RealtimeVoiceSessionConnection): void {
|
||||
this.config.onEvent?.({
|
||||
const bridgeEvent = {
|
||||
direction: "server",
|
||||
type: event.type,
|
||||
detail: this.describeServerEvent(event),
|
||||
@@ -28,7 +32,11 @@ export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
...((event.response_id ?? event.response?.id)
|
||||
? { responseId: event.response_id ?? event.response?.id }
|
||||
: {}),
|
||||
});
|
||||
} as const;
|
||||
const emitBridgeEvent = () => this.config.onEvent?.(bridgeEvent);
|
||||
if (event.type !== "response.done" || !this.acceptsEvent(connection)) {
|
||||
emitBridgeEvent();
|
||||
}
|
||||
if (!this.acceptsEvent(connection)) {
|
||||
return;
|
||||
}
|
||||
@@ -51,6 +59,8 @@ export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
return;
|
||||
}
|
||||
if (event.type === "conversation.item.created") {
|
||||
// Session resumption replays already-finalized conversation items without
|
||||
// another response.done; deliver that completed history at its replay boundary.
|
||||
this.emitCompletedToolCall(item, event);
|
||||
}
|
||||
return;
|
||||
@@ -130,41 +140,64 @@ export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
const output = Array.isArray(event.response?.output)
|
||||
? event.response.output.filter(isRecord)
|
||||
: [];
|
||||
const outcome = normalizeRealtimeVoiceResponseOutcome({
|
||||
providerLabel: "xAI realtime voice",
|
||||
response: event.response,
|
||||
responseId: event.response_id,
|
||||
});
|
||||
let callbackError: unknown;
|
||||
const invoke = (callback: () => void) => {
|
||||
try {
|
||||
callback();
|
||||
} catch (error) {
|
||||
callbackError ??= error;
|
||||
}
|
||||
};
|
||||
try {
|
||||
if (status === undefined || status === "completed") {
|
||||
for (const item of output) {
|
||||
this.emitCompletedToolCall(item, event);
|
||||
invoke(() => this.config.onResponseDone?.(outcome));
|
||||
invoke(emitBridgeEvent);
|
||||
invoke(() => {
|
||||
if (status === "completed") {
|
||||
for (const [itemId, toolCall] of this.toolCallBuffers) {
|
||||
this.emitToolCallOnce({
|
||||
itemId,
|
||||
callId: toolCall.callId,
|
||||
name: toolCall.name,
|
||||
rawArgs: toolCall.args,
|
||||
});
|
||||
}
|
||||
for (const item of output) {
|
||||
this.emitCompletedToolCall(item, event);
|
||||
}
|
||||
}
|
||||
}
|
||||
const terminalTranscript = output
|
||||
.filter((item) => item.type === "message" && item.role === "assistant")
|
||||
.flatMap((item) => (Array.isArray(item.content) ? item.content.filter(isRecord) : []))
|
||||
.map((content) =>
|
||||
typeof content.transcript === "string"
|
||||
? content.transcript
|
||||
: typeof content.text === "string"
|
||||
? content.text
|
||||
: "",
|
||||
)
|
||||
.join("");
|
||||
this.flushAssistantTranscript(terminalTranscript);
|
||||
if (status === "failed" || status === "incomplete") {
|
||||
const details = event.response?.status_details;
|
||||
const error = isRecord(details) ? details.error : undefined;
|
||||
const reason = isRecord(details) ? normalizeOptionalString(details.reason) : undefined;
|
||||
const message = error
|
||||
? readXaiRealtimeErrorDetail(error)
|
||||
: `xAI realtime voice response ${status}${reason ? `: ${reason}` : ""}`;
|
||||
this.config.onError?.(new Error(message));
|
||||
}
|
||||
const terminalTranscript = output
|
||||
.filter((item) => item.type === "message" && item.role === "assistant")
|
||||
.flatMap((item) => (Array.isArray(item.content) ? item.content.filter(isRecord) : []))
|
||||
.map((content) =>
|
||||
typeof content.transcript === "string"
|
||||
? content.transcript
|
||||
: typeof content.text === "string"
|
||||
? content.text
|
||||
: "",
|
||||
)
|
||||
.join("");
|
||||
this.flushAssistantTranscript(terminalTranscript);
|
||||
});
|
||||
} finally {
|
||||
// Keep the response active through terminal tool discovery: callbacks can
|
||||
// submit results synchronously and must not start the next response early.
|
||||
this.responseActive = false;
|
||||
this.responseCreateInFlight = false;
|
||||
this.responseCancelInFlight = false;
|
||||
this.toolCallBuffers.clear();
|
||||
this.finalizedToolCallItems.clear();
|
||||
this.flushPendingResponseCreate();
|
||||
}
|
||||
if (callbackError) {
|
||||
throw callbackError instanceof Error
|
||||
? callbackError
|
||||
: new Error("xAI realtime response callback failed", { cause: callbackError });
|
||||
}
|
||||
return;
|
||||
}
|
||||
case "response.function_call_arguments.delta": {
|
||||
@@ -183,19 +216,25 @@ export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
}
|
||||
case "response.function_call_arguments.done": {
|
||||
const key = event.item_id ?? "unknown";
|
||||
if (this.finalizedToolCallItems.has(key)) {
|
||||
return;
|
||||
}
|
||||
const buffered = this.toolCallBuffers.get(key);
|
||||
// xAI's documented Function Call Flow requires executing finalized
|
||||
// arguments immediately so tool results can continue the response.
|
||||
this.emitToolCallOnce({
|
||||
itemId: event.item_id,
|
||||
callId: buffered?.callId || event.call_id,
|
||||
name: buffered?.name || event.name,
|
||||
// The done payload owns the final JSON; streamed chunks may be stale or incomplete.
|
||||
rawArgs: event.arguments ?? buffered?.args,
|
||||
});
|
||||
this.toolCallBuffers.delete(key);
|
||||
// Keep finalized arguments for diagnostics only. response.done with a completed
|
||||
// response is the authoritative execution boundary for provider tool calls.
|
||||
if (event.item_id) {
|
||||
this.finalizedToolCallItems.add(event.item_id);
|
||||
this.toolCallBuffers.set(event.item_id, {
|
||||
name: buffered?.name || event.name || "",
|
||||
callId: buffered?.callId || event.call_id || "",
|
||||
args: event.arguments ?? buffered?.args ?? "",
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
case "response.output_item.done":
|
||||
this.bufferCompletedToolCall(event.item, event);
|
||||
return;
|
||||
case "error":
|
||||
this.handleErrorEvent(event.error);
|
||||
default:
|
||||
@@ -204,6 +243,7 @@ export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
|
||||
protected resetInputTranscripts(): void {
|
||||
this.inputTranscriptReplacements.clear();
|
||||
this.finalizedToolCallItems.clear();
|
||||
}
|
||||
|
||||
private emitCompletedToolCall(item: XaiRealtimeEvent["item"], event: XaiRealtimeEvent): void {
|
||||
@@ -219,6 +259,21 @@ export abstract class XaiRealtimeVoiceEvents extends XaiRealtimeVoiceProtocol {
|
||||
}
|
||||
}
|
||||
|
||||
private bufferCompletedToolCall(item: XaiRealtimeEvent["item"], event: XaiRealtimeEvent): void {
|
||||
if (item?.type !== "function_call" || (item.status && item.status !== "completed")) {
|
||||
return;
|
||||
}
|
||||
const itemId = item.id ?? event.item_id;
|
||||
if (!itemId) {
|
||||
return;
|
||||
}
|
||||
this.toolCallBuffers.set(itemId, {
|
||||
name: item.name ?? "",
|
||||
callId: item.call_id ?? "",
|
||||
args: item.arguments ?? "",
|
||||
});
|
||||
}
|
||||
|
||||
private appendAssistantTranscriptDelta(delta: string): void {
|
||||
if (this.assistantTranscriptFinalized) {
|
||||
this.assistantTranscriptBuffer = "";
|
||||
|
||||
@@ -1055,6 +1055,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
call_id: "call_1",
|
||||
arguments: JSON.stringify({ question: "delegate this" }),
|
||||
});
|
||||
socket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
expect(onToolCall).toHaveBeenCalledTimes(1);
|
||||
expect(onToolCall).toHaveBeenCalledWith({
|
||||
@@ -1107,6 +1108,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
name: "lookup_weather",
|
||||
arguments: finalArguments,
|
||||
});
|
||||
socket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
expect(onToolCall).toHaveBeenCalledWith({
|
||||
itemId: "item_tool_1",
|
||||
@@ -1161,6 +1163,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
socket.emitServer(event);
|
||||
}
|
||||
socket.emitServer(invalidEvents[0]);
|
||||
socket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
expect(onToolCall).not.toHaveBeenCalled();
|
||||
expect(
|
||||
@@ -1197,7 +1200,9 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
},
|
||||
})),
|
||||
);
|
||||
expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([]);
|
||||
expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([
|
||||
{ type: "response.create" },
|
||||
]);
|
||||
|
||||
socket.emitServer({ type: "response.done" });
|
||||
expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([
|
||||
@@ -1238,6 +1243,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
arguments: rawArgs,
|
||||
});
|
||||
}
|
||||
socket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
expect(onToolCall).not.toHaveBeenCalled();
|
||||
expect(
|
||||
@@ -1386,6 +1392,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
arguments: JSON.stringify({ question: callId }),
|
||||
});
|
||||
}
|
||||
socket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
await bridge.submitToolResult("call_1", { text: "first" });
|
||||
expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([]);
|
||||
@@ -1534,6 +1541,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
arguments: JSON.stringify({ question: callId }),
|
||||
});
|
||||
}
|
||||
firstSocket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
firstSocket.close(1006, "connection lost");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
@@ -1587,7 +1595,6 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
arguments: JSON.stringify({ question: "recover me" }),
|
||||
},
|
||||
});
|
||||
|
||||
expect(onToolCall).toHaveBeenCalledWith({
|
||||
itemId: "item_replayed_call",
|
||||
callId: "call_replayed",
|
||||
@@ -1669,6 +1676,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
name: "openclaw_agent_consult",
|
||||
arguments: JSON.stringify({ question: "recover output" }),
|
||||
});
|
||||
firstSocket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
await bridge.submitToolResult("call_lost_output", { text: "recovered" });
|
||||
|
||||
firstSocket.close(1006, "output acknowledgement lost");
|
||||
@@ -1702,6 +1710,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
name: "openclaw_agent_consult",
|
||||
arguments: JSON.stringify({ question: "saved output" }),
|
||||
});
|
||||
firstSocket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
await bridge.submitToolResult("call_saved_output", { text: "saved" });
|
||||
firstSocket.emitServer({
|
||||
type: "conversation.item.added",
|
||||
@@ -1766,6 +1775,7 @@ describe("buildXaiRealtimeVoiceProvider", () => {
|
||||
arguments: JSON.stringify({ question: callId }),
|
||||
});
|
||||
}
|
||||
firstSocket.emitServer({ type: "response.done", response: { status: "completed" } });
|
||||
|
||||
firstSocket.close(1006, "connection lost");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { createServer } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import type { RealtimeVoiceResponseOutcome } from "openclaw/plugin-sdk/realtime-voice";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type WebSocket from "ws";
|
||||
import { WebSocketServer } from "ws";
|
||||
@@ -7,13 +8,16 @@ import { buildXaiRealtimeVoiceProvider } from "./realtime-voice-provider.js";
|
||||
|
||||
type RealtimeOutcome = {
|
||||
errors: string[];
|
||||
outcomes: RealtimeVoiceResponseOutcome[];
|
||||
transcripts: Array<{ speaker: string; text: string; final: boolean }>;
|
||||
tools: Array<{ itemId: string; callId: string; name: string; args: unknown }>;
|
||||
};
|
||||
|
||||
type CaptureRealtimeOutcomeOptions = {
|
||||
completeQueuedResponse?: boolean;
|
||||
queuedUserMessage?: string;
|
||||
onClientEvent?: (event: Record<string, unknown>) => void;
|
||||
throwOnResponseDone?: boolean;
|
||||
};
|
||||
|
||||
async function waitForFixtureEvent(promise: Promise<void>, label: string): Promise<void> {
|
||||
@@ -35,7 +39,7 @@ async function captureRealtimeOutcome(
|
||||
options: CaptureRealtimeOutcomeOptions = {},
|
||||
): Promise<RealtimeOutcome> {
|
||||
const events = Array.isArray(eventInput) ? eventInput : [eventInput];
|
||||
const outcome: RealtimeOutcome = { errors: [], transcripts: [], tools: [] };
|
||||
const outcome: RealtimeOutcome = { errors: [], outcomes: [], transcripts: [], tools: [] };
|
||||
let markServerEventHandled: () => void = () => {};
|
||||
const serverEventHandled = new Promise<void>((resolve) => {
|
||||
markServerEventHandled = resolve;
|
||||
@@ -44,6 +48,10 @@ async function captureRealtimeOutcome(
|
||||
const responseCreatedHandled = new Promise<void>((resolve) => {
|
||||
markResponseCreatedHandled = resolve;
|
||||
});
|
||||
let markQueuedResponseCompleted: () => void = () => {};
|
||||
const queuedResponseCompleted = new Promise<void>((resolve) => {
|
||||
markQueuedResponseCompleted = resolve;
|
||||
});
|
||||
const server = createServer();
|
||||
const sockets = new Set<WebSocket>();
|
||||
let queuedTurnTriggered = false;
|
||||
@@ -58,6 +66,15 @@ async function captureRealtimeOutcome(
|
||||
>;
|
||||
options.onClientEvent?.(clientEvent);
|
||||
if (clientEvent.type === "response.create" && options.queuedUserMessage) {
|
||||
if (options.completeQueuedResponse) {
|
||||
ws.send(JSON.stringify({ type: "response.created", response: { id: "response_2" } }));
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: "response.done",
|
||||
response: { id: "response_2", status: "completed" },
|
||||
}),
|
||||
);
|
||||
}
|
||||
markServerEventHandled();
|
||||
return;
|
||||
}
|
||||
@@ -94,6 +111,15 @@ async function captureRealtimeOutcome(
|
||||
onAudio() {},
|
||||
onClearAudio() {},
|
||||
onError: (error) => outcome.errors.push(error.message),
|
||||
onResponseDone: (responseOutcome) => {
|
||||
outcome.outcomes.push(responseOutcome);
|
||||
if (responseOutcome.responseId === "response_2") {
|
||||
markQueuedResponseCompleted();
|
||||
}
|
||||
if (options.throwOnResponseDone && responseOutcome.responseId === "response_1") {
|
||||
throw new Error("consumer callback failed");
|
||||
}
|
||||
},
|
||||
onTranscript: (speaker, text, final) => outcome.transcripts.push({ speaker, text, final }),
|
||||
onToolCall: (tool) => outcome.tools.push(tool),
|
||||
onEvent: (observed) => {
|
||||
@@ -114,6 +140,9 @@ async function captureRealtimeOutcome(
|
||||
await waitForFixtureEvent(responseCreatedHandled, "response.created");
|
||||
bridge.sendUserMessage?.(options.queuedUserMessage);
|
||||
await waitForFixtureEvent(serverEventHandled, "the queued response.create");
|
||||
if (options.completeQueuedResponse) {
|
||||
await waitForFixtureEvent(queuedResponseCompleted, "the completed queued response");
|
||||
}
|
||||
} else {
|
||||
await serverEventHandled;
|
||||
}
|
||||
@@ -147,6 +176,30 @@ const expectedTool = {
|
||||
};
|
||||
|
||||
describe("xAI realtime terminal event ownership", () => {
|
||||
it("drains a queued follow-up when the terminal consumer throws", async () => {
|
||||
const outcome = await captureRealtimeOutcome(
|
||||
{
|
||||
type: "response.done",
|
||||
response: { id: "response_1", status: "failed" },
|
||||
},
|
||||
{
|
||||
completeQueuedResponse: true,
|
||||
queuedUserMessage: "Continue after the terminal callback fails.",
|
||||
throwOnResponseDone: true,
|
||||
},
|
||||
);
|
||||
|
||||
expect(outcome.errors).toEqual([]);
|
||||
expect(outcome.outcomes).toEqual([
|
||||
{
|
||||
responseId: "response_1",
|
||||
status: "failed",
|
||||
message: "xAI realtime voice response failed",
|
||||
},
|
||||
{ responseId: "response_2", status: "completed" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("flushes a queued turn after malformed terminal output over a real WebSocket", async () => {
|
||||
const clientEventTypes: string[] = [];
|
||||
|
||||
@@ -165,7 +218,12 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
},
|
||||
);
|
||||
|
||||
expect(outcome).toEqual({ errors: [], transcripts: [], tools: [] });
|
||||
expect(outcome).toEqual({
|
||||
errors: [],
|
||||
outcomes: [{ status: "completed" }],
|
||||
transcripts: [],
|
||||
tools: [],
|
||||
});
|
||||
expect(clientEventTypes).toEqual([
|
||||
"session.update",
|
||||
"conversation.item.create",
|
||||
@@ -201,7 +259,11 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
type: "response.done",
|
||||
response: { status: "failed", status_details: { error: { code: "rate_limit_exceeded" } } },
|
||||
},
|
||||
expected: { errors: ["rate_limit_exceeded"], transcripts: [], tools: [] },
|
||||
expected: {
|
||||
errors: ["xAI realtime voice response failed: rate_limit_exceeded"],
|
||||
transcripts: [],
|
||||
tools: [],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "surfaces incomplete responses with their authoritative reason",
|
||||
@@ -234,7 +296,7 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
expected: { errors: [], transcripts: [], tools: [expectedTool] },
|
||||
},
|
||||
{
|
||||
name: "retains immediate authoritative function-call argument completion",
|
||||
name: "buffers authoritative function-call arguments until response completion",
|
||||
event: {
|
||||
type: "response.function_call_arguments.done",
|
||||
item_id: completedTool.id,
|
||||
@@ -242,7 +304,7 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
name: completedTool.name,
|
||||
arguments: completedTool.arguments,
|
||||
},
|
||||
expected: { errors: [], transcripts: [], tools: [expectedTool] },
|
||||
expected: { errors: [], transcripts: [], tools: [] },
|
||||
},
|
||||
{
|
||||
name: "preserves required streamed-call timing when the response later fails",
|
||||
@@ -259,7 +321,7 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
expected: {
|
||||
errors: ["xAI realtime voice response failed"],
|
||||
transcripts: [],
|
||||
tools: [expectedTool],
|
||||
tools: [],
|
||||
},
|
||||
},
|
||||
{
|
||||
@@ -279,7 +341,7 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
expected: { errors: [], transcripts: [], tools: [expectedTool] },
|
||||
},
|
||||
{
|
||||
name: "deduplicates immediate tool delivery against terminal output",
|
||||
name: "releases finalized tool arguments only after a completed response",
|
||||
event: [
|
||||
{
|
||||
type: "response.function_call_arguments.done",
|
||||
@@ -288,7 +350,7 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
name: completedTool.name,
|
||||
arguments: completedTool.arguments,
|
||||
},
|
||||
{ type: "response.done", response: { status: "completed", output: [completedTool] } },
|
||||
{ type: "response.done", response: { status: "completed" } },
|
||||
],
|
||||
expected: { errors: [], transcripts: [], tools: [expectedTool] },
|
||||
},
|
||||
@@ -312,7 +374,54 @@ describe("xAI realtime terminal event ownership", () => {
|
||||
},
|
||||
expected: { errors: [], transcripts: [], tools: [] },
|
||||
},
|
||||
{
|
||||
name: "fails closed when response status is missing",
|
||||
event: { type: "response.done", response: {} },
|
||||
expected: {
|
||||
errors: ["xAI realtime voice response failed: missing terminal status"],
|
||||
transcripts: [],
|
||||
tools: [],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "fails closed when response status is invalid",
|
||||
event: { type: "response.done", response: { status: "in_progress" } },
|
||||
expected: {
|
||||
errors: ["xAI realtime voice response failed: invalid status in_progress"],
|
||||
transcripts: [],
|
||||
tools: [],
|
||||
},
|
||||
},
|
||||
])("$name", async ({ event, expected }) => {
|
||||
expect(await captureRealtimeOutcome(event)).toEqual(expected);
|
||||
const actual = await captureRealtimeOutcome(event);
|
||||
expect(actual.errors).toEqual([]);
|
||||
expect(actual.transcripts).toEqual(expected.transcripts);
|
||||
expect(actual.tools).toEqual(expected.tools);
|
||||
const events = Array.isArray(event) ? event : [event];
|
||||
const responseDone = events.findLast((candidate) => candidate.type === "response.done") as
|
||||
| { response?: { status?: string } }
|
||||
| undefined;
|
||||
if (!responseDone) {
|
||||
expect(actual.outcomes).toEqual([]);
|
||||
return;
|
||||
}
|
||||
expect(actual.outcomes).toHaveLength(1);
|
||||
const rawStatus = responseDone.response?.status;
|
||||
if (
|
||||
rawStatus !== "completed" &&
|
||||
rawStatus !== "cancelled" &&
|
||||
rawStatus !== "failed" &&
|
||||
rawStatus !== "incomplete"
|
||||
) {
|
||||
expect(actual.outcomes[0]).toMatchObject({
|
||||
status: "failed",
|
||||
reason: "invalid_response_status",
|
||||
});
|
||||
} else {
|
||||
expect(actual.outcomes[0]?.status).toBe(rawStatus);
|
||||
}
|
||||
if (expected.errors[0]) {
|
||||
expect(actual.outcomes[0]).toMatchObject({ message: expected.errors[0] });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -313,8 +313,10 @@ export const talkClientHandlers: GatewayRequestHandlers = {
|
||||
const gatewayControlOwner = wantsGatewayControl
|
||||
? createTalkClientGatewayControlOwner({
|
||||
voiceSessionId: activeVoiceSessionId!,
|
||||
providerId: resolution.provider.id,
|
||||
sessionKey,
|
||||
connId: ownerConnId!,
|
||||
context,
|
||||
runAgentConsult: consultRunner.runArgs,
|
||||
appendTranscript: ({ entryId, role, text }) =>
|
||||
appendClientVoiceTranscript({
|
||||
@@ -339,7 +341,6 @@ export const talkClientHandlers: GatewayRequestHandlers = {
|
||||
config: runtimeConfig,
|
||||
});
|
||||
},
|
||||
warn: (message) => context.logGateway.warn(message),
|
||||
})
|
||||
: undefined;
|
||||
const browserSessionRequest: InternalRealtimeVoiceBrowserSessionCreateRequest = {
|
||||
|
||||
@@ -14,7 +14,78 @@ function deferred<T>() {
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
function controlContext(
|
||||
warn = vi.fn(),
|
||||
onTalkEvent?: (event: { type: string; payload: unknown }) => void,
|
||||
) {
|
||||
return {
|
||||
logGateway: { warn },
|
||||
broadcastToConnIds: vi.fn((_name: string, payload: { talkEvent?: unknown }) => {
|
||||
if (payload.talkEvent) {
|
||||
onTalkEvent?.(payload.talkEvent as { type: string; payload: unknown });
|
||||
}
|
||||
}),
|
||||
} as never;
|
||||
}
|
||||
|
||||
describe("Talk client Gateway control owner", () => {
|
||||
it.each(["failed", "incomplete"] as const)(
|
||||
"keeps Gateway-controlled browser Talk reusable after a %s response",
|
||||
async (status) => {
|
||||
const warn = vi.fn();
|
||||
const closeProvider = vi.fn(async () => undefined);
|
||||
const closeLogicalSession = vi.fn(async () => undefined);
|
||||
const talkEvents: Array<{ type: string; payload: unknown }> = [];
|
||||
const owner = createTalkClientGatewayControlOwner({
|
||||
voiceSessionId: `voice-${status}`,
|
||||
providerId: "openai",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-gateway",
|
||||
context: controlContext(warn, (event) => talkEvents.push(event)),
|
||||
runAgentConsult: vi.fn(async () => ({ text: "done" })),
|
||||
appendTranscript: vi.fn(async () => undefined),
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession,
|
||||
});
|
||||
owner.activate(closeProvider);
|
||||
owner.control.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.created",
|
||||
responseId: "response-1",
|
||||
});
|
||||
const firstOutcome = {
|
||||
status,
|
||||
responseId: "response-1",
|
||||
message: `provider ${status}`,
|
||||
} as const;
|
||||
owner.control.onResponseDone?.(firstOutcome);
|
||||
owner.control.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.done",
|
||||
responseId: "response-1",
|
||||
});
|
||||
owner.control.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.created",
|
||||
responseId: "response-2",
|
||||
});
|
||||
owner.control.onResponseDone?.({ status: "completed", responseId: "response-2" });
|
||||
owner.control.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.done",
|
||||
responseId: "response-2",
|
||||
});
|
||||
|
||||
expect(talkEvents.filter((event) => event.type === "session.error")).toHaveLength(1);
|
||||
expect(talkEvents.filter((event) => event.type === "turn.ended")).toHaveLength(2);
|
||||
expect(warn).toHaveBeenCalledWith(`talk Gateway control provider ${status}`);
|
||||
expect(closeProvider).not.toHaveBeenCalled();
|
||||
expect(closeLogicalSession).not.toHaveBeenCalled();
|
||||
|
||||
await owner.close();
|
||||
},
|
||||
);
|
||||
|
||||
it("persists sideband transcripts, completes consults, and closes idempotently", async () => {
|
||||
const consultResult = deferred<{ text: string }>();
|
||||
const runAgentConsult = vi.fn(async () => await consultResult.promise);
|
||||
@@ -37,11 +108,11 @@ describe("Talk client Gateway control owner", () => {
|
||||
voiceSessionId: "voice-gateway",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-gateway",
|
||||
context: controlContext(),
|
||||
runAgentConsult,
|
||||
appendTranscript,
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession,
|
||||
warn: vi.fn(),
|
||||
});
|
||||
owner.control.bindBridge(bridge);
|
||||
owner.activate(closeProvider);
|
||||
@@ -95,11 +166,11 @@ describe("Talk client Gateway control owner", () => {
|
||||
voiceSessionId: "voice-control",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-control",
|
||||
context: controlContext(),
|
||||
runAgentConsult,
|
||||
appendTranscript: vi.fn(async () => undefined),
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession: vi.fn(async () => undefined),
|
||||
warn: vi.fn(),
|
||||
});
|
||||
owner.control.bindBridge(bridge);
|
||||
owner.activate(vi.fn(async () => undefined));
|
||||
@@ -168,12 +239,12 @@ describe("Talk client Gateway control owner", () => {
|
||||
voiceSessionId: "voice-spoken-control",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-spoken-control",
|
||||
context: controlContext(),
|
||||
runAgentConsult,
|
||||
controlAgentRun,
|
||||
appendTranscript: vi.fn(async () => undefined),
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession: vi.fn(async () => undefined),
|
||||
warn: vi.fn(),
|
||||
});
|
||||
owner.control.bindBridge(bridge);
|
||||
owner.activate(vi.fn(async () => undefined));
|
||||
@@ -216,11 +287,11 @@ describe("Talk client Gateway control owner", () => {
|
||||
voiceSessionId: "voice-disconnect",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-disconnect",
|
||||
context: controlContext(),
|
||||
runAgentConsult: vi.fn(async () => ({ text: "done" })),
|
||||
appendTranscript: vi.fn(async () => undefined),
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession,
|
||||
warn: vi.fn(),
|
||||
});
|
||||
owner.activate(closeProvider);
|
||||
|
||||
@@ -236,11 +307,11 @@ describe("Talk client Gateway control owner", () => {
|
||||
voiceSessionId: "voice-close-error",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-close-error",
|
||||
context: controlContext(),
|
||||
runAgentConsult: vi.fn(async () => ({ text: "done" })),
|
||||
appendTranscript: vi.fn(async () => undefined),
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession,
|
||||
warn: vi.fn(),
|
||||
});
|
||||
owner.activate(vi.fn(() => Promise.reject(new Error("provider close failed"))));
|
||||
|
||||
@@ -265,11 +336,11 @@ describe("Talk client Gateway control owner", () => {
|
||||
voiceSessionId: "voice-replacement",
|
||||
sessionKey: "agent:main:main",
|
||||
connId: "conn-replacement",
|
||||
context: controlContext(),
|
||||
runAgentConsult,
|
||||
appendTranscript,
|
||||
flushTranscript: vi.fn(async () => undefined),
|
||||
closeLogicalSession,
|
||||
warn: vi.fn(),
|
||||
};
|
||||
const firstBridge = {
|
||||
connect: vi.fn(async () => undefined),
|
||||
|
||||
@@ -26,6 +26,11 @@ import type {
|
||||
RealtimeVoiceGatewayControl,
|
||||
RealtimeVoiceToolCallEvent,
|
||||
} from "../talk/provider-types.js";
|
||||
import {
|
||||
createRealtimeVoiceSessionHarness,
|
||||
handleRealtimeVoiceHarnessBridgeEvent,
|
||||
} from "../talk/realtime-session-harness.js";
|
||||
import type { TalkEvent } from "../talk/talk-events.js";
|
||||
import { registerChatAbortController } from "./chat-abort.js";
|
||||
import type { GatewayRequestContext } from "./server-methods/shared-types.js";
|
||||
import { formatError } from "./server-utils.js";
|
||||
@@ -291,8 +296,10 @@ export function createTalkClientAgentConsultRunner(params: {
|
||||
|
||||
export function createTalkClientGatewayControlOwner(params: {
|
||||
voiceSessionId: string;
|
||||
providerId?: string;
|
||||
sessionKey: string;
|
||||
connId: string;
|
||||
context: Pick<GatewayRequestContext, "broadcastToConnIds" | "logGateway">;
|
||||
runAgentConsult: (args: unknown, signal: AbortSignal) => Promise<{ text: string }>;
|
||||
appendTranscript: (entry: {
|
||||
entryId: string;
|
||||
@@ -306,7 +313,6 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
text: string;
|
||||
mode?: unknown;
|
||||
}) => Promise<RealtimeVoiceAgentControlResult>;
|
||||
warn: (message: string) => void;
|
||||
}): GatewayControlOwner {
|
||||
let bridge: RealtimeVoiceBridge | undefined;
|
||||
let closeProvider: (() => Promise<void>) | undefined;
|
||||
@@ -316,6 +322,33 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
const entryPrefix = `gateway-${randomUUID()}`;
|
||||
const consultQueue = createRealtimeControlQueue();
|
||||
const consultControllers = new Map<string, AbortController>();
|
||||
const warn = (message: string) => params.context.logGateway.warn(message);
|
||||
const talkPayload = () => ({ voiceSessionId: params.voiceSessionId });
|
||||
const harness = createRealtimeVoiceSessionHarness({
|
||||
talk: {
|
||||
sessionId: params.voiceSessionId,
|
||||
mode: "realtime",
|
||||
transport: "webrtc",
|
||||
brain: "agent-consult",
|
||||
provider: params.providerId,
|
||||
},
|
||||
talkPayloads: {
|
||||
turnStarted: talkPayload,
|
||||
turnEnded: (reason) => ({ ...talkPayload(), reason }),
|
||||
inputAudioDelta: (audio) => ({ ...talkPayload(), byteLength: audio.byteLength }),
|
||||
outputAudioStarted: talkPayload,
|
||||
outputAudioDelta: (audio) => ({ ...talkPayload(), byteLength: audio.byteLength }),
|
||||
outputAudioDone: (reason) => ({ ...talkPayload(), reason }),
|
||||
},
|
||||
onTalkEvent: (talkEvent: TalkEvent) =>
|
||||
params.context.broadcastToConnIds(
|
||||
"talk.event",
|
||||
{ voiceSessionId: params.voiceSessionId, talkEvent },
|
||||
new Set([params.connId]),
|
||||
{ dropIfSlow: talkEvent.final !== true },
|
||||
),
|
||||
captureBridgeEvents: false,
|
||||
});
|
||||
|
||||
const submit = async (callId: string, result: unknown): Promise<void> => {
|
||||
if (!bridge) {
|
||||
@@ -370,7 +403,7 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
hasActiveRun: () => consultControllers.size > 0,
|
||||
execute: applyControl,
|
||||
speak: (message) => bridge?.sendUserMessage?.(message),
|
||||
warn: params.warn,
|
||||
warn,
|
||||
});
|
||||
|
||||
const handleToolCall = (event: RealtimeVoiceToolCallEvent): void => {
|
||||
@@ -387,7 +420,7 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
return;
|
||||
}
|
||||
void admission.completion.catch((error: unknown) => {
|
||||
params.warn(`talk Gateway control consult failed: ${formatError(error)}`);
|
||||
warn(`talk Gateway control consult failed: ${formatError(error)}`);
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -405,18 +438,35 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
void submit(event.callId, {
|
||||
error: `Unsupported realtime Talk tool: ${event.name}`,
|
||||
}).catch((error: unknown) => {
|
||||
params.warn(`talk Gateway control rejection failed: ${formatError(error)}`);
|
||||
warn(`talk Gateway control rejection failed: ${formatError(error)}`);
|
||||
});
|
||||
};
|
||||
|
||||
const handleTranscript = (role: "user" | "assistant", text: string, final: boolean): void => {
|
||||
if (closed || !final || !text.trim()) {
|
||||
if (closed || !text.trim()) {
|
||||
return;
|
||||
}
|
||||
const turnId = harness.ensureTurn();
|
||||
harness.emit({
|
||||
type:
|
||||
role === "assistant"
|
||||
? final
|
||||
? "output.text.done"
|
||||
: "output.text.delta"
|
||||
: final
|
||||
? "transcript.done"
|
||||
: "transcript.delta",
|
||||
turnId,
|
||||
payload: role === "assistant" ? { text } : { role, text },
|
||||
final,
|
||||
});
|
||||
if (!final) {
|
||||
return;
|
||||
}
|
||||
transcriptSequence += 1;
|
||||
const entryId = `${entryPrefix}-${transcriptSequence}`;
|
||||
void params.appendTranscript({ entryId, role, text }).catch((error: unknown) => {
|
||||
params.warn(`talk Gateway control transcript failed: ${formatError(error)}`);
|
||||
warn(`talk Gateway control transcript failed: ${formatError(error)}`);
|
||||
});
|
||||
if (role === "user") {
|
||||
runControl.handleSpoken(text, params.flushTranscript());
|
||||
@@ -430,12 +480,46 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
bindBridge: (nextBridge) => {
|
||||
bridge = nextBridge;
|
||||
},
|
||||
onEvent: (event) => {
|
||||
const legacyOutcome = handleRealtimeVoiceHarnessBridgeEvent(harness, event);
|
||||
if (
|
||||
legacyOutcome &&
|
||||
(legacyOutcome.status === "failed" || legacyOutcome.status === "incomplete")
|
||||
) {
|
||||
warn(`talk Gateway control ${legacyOutcome.message}`);
|
||||
}
|
||||
if (
|
||||
event.direction === "server" &&
|
||||
(event.type === "conversation.output_audio.delta" ||
|
||||
event.type === "response.audio.delta" ||
|
||||
event.type === "response.output_audio.delta")
|
||||
) {
|
||||
const turnId = harness.ensureTurn();
|
||||
harness.talk.startOutputAudio({ turnId, payload: talkPayload() });
|
||||
}
|
||||
},
|
||||
onTranscript: handleTranscript,
|
||||
onToolCall: handleToolCall,
|
||||
onError: (error) => params.warn(`talk Gateway control provider error: ${error.message}`),
|
||||
onResponseDone: (outcome) => {
|
||||
const terminal = harness.finishResponse(outcome);
|
||||
if (terminal.ok && (outcome.status === "failed" || outcome.status === "incomplete")) {
|
||||
warn(`talk Gateway control ${outcome.message}`);
|
||||
}
|
||||
},
|
||||
onReady: () => harness.emit({ type: "session.ready", payload: talkPayload() }),
|
||||
onError: (error) => {
|
||||
warn(`talk Gateway control provider error: ${error.message}`);
|
||||
harness.emit({
|
||||
type: "session.error",
|
||||
payload: { ...talkPayload(), message: error.message },
|
||||
final: true,
|
||||
});
|
||||
},
|
||||
onClose: () => {
|
||||
harness.emit({ type: "session.closed", payload: talkPayload(), final: true });
|
||||
harness.close();
|
||||
void owner.close({ skipProvider: true }).catch((error: unknown) => {
|
||||
params.warn(`talk Gateway control close failed: ${formatError(error)}`);
|
||||
warn(`talk Gateway control close failed: ${formatError(error)}`);
|
||||
});
|
||||
},
|
||||
},
|
||||
@@ -447,14 +531,14 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
void previous
|
||||
.close({ preserveLogicalSession: true, preserveRuns: true })
|
||||
.catch((error: unknown) => {
|
||||
params.warn(`talk replaced Gateway transport close failed: ${formatError(error)}`);
|
||||
warn(`talk replaced Gateway transport close failed: ${formatError(error)}`);
|
||||
});
|
||||
}
|
||||
registerTalkConnectionCleanup(params.connId, "browser-control", () => {
|
||||
for (const current of owners.values()) {
|
||||
if (current.connId === params.connId) {
|
||||
void current.close().catch((error: unknown) => {
|
||||
params.warn(`talk disconnected Gateway control close failed: ${formatError(error)}`);
|
||||
warn(`talk disconnected Gateway control close failed: ${formatError(error)}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -468,6 +552,7 @@ export function createTalkClientGatewayControlOwner(params: {
|
||||
// can re-enter close without starting a second cleanup.
|
||||
closing = Promise.resolve().then(async () => {
|
||||
closed = true;
|
||||
harness.close();
|
||||
if (owners.get(params.voiceSessionId) === owner) {
|
||||
owners.delete(params.voiceSessionId);
|
||||
}
|
||||
|
||||
@@ -315,27 +315,42 @@ export function createTalkRealtimeRelaySession(
|
||||
) {
|
||||
currentOutputItemId = event.itemId ?? currentOutputItemId;
|
||||
currentOutputResponseId = event.responseId ?? currentOutputResponseId;
|
||||
}
|
||||
},
|
||||
onResponseDone: (outcome) => {
|
||||
const relay = getActiveRelay();
|
||||
if (!relay) {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
event.type === "response.audio.done" ||
|
||||
event.type === "response.output_audio.done" ||
|
||||
event.type === "conversation.output_audio.done" ||
|
||||
event.type === "response.done" ||
|
||||
event.type === "response.cancelled"
|
||||
) {
|
||||
emit({
|
||||
relaySessionId,
|
||||
type: "audioDone",
|
||||
...((event.itemId ?? currentOutputItemId)
|
||||
? { itemId: event.itemId ?? currentOutputItemId }
|
||||
: {}),
|
||||
...((event.responseId ?? currentOutputResponseId)
|
||||
? { responseId: event.responseId ?? currentOutputResponseId }
|
||||
: {}),
|
||||
const terminalTalkEvent = harness.talk.recentEvents.at(-1);
|
||||
broadcastToOwner(params.context, params.connId, {
|
||||
relaySessionId,
|
||||
type: "audioDone",
|
||||
...(currentOutputItemId ? { itemId: currentOutputItemId } : {}),
|
||||
...((outcome.responseId ?? currentOutputResponseId)
|
||||
? { responseId: outcome.responseId ?? currentOutputResponseId }
|
||||
: {}),
|
||||
...(terminalTalkEvent &&
|
||||
(terminalTalkEvent.type === "turn.ended" || terminalTalkEvent.type === "turn.cancelled")
|
||||
? { talkEvent: terminalTalkEvent }
|
||||
: {}),
|
||||
});
|
||||
currentOutputItemId = undefined;
|
||||
currentOutputResponseId = undefined;
|
||||
if (outcome.status === "failed" || outcome.status === "incomplete") {
|
||||
const issue = realtimeRelayIssue({
|
||||
message: outcome.message,
|
||||
provider: params.provider.id,
|
||||
model: params.model,
|
||||
phase: "response",
|
||||
});
|
||||
const errorTalkEvent = harness.talk.recentEvents.findLast(
|
||||
(event) => event.type === "session.error" && event.payload === outcome,
|
||||
);
|
||||
broadcastToOwner(params.context, params.connId, {
|
||||
...relayIssuePayload(relaySessionId, issue),
|
||||
...(errorTalkEvent ? { talkEvent: errorTalkEvent } : {}),
|
||||
});
|
||||
currentOutputItemId = undefined;
|
||||
currentOutputResponseId = undefined;
|
||||
}
|
||||
},
|
||||
onTranscript: (role, text, final) => {
|
||||
|
||||
@@ -78,6 +78,113 @@ function stopTalkRealtimeRelaySession(
|
||||
}
|
||||
|
||||
describe("talk realtime gateway relay", () => {
|
||||
it.each([
|
||||
[
|
||||
{ status: "failed" as const, responseId: "response-1", message: "provider failed" },
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{
|
||||
status: "incomplete" as const,
|
||||
responseId: "response-1",
|
||||
reason: "max_output_tokens",
|
||||
message: "provider response incomplete",
|
||||
},
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{ status: "cancelled" as const, responseId: "response-1", reason: "client_cancelled" },
|
||||
"turn.cancelled",
|
||||
],
|
||||
])("keeps a relay reusable after each terminal response", async (outcome, terminalType) => {
|
||||
let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined;
|
||||
const close = vi.fn();
|
||||
const provider: RealtimeVoiceProviderPlugin = {
|
||||
id: "relay-test",
|
||||
label: "Relay Test",
|
||||
isConfigured: () => true,
|
||||
createBridge: (request) => {
|
||||
bridgeRequest = request;
|
||||
return makeRelayTransport({ close });
|
||||
},
|
||||
};
|
||||
const events: Array<{ payload: unknown }> = [];
|
||||
const context = {
|
||||
broadcastToConnIds: (_event: string, payload: unknown) => events.push({ payload }),
|
||||
} as never;
|
||||
const session = createTalkRealtimeRelaySession({
|
||||
context,
|
||||
connId: "conn-1",
|
||||
provider,
|
||||
providerConfig: {},
|
||||
instructions: "be brief",
|
||||
tools: [],
|
||||
});
|
||||
await Promise.resolve();
|
||||
if (!bridgeRequest) {
|
||||
throw new Error("expected realtime bridge request");
|
||||
}
|
||||
|
||||
sendTalkRealtimeRelayAudio({
|
||||
relaySessionId: session.relaySessionId,
|
||||
connId: "conn-1",
|
||||
audioBase64: Buffer.from("first").toString("base64"),
|
||||
timestamp: 1,
|
||||
});
|
||||
bridgeRequest.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.created",
|
||||
responseId: outcome.responseId,
|
||||
});
|
||||
bridgeRequest.onResponseDone?.(outcome);
|
||||
bridgeRequest.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: outcome.responseId,
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
const firstPayloads = events.map(({ payload }) => payload as Record<string, unknown>);
|
||||
const firstTalkEvents = firstPayloads
|
||||
.map((payload) => payload.talkEvent)
|
||||
.filter((event): event is Record<string, unknown> => Boolean(event));
|
||||
expect(firstTalkEvents.filter((event) => event.type === terminalType)).toHaveLength(1);
|
||||
expect(firstPayloads.filter((payload) => payload.type === "error")).toHaveLength(
|
||||
outcome.status === "cancelled" ? 0 : 1,
|
||||
);
|
||||
expect(firstPayloads.filter((payload) => payload.type === "audioDone")).toHaveLength(1);
|
||||
expect(relaySessions.has(session.relaySessionId)).toBe(true);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
|
||||
sendTalkRealtimeRelayAudio({
|
||||
relaySessionId: session.relaySessionId,
|
||||
connId: "conn-1",
|
||||
audioBase64: Buffer.from("later").toString("base64"),
|
||||
timestamp: 2,
|
||||
});
|
||||
bridgeRequest.onEvent?.({
|
||||
direction: "server",
|
||||
type: "response.created",
|
||||
responseId: "response-2",
|
||||
});
|
||||
bridgeRequest.onResponseDone?.({ status: "completed", responseId: "response-2" });
|
||||
bridgeRequest.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: "response-2",
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
expect(
|
||||
events.filter(
|
||||
({ payload }) =>
|
||||
typeof payload === "object" &&
|
||||
payload !== null &&
|
||||
(payload as Record<string, unknown>).type === "audioDone",
|
||||
),
|
||||
).toHaveLength(2);
|
||||
expect(relaySessions.has(session.relaySessionId)).toBe(true);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
for (const [relaySessionId, connId] of activeRelaySessions) {
|
||||
try {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { normalizeOptionalString as readLogString } from "@openclaw/normalization-core/string-coerce";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import type { RuntimeLogger } from "../plugins/runtime/types.js";
|
||||
import type {
|
||||
RealtimeTranscriptionProviderPlugin,
|
||||
RealtimeVoiceProviderPlugin,
|
||||
@@ -10,9 +11,33 @@ import {
|
||||
} from "../realtime-transcription/provider-registry.js";
|
||||
import type { RealtimeTranscriptionProviderConfig } from "../realtime-transcription/provider-types.js";
|
||||
import { resolveConfiguredRealtimeVoiceProvider } from "../talk/provider-resolver.js";
|
||||
import type { RealtimeVoiceProviderConfig } from "../talk/provider-types.js";
|
||||
import type {
|
||||
RealtimeVoiceBridgeEvent,
|
||||
RealtimeVoiceProviderConfig,
|
||||
RealtimeVoiceResponseOutcome,
|
||||
} from "../talk/provider-types.js";
|
||||
import type { RealtimeVoiceSessionHarness } from "../talk/realtime-session-harness.js";
|
||||
import { truncateUtf16Safe } from "../utils.js";
|
||||
import type { MeetingRealtimeAudioFormat } from "./realtime-audio-format.js";
|
||||
import type { createMeetingRealtimeOutputOwner } from "./realtime-output-owner.js";
|
||||
|
||||
const MEETING_REALTIME_CANCELLATION_RACE_DETAIL = "Cancellation failed: no active response found";
|
||||
|
||||
type MeetingRealtimeLifecycleHandlersParams = {
|
||||
clearOutputPlayback: () => void;
|
||||
getContinuityResetActive: () => boolean;
|
||||
harness: RealtimeVoiceSessionHarness;
|
||||
invalidateOutputPlayback: () => void;
|
||||
logScope: string;
|
||||
logger: RuntimeLogger;
|
||||
outputOwner: ReturnType<typeof createMeetingRealtimeOutputOwner>;
|
||||
outputTalkPayload: { bridgeId: string } | { meetingSessionId: string };
|
||||
realtimeLogScope: string;
|
||||
resetToolContinuity: (reason: string) => void;
|
||||
setContinuityResetActive: (active: boolean) => void;
|
||||
setOutputGenerationActive: (active: boolean) => void;
|
||||
setRealtimeReady: (ready: boolean) => void;
|
||||
};
|
||||
|
||||
type MeetingRealtimeProviderSelectionConfig = {
|
||||
realtime: {
|
||||
@@ -200,3 +225,90 @@ export function normalizeMeetingTtsPromptText(text: string | undefined): string
|
||||
}
|
||||
return trimmed;
|
||||
}
|
||||
|
||||
export function createMeetingRealtimeLifecycleHandlers(
|
||||
params: MeetingRealtimeLifecycleHandlersParams,
|
||||
) {
|
||||
const onEvent = (event: RealtimeVoiceBridgeEvent) => {
|
||||
if (event.direction === "server" && event.type === "session.created") {
|
||||
params.setContinuityResetActive(false);
|
||||
}
|
||||
if (event.direction === "client" && event.type === "session.continuity.reset") {
|
||||
if (params.getContinuityResetActive()) {
|
||||
return;
|
||||
}
|
||||
params.setContinuityResetActive(true);
|
||||
params.setRealtimeReady(false);
|
||||
params.outputOwner.reset();
|
||||
params.setOutputGenerationActive(false);
|
||||
params.resetToolContinuity(event.type);
|
||||
const turnId = params.harness.talk.activeTurnId;
|
||||
params.invalidateOutputPlayback();
|
||||
params.harness.flushOutput(params.clearOutputPlayback);
|
||||
params.harness.finishOutputAudio(event.type);
|
||||
if (turnId) {
|
||||
params.harness.talk.cancelTurn({
|
||||
turnId,
|
||||
payload: { ...params.outputTalkPayload, reason: event.type },
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
params.outputOwner.noteEvent(event);
|
||||
if (event.type === "input_audio_buffer.speech_started") {
|
||||
params.harness.ensureTurn();
|
||||
} else if (event.type === "input_audio_buffer.speech_stopped") {
|
||||
const turnId = params.harness.talk.activeTurnId;
|
||||
if (!turnId) {
|
||||
return;
|
||||
}
|
||||
params.harness.emit({
|
||||
type: "input.audio.committed",
|
||||
turnId,
|
||||
payload: { ...params.outputTalkPayload, source: event.type },
|
||||
final: true,
|
||||
});
|
||||
} else if (
|
||||
event.type === "error" &&
|
||||
event.detail === MEETING_REALTIME_CANCELLATION_RACE_DETAIL
|
||||
) {
|
||||
if (params.outputOwner.clearBlocked()) {
|
||||
params.setOutputGenerationActive(false);
|
||||
params.harness.finishOutputAudio(event.type);
|
||||
}
|
||||
} else if (event.type === "error") {
|
||||
params.harness.emit({
|
||||
type: "session.error",
|
||||
payload: { message: event.detail ?? "Realtime provider error" },
|
||||
final: true,
|
||||
});
|
||||
}
|
||||
if (
|
||||
event.type === "error" ||
|
||||
event.type === "response.done" ||
|
||||
event.type === "input_audio_buffer.speech_started" ||
|
||||
event.type === "input_audio_buffer.speech_stopped" ||
|
||||
event.type === "conversation.item.input_audio_transcription.completed" ||
|
||||
event.type === "conversation.item.input_audio_transcription.failed"
|
||||
) {
|
||||
const detail = event.detail ? ` ${event.detail}` : "";
|
||||
params.logger.info(
|
||||
`${params.logScope} ${params.realtimeLogScope} ${event.direction}:${event.type}${detail}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
const onResponseDone = (outcome: RealtimeVoiceResponseOutcome) => {
|
||||
if (!params.outputOwner.terminal(outcome.responseId)) {
|
||||
return;
|
||||
}
|
||||
params.setOutputGenerationActive(false);
|
||||
if (outcome.status === "failed" || outcome.status === "incomplete") {
|
||||
params.logger.warn(
|
||||
`${params.logScope} ${params.realtimeLogScope} response ${outcome.status}: ${outcome.message}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
return { onEvent, onResponseDone };
|
||||
}
|
||||
|
||||
@@ -134,6 +134,72 @@ async function createEngineFixture(options?: {
|
||||
}
|
||||
|
||||
describe("meeting realtime engine output ownership", () => {
|
||||
it.each([
|
||||
[{ status: "completed" as const, responseId: "response-1" }, "turn.ended"],
|
||||
[
|
||||
{ status: "failed" as const, responseId: "response-1", message: "provider failed" },
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{
|
||||
status: "incomplete" as const,
|
||||
responseId: "response-1",
|
||||
reason: "max_output_tokens",
|
||||
message: "provider response incomplete",
|
||||
},
|
||||
"turn.ended",
|
||||
],
|
||||
[
|
||||
{ status: "cancelled" as const, responseId: "response-1", reason: "client_cancelled" },
|
||||
"turn.cancelled",
|
||||
],
|
||||
])("finishes each response once and accepts a later response", async (outcome, terminalType) => {
|
||||
const fixture = await createEngineFixture();
|
||||
try {
|
||||
fixture.callbacks.onTranscript?.("user", "first turn", true);
|
||||
fixture.announceOutputResponse("response-1");
|
||||
fixture.sendOutputAudio(Buffer.from([1]), "response-1");
|
||||
await vi.waitFor(() => expect(fixture.writeOutput).toHaveBeenCalledTimes(1));
|
||||
fixture.callbacks.onResponseDone?.(outcome);
|
||||
fixture.callbacks.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: outcome.responseId,
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
const firstEvents = fixture.handle.getHealth().recentTalkEvents;
|
||||
expect(firstEvents.filter((event) => event.type === terminalType)).toHaveLength(1);
|
||||
expect(firstEvents.filter((event) => event.type === "output.audio.done")).toHaveLength(1);
|
||||
expect(firstEvents.filter((event) => event.type === "session.error")).toHaveLength(
|
||||
outcome.status === "failed" || outcome.status === "incomplete" ? 1 : 0,
|
||||
);
|
||||
expect(fixture.handle.getHealth().bridgeClosed).toBe(false);
|
||||
|
||||
fixture.releaseWrite(0);
|
||||
fixture.callbacks.onTranscript?.("user", "later turn", true);
|
||||
fixture.announceOutputResponse("response-2");
|
||||
fixture.sendOutputAudio(Buffer.from([2]), "response-2");
|
||||
await vi.waitFor(() => expect(fixture.writeOutput).toHaveBeenCalledTimes(2));
|
||||
fixture.callbacks.onResponseDone?.({ status: "completed", responseId: "response-2" });
|
||||
fixture.callbacks.onEvent?.({
|
||||
direction: "server",
|
||||
responseId: "response-2",
|
||||
type: "response.done",
|
||||
});
|
||||
|
||||
const finalEvents = fixture.handle.getHealth().recentTalkEvents;
|
||||
expect(
|
||||
finalEvents.filter(
|
||||
(event) => event.type === "turn.ended" || event.type === "turn.cancelled",
|
||||
),
|
||||
).toHaveLength(2);
|
||||
expect(finalEvents.filter((event) => event.type === "output.audio.done")).toHaveLength(2);
|
||||
fixture.releaseWrite(1);
|
||||
} finally {
|
||||
await fixture.handle.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("rearms continuity reset when the provider creates a fresh session before ready", async () => {
|
||||
const fixture = await createEngineFixture();
|
||||
try {
|
||||
|
||||
@@ -20,6 +20,7 @@ import type {
|
||||
} from "./realtime-audio-transport.js";
|
||||
import {
|
||||
buildMeetingSpeakExactUserMessage,
|
||||
createMeetingRealtimeLifecycleHandlers,
|
||||
formatMeetingTranscriptSummaryLog,
|
||||
formatMeetingRealtimeVoiceModelLog,
|
||||
meetingOutputBytesPerMs,
|
||||
@@ -97,7 +98,6 @@ export const MEETING_TRANSCRIPT_ECHO_LOOKBACK_MS = 45_000;
|
||||
const MEETING_REALTIME_OUTPUT_MAX_PENDING_MS = 2_000;
|
||||
const MEETING_REALTIME_OUTPUT_MAX_WRITE_MS = 500;
|
||||
const MEETING_REALTIME_OUTPUT_MAX_PENDING_FRAMES = 256;
|
||||
const MEETING_REALTIME_CANCELLATION_RACE_DETAIL = "Cancellation failed: no active response found";
|
||||
export async function startMeetingRealtimeEngine(params: {
|
||||
config: MeetingRealtimeEngineConfig;
|
||||
fullConfig: OpenClawConfig;
|
||||
@@ -466,6 +466,21 @@ export async function startMeetingRealtimeEngine(params: {
|
||||
`${params.platform.displayName} audio transport failed before realtime provider setup`,
|
||||
);
|
||||
}
|
||||
const lifecycleHandlers = createMeetingRealtimeLifecycleHandlers({
|
||||
clearOutputPlayback,
|
||||
getContinuityResetActive: () => continuityResetActive,
|
||||
harness,
|
||||
invalidateOutputPlayback,
|
||||
logger: params.logger,
|
||||
logScope: params.platform.logScope,
|
||||
outputOwner,
|
||||
outputTalkPayload,
|
||||
realtimeLogScope,
|
||||
resetToolContinuity: (reason) => toolContinuity.reset(reason),
|
||||
setContinuityResetActive: (active) => (continuityResetActive = active),
|
||||
setOutputGenerationActive: (active) => (outputGenerationActive = active),
|
||||
setRealtimeReady: (ready) => (realtimeReady = ready),
|
||||
});
|
||||
try {
|
||||
bridge = harness.createBridge({
|
||||
provider: resolved.provider,
|
||||
@@ -550,82 +565,8 @@ export async function startMeetingRealtimeEngine(params: {
|
||||
}
|
||||
}
|
||||
},
|
||||
onEvent: (event) => {
|
||||
if (event.direction === "server" && event.type === "session.created") {
|
||||
continuityResetActive = false;
|
||||
}
|
||||
if (event.direction === "client" && event.type === "session.continuity.reset") {
|
||||
if (continuityResetActive) {
|
||||
return;
|
||||
}
|
||||
continuityResetActive = true;
|
||||
realtimeReady = false;
|
||||
outputOwner.reset();
|
||||
outputGenerationActive = false;
|
||||
toolContinuity.reset(event.type);
|
||||
const turnId = harness.talk.activeTurnId;
|
||||
invalidateOutputPlayback();
|
||||
harness.flushOutput(clearOutputPlayback);
|
||||
harness.finishOutputAudio(event.type);
|
||||
if (turnId) {
|
||||
harness.talk.cancelTurn({
|
||||
turnId,
|
||||
payload: { ...outputTalkPayload, reason: event.type },
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
outputOwner.noteEvent(event);
|
||||
if (event.type === "input_audio_buffer.speech_started") {
|
||||
harness.ensureTurn();
|
||||
} else if (event.type === "input_audio_buffer.speech_stopped") {
|
||||
const turnId = harness.talk.activeTurnId;
|
||||
if (!turnId) {
|
||||
return;
|
||||
}
|
||||
harness.emit({
|
||||
type: "input.audio.committed",
|
||||
turnId,
|
||||
payload: { ...outputTalkPayload, source: event.type },
|
||||
final: true,
|
||||
});
|
||||
} else if (event.type === "response.done" || event.type === "response.cancelled") {
|
||||
if (outputOwner.terminal(event.responseId)) {
|
||||
outputGenerationActive = false;
|
||||
harness.finishOutputAudio(event.type);
|
||||
if (event.type === "response.done") {
|
||||
harness.endTurn(event.type);
|
||||
}
|
||||
}
|
||||
} else if (
|
||||
event.type === "error" &&
|
||||
event.detail === MEETING_REALTIME_CANCELLATION_RACE_DETAIL
|
||||
) {
|
||||
if (outputOwner.clearBlocked()) {
|
||||
outputGenerationActive = false;
|
||||
harness.finishOutputAudio(event.type);
|
||||
}
|
||||
} else if (event.type === "error") {
|
||||
harness.emit({
|
||||
type: "session.error",
|
||||
payload: { message: event.detail ?? "Realtime provider error" },
|
||||
final: true,
|
||||
});
|
||||
}
|
||||
if (
|
||||
event.type === "error" ||
|
||||
event.type === "response.done" ||
|
||||
event.type === "input_audio_buffer.speech_started" ||
|
||||
event.type === "input_audio_buffer.speech_stopped" ||
|
||||
event.type === "conversation.item.input_audio_transcription.completed" ||
|
||||
event.type === "conversation.item.input_audio_transcription.failed"
|
||||
) {
|
||||
const detail = event.detail ? ` ${event.detail}` : "";
|
||||
params.logger.info(
|
||||
`${params.platform.logScope} ${realtimeLogScope} ${event.direction}:${event.type}${detail}`,
|
||||
);
|
||||
}
|
||||
},
|
||||
onEvent: lifecycleHandlers.onEvent,
|
||||
onResponseDone: lifecycleHandlers.onResponseDone,
|
||||
onToolCall: (event, session) =>
|
||||
toolContinuity.run({
|
||||
session,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
createRealtimeVoiceAudioQueue,
|
||||
normalizeRealtimeVoiceResponseOutcome,
|
||||
RealtimeVoiceSessionLifecycle,
|
||||
type RealtimeVoiceSessionConnection,
|
||||
} from "./realtime-voice.js";
|
||||
@@ -236,6 +237,73 @@ describe("RealtimeVoiceSessionLifecycle", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("normalizeRealtimeVoiceResponseOutcome", () => {
|
||||
it.each([
|
||||
[
|
||||
{ id: "resp-complete", status: "completed" },
|
||||
{ responseId: "resp-complete", status: "completed" },
|
||||
],
|
||||
[
|
||||
{ id: "resp-cancel", status: "cancelled", status_details: { reason: "client_cancelled" } },
|
||||
{ responseId: "resp-cancel", status: "cancelled", reason: "client_cancelled" },
|
||||
],
|
||||
[
|
||||
{
|
||||
id: "resp-failed",
|
||||
status: "failed",
|
||||
status_details: {
|
||||
reason: "provider_error",
|
||||
error: { code: "rate_limit", type: "server_error", message: "slow down" },
|
||||
},
|
||||
},
|
||||
{
|
||||
responseId: "resp-failed",
|
||||
status: "failed",
|
||||
reason: "provider_error",
|
||||
error: { code: "rate_limit", type: "server_error", message: "slow down" },
|
||||
message: "Test response failed: provider_error: slow down",
|
||||
},
|
||||
],
|
||||
[
|
||||
{ status: "incomplete", status_details: { reason: "max_output_tokens" } },
|
||||
{
|
||||
responseId: "resp-fallback",
|
||||
status: "incomplete",
|
||||
reason: "max_output_tokens",
|
||||
message: "Test response incomplete: max_output_tokens",
|
||||
},
|
||||
],
|
||||
[
|
||||
{ id: "", status: "unexpected" },
|
||||
{
|
||||
responseId: "resp-fallback",
|
||||
status: "failed",
|
||||
reason: "invalid_response_status",
|
||||
error: { type: "invalid_response_status", message: "invalid status unexpected" },
|
||||
message: "Test response failed: invalid status unexpected",
|
||||
},
|
||||
],
|
||||
[
|
||||
undefined,
|
||||
{
|
||||
responseId: "resp-fallback",
|
||||
status: "failed",
|
||||
reason: "invalid_response_status",
|
||||
error: { type: "invalid_response_status", message: "missing terminal status" },
|
||||
message: "Test response failed: missing terminal status",
|
||||
},
|
||||
],
|
||||
])("normalizes %#", (response, expected) => {
|
||||
expect(
|
||||
normalizeRealtimeVoiceResponseOutcome({
|
||||
providerLabel: "Test",
|
||||
response,
|
||||
responseId: "resp-fallback",
|
||||
}),
|
||||
).toEqual(expected);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createRealtimeVoiceAudioQueue", () => {
|
||||
it("releases byte budget as queued audio is consumed", () => {
|
||||
const queue = createRealtimeVoiceAudioQueue("reject-newest");
|
||||
|
||||
@@ -17,12 +17,15 @@ export type {
|
||||
RealtimeVoiceProviderConfiguredContext,
|
||||
RealtimeVoiceProviderId,
|
||||
RealtimeVoiceProviderResolveConfigContext,
|
||||
RealtimeVoiceResponseError,
|
||||
RealtimeVoiceResponseOutcome,
|
||||
RealtimeVoiceRole,
|
||||
RealtimeVoiceTool,
|
||||
RealtimeVoiceToolCallEvent,
|
||||
RealtimeVoiceToolResultOptions,
|
||||
} from "../talk/provider-types.js";
|
||||
export {
|
||||
normalizeRealtimeVoiceResponseOutcome,
|
||||
REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ,
|
||||
REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ,
|
||||
} from "../talk/provider-types.js";
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
// Talk provider types describe realtime voice provider configuration and APIs.
|
||||
import { isRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import type { TalkTransport } from "./talk-events.js";
|
||||
|
||||
@@ -67,6 +69,81 @@ export type RealtimeVoiceBridgeEvent = {
|
||||
responseId?: string;
|
||||
};
|
||||
|
||||
export type RealtimeVoiceResponseError = {
|
||||
code?: string;
|
||||
message?: string;
|
||||
type?: string;
|
||||
};
|
||||
|
||||
type RealtimeVoiceResponseOutcomeBase = {
|
||||
responseId?: string;
|
||||
};
|
||||
|
||||
export type RealtimeVoiceResponseOutcome =
|
||||
| (RealtimeVoiceResponseOutcomeBase & { status: "completed" })
|
||||
| (RealtimeVoiceResponseOutcomeBase & { status: "cancelled"; reason?: string })
|
||||
| (RealtimeVoiceResponseOutcomeBase & {
|
||||
status: "failed" | "incomplete";
|
||||
reason?: string;
|
||||
error?: RealtimeVoiceResponseError;
|
||||
message: string;
|
||||
});
|
||||
|
||||
/** Normalizes OpenAI-style realtime response status details into the shared Talk contract. */
|
||||
export function normalizeRealtimeVoiceResponseOutcome(params: {
|
||||
providerLabel: string;
|
||||
response: unknown;
|
||||
responseId?: unknown;
|
||||
}): RealtimeVoiceResponseOutcome {
|
||||
const response = isRecord(params.response) ? params.response : undefined;
|
||||
const details = isRecord(response?.status_details) ? response.status_details : undefined;
|
||||
const rawError = isRecord(details?.error) ? details.error : undefined;
|
||||
const code = normalizeOptionalString(rawError?.code);
|
||||
const errorMessage = normalizeOptionalString(rawError?.message);
|
||||
const errorType = normalizeOptionalString(rawError?.type);
|
||||
const error =
|
||||
code || errorMessage || errorType
|
||||
? {
|
||||
...(code ? { code } : {}),
|
||||
...(errorMessage ? { message: errorMessage } : {}),
|
||||
...(errorType ? { type: errorType } : {}),
|
||||
}
|
||||
: undefined;
|
||||
const reason = normalizeOptionalString(details?.reason);
|
||||
const responseId =
|
||||
normalizeOptionalString(response?.id) ?? normalizeOptionalString(params.responseId);
|
||||
const base = responseId ? { responseId } : {};
|
||||
switch (response?.status) {
|
||||
case "completed":
|
||||
return { ...base, status: "completed" };
|
||||
case "cancelled":
|
||||
return { ...base, status: "cancelled", ...(reason ? { reason } : {}) };
|
||||
case "failed":
|
||||
case "incomplete": {
|
||||
const status = response.status;
|
||||
const detail = [reason, errorMessage ?? code ?? errorType].filter(Boolean).join(": ");
|
||||
return {
|
||||
...base,
|
||||
status,
|
||||
...(reason ? { reason } : {}),
|
||||
...(error ? { error } : {}),
|
||||
message: `${params.providerLabel} response ${status}${detail ? `: ${detail}` : ""}`,
|
||||
};
|
||||
}
|
||||
default: {
|
||||
const rawStatus = normalizeOptionalString(response?.status);
|
||||
const detail = rawStatus ? `invalid status ${rawStatus}` : "missing terminal status";
|
||||
return {
|
||||
...base,
|
||||
status: "failed",
|
||||
reason: "invalid_response_status",
|
||||
error: { type: "invalid_response_status", message: detail },
|
||||
message: `${params.providerLabel} response failed: ${detail}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export type RealtimeVoiceAudioClearReason = "barge-in";
|
||||
|
||||
export type RealtimeVoiceBridgeCallbacks = {
|
||||
@@ -75,6 +152,7 @@ export type RealtimeVoiceBridgeCallbacks = {
|
||||
onMark?: (markName: string) => void;
|
||||
onTranscript?: (role: RealtimeVoiceRole, text: string, isFinal: boolean) => void;
|
||||
onEvent?: (event: RealtimeVoiceBridgeEvent) => void;
|
||||
onResponseDone?: (outcome: RealtimeVoiceResponseOutcome) => void;
|
||||
onToolCall?: (event: RealtimeVoiceToolCallEvent) => void;
|
||||
onReady?: () => void;
|
||||
onError?: (error: Error) => void;
|
||||
|
||||
@@ -45,6 +45,98 @@ function makeBridge(overrides: Partial<RealtimeVoiceBridge> = {}): RealtimeVoice
|
||||
}
|
||||
|
||||
describe("realtime voice session harness", () => {
|
||||
it.each(["completed", "cancelled", "failed", "incomplete"] as const)(
|
||||
"settles one output span and turn for %s responses",
|
||||
(status) => {
|
||||
const harness = createHarness();
|
||||
harness.recordOutputAudio(Buffer.from([1, 2]));
|
||||
const outcome =
|
||||
status === "failed" || status === "incomplete"
|
||||
? ({ status, responseId: `resp-${status}`, message: `${status} message` } as const)
|
||||
: ({ status, responseId: `resp-${status}` } as const);
|
||||
|
||||
expect(harness.finishResponse(outcome).ok).toBe(true);
|
||||
expect(harness.finishResponse(outcome)).toEqual({ ok: false, reason: "no_active_turn" });
|
||||
expect(harness.talk.recentEvents.map((event) => event.type)).toEqual(
|
||||
status === "failed" || status === "incomplete"
|
||||
? [
|
||||
"turn.started",
|
||||
"output.audio.started",
|
||||
"output.audio.delta",
|
||||
"output.audio.done",
|
||||
"session.error",
|
||||
"turn.ended",
|
||||
]
|
||||
: [
|
||||
"turn.started",
|
||||
"output.audio.started",
|
||||
"output.audio.delta",
|
||||
"output.audio.done",
|
||||
status === "cancelled" ? "turn.cancelled" : "turn.ended",
|
||||
],
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("uses a legacy terminal event only when no typed outcome settled that response", () => {
|
||||
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
|
||||
const onResponseDone = vi.fn();
|
||||
const provider: RealtimeVoiceProviderPlugin = {
|
||||
id: "test",
|
||||
label: "Test",
|
||||
isConfigured: () => true,
|
||||
createBridge: (request) => {
|
||||
callbacks = request;
|
||||
return makeBridge();
|
||||
},
|
||||
};
|
||||
const harness = createHarness();
|
||||
harness.createBridge({
|
||||
provider,
|
||||
providerConfig: {},
|
||||
audioSink: { sendAudio: vi.fn() },
|
||||
onResponseDone,
|
||||
});
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.created", responseId: "resp-1" });
|
||||
callbacks?.onResponseDone?.({ status: "completed", responseId: "resp-1" });
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.done", responseId: "resp-1" });
|
||||
|
||||
expect(onResponseDone).toHaveBeenCalledOnce();
|
||||
expect(harness.talk.recentEvents.filter((event) => event.type === "turn.ended")).toHaveLength(
|
||||
1,
|
||||
);
|
||||
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.created", responseId: "resp-2" });
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.cancelled", responseId: "resp-2" });
|
||||
expect(onResponseDone).toHaveBeenLastCalledWith({
|
||||
status: "cancelled",
|
||||
responseId: "resp-2",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not let a delayed duplicate terminal event settle a newer turn", () => {
|
||||
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
|
||||
const provider: RealtimeVoiceProviderPlugin = {
|
||||
id: "test",
|
||||
label: "Test",
|
||||
isConfigured: () => true,
|
||||
createBridge: (request) => {
|
||||
callbacks = request;
|
||||
return makeBridge();
|
||||
},
|
||||
};
|
||||
const harness = createHarness();
|
||||
harness.createBridge({ provider, providerConfig: {}, audioSink: { sendAudio: vi.fn() } });
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.created", responseId: "resp-old" });
|
||||
callbacks?.onResponseDone?.({ status: "completed", responseId: "resp-old" });
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.created", responseId: "resp-new" });
|
||||
callbacks?.onEvent?.({ direction: "server", type: "response.done", responseId: "resp-old" });
|
||||
|
||||
expect(harness.talk.activeTurnId).toBeDefined();
|
||||
expect(harness.talk.recentEvents.filter((event) => event.type === "turn.ended")).toHaveLength(
|
||||
1,
|
||||
);
|
||||
});
|
||||
it("keeps shared Talk events ordered across input, output, and turn completion", () => {
|
||||
const harness = createHarness();
|
||||
|
||||
|
||||
@@ -14,7 +14,12 @@ import {
|
||||
type RealtimeVoiceOutputActivityDelta,
|
||||
type RealtimeVoiceOutputActivityTracker,
|
||||
} from "./output-activity-tracker.js";
|
||||
import type { RealtimeVoiceBargeInOptions, RealtimeVoiceRole } from "./provider-types.js";
|
||||
import type {
|
||||
RealtimeVoiceBargeInOptions,
|
||||
RealtimeVoiceBridgeEvent,
|
||||
RealtimeVoiceResponseOutcome,
|
||||
RealtimeVoiceRole,
|
||||
} from "./provider-types.js";
|
||||
import {
|
||||
extendRealtimeVoiceOutputEchoSuppression,
|
||||
getRealtimeVoiceBridgeEventHealth,
|
||||
@@ -35,8 +40,31 @@ import {
|
||||
createTalkSessionController,
|
||||
type TalkSessionController,
|
||||
type TalkSessionControllerParams,
|
||||
type TalkTurnResult,
|
||||
} from "./talk-session-controller.js";
|
||||
|
||||
const MAX_SETTLED_RESPONSE_IDS = 64;
|
||||
|
||||
type RealtimeVoiceHarnessResponseOwner = {
|
||||
claimResponseEvent(event: RealtimeVoiceBridgeEvent): void;
|
||||
finishLegacyEvent(event: RealtimeVoiceBridgeEvent): RealtimeVoiceResponseOutcome | undefined;
|
||||
};
|
||||
|
||||
const harnessResponseOwners = new WeakMap<
|
||||
RealtimeVoiceSessionHarness,
|
||||
RealtimeVoiceHarnessResponseOwner
|
||||
>();
|
||||
|
||||
/** Core-only adapter for direct provider bridges that cannot use createBridge(). */
|
||||
export function handleRealtimeVoiceHarnessBridgeEvent(
|
||||
harness: RealtimeVoiceSessionHarness,
|
||||
event: RealtimeVoiceBridgeEvent,
|
||||
): RealtimeVoiceResponseOutcome | undefined {
|
||||
const owner = harnessResponseOwners.get(harness);
|
||||
owner?.claimResponseEvent(event);
|
||||
return owner?.finishLegacyEvent(event);
|
||||
}
|
||||
|
||||
type RealtimeVoiceSessionHarnessTalkPayloads = {
|
||||
turnStarted: () => unknown;
|
||||
turnEnded: (reason: string) => unknown;
|
||||
@@ -86,6 +114,7 @@ export type RealtimeVoiceSessionHarness<TForcedConsultContext = unknown> = {
|
||||
emit<TPayload>(input: TalkEventInput<TPayload>): TalkEvent<TPayload>;
|
||||
ensureTurn(): string;
|
||||
endTurn(reason?: string): void;
|
||||
finishResponse(outcome: RealtimeVoiceResponseOutcome): TalkTurnResult;
|
||||
finishOutputAudio(reason: string): void;
|
||||
flushOutput(flush: () => void): void;
|
||||
getHealth(params: {
|
||||
@@ -120,6 +149,11 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
|
||||
let suppressInputUntilMs = 0;
|
||||
let lastOutputPlayableUntilMs = 0;
|
||||
let outputFlushGeneration = 0;
|
||||
let responseOwnerTurnId: string | undefined;
|
||||
let responseOwnerId: string | undefined;
|
||||
let suppressNextUnkeyedLegacyTerminal = false;
|
||||
const settledResponseIds = new Set<string>();
|
||||
const settledResponseIdOrder: string[] = [];
|
||||
const transcript: RealtimeVoiceTranscriptEntry[] = [];
|
||||
const bridgeEvents: RealtimeVoiceBridgeEventLogEntry[] = [];
|
||||
const outputActivity = createRealtimeVoiceOutputActivityTracker();
|
||||
@@ -144,7 +178,106 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
|
||||
})
|
||||
: undefined;
|
||||
|
||||
const ensureTurn = () => talk.ensureTurn({ payload: params.talkPayloads.turnStarted() }).turnId;
|
||||
const ensureTurn = () => {
|
||||
const turnId = talk.ensureTurn({ payload: params.talkPayloads.turnStarted() }).turnId;
|
||||
responseOwnerTurnId ??= turnId;
|
||||
return turnId;
|
||||
};
|
||||
|
||||
const rememberSettledResponse = (responseId: string | undefined): void => {
|
||||
if (!responseId || settledResponseIds.has(responseId)) {
|
||||
return;
|
||||
}
|
||||
settledResponseIds.add(responseId);
|
||||
settledResponseIdOrder.push(responseId);
|
||||
if (settledResponseIdOrder.length > MAX_SETTLED_RESPONSE_IDS) {
|
||||
const oldest = settledResponseIdOrder.shift();
|
||||
if (oldest) {
|
||||
settledResponseIds.delete(oldest);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const claimResponseEvent = (event: RealtimeVoiceBridgeEvent): void => {
|
||||
if (event.direction !== "server" || event.type !== "response.created") {
|
||||
return;
|
||||
}
|
||||
responseOwnerTurnId = ensureTurn();
|
||||
responseOwnerId = event.responseId;
|
||||
suppressNextUnkeyedLegacyTerminal = false;
|
||||
};
|
||||
|
||||
const finishResponse = (
|
||||
outcome: RealtimeVoiceResponseOutcome,
|
||||
source: "typed" | "legacy" | "manual",
|
||||
): TalkTurnResult => {
|
||||
if (outcome.responseId && settledResponseIds.has(outcome.responseId)) {
|
||||
return { ok: false, reason: "no_active_turn" };
|
||||
}
|
||||
if (outcome.responseId && responseOwnerId && outcome.responseId !== responseOwnerId) {
|
||||
return { ok: false, reason: "stale_turn" };
|
||||
}
|
||||
const turnId = responseOwnerTurnId ?? talk.activeTurnId;
|
||||
if (!turnId) {
|
||||
return { ok: false, reason: "no_active_turn" };
|
||||
}
|
||||
if (talk.activeTurnId !== turnId) {
|
||||
return { ok: false, reason: "stale_turn" };
|
||||
}
|
||||
talk.finishOutputAudio({
|
||||
turnId,
|
||||
payload: params.talkPayloads.outputAudioDone(outcome.status),
|
||||
});
|
||||
if (outcome.status === "failed" || outcome.status === "incomplete") {
|
||||
talk.emit({
|
||||
type: "session.error",
|
||||
turnId,
|
||||
payload: outcome,
|
||||
final: true,
|
||||
});
|
||||
}
|
||||
const payload = params.talkPayloads.turnEnded(outcome.status);
|
||||
const result =
|
||||
outcome.status === "cancelled"
|
||||
? talk.cancelTurn({ turnId, payload })
|
||||
: talk.endTurn({ turnId, payload });
|
||||
if (result.ok) {
|
||||
rememberSettledResponse(outcome.responseId);
|
||||
if (!outcome.responseId && source === "typed") {
|
||||
// Current typed providers emit the legacy bridge event in the same dispatch.
|
||||
// Suppress that unkeyed twin without treating arbitrary later events as typed.
|
||||
suppressNextUnkeyedLegacyTerminal = true;
|
||||
}
|
||||
if (!responseOwnerId || !outcome.responseId || responseOwnerId === outcome.responseId) {
|
||||
responseOwnerTurnId = undefined;
|
||||
responseOwnerId = undefined;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
};
|
||||
|
||||
const finishLegacyEvent = (
|
||||
event: RealtimeVoiceBridgeEvent,
|
||||
): RealtimeVoiceResponseOutcome | undefined => {
|
||||
if (
|
||||
event.direction !== "server" ||
|
||||
(event.type !== "response.done" && event.type !== "response.cancelled")
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
if (event.responseId && settledResponseIds.has(event.responseId)) {
|
||||
return undefined;
|
||||
}
|
||||
if (!event.responseId && suppressNextUnkeyedLegacyTerminal) {
|
||||
suppressNextUnkeyedLegacyTerminal = false;
|
||||
return undefined;
|
||||
}
|
||||
const outcome: RealtimeVoiceResponseOutcome = {
|
||||
status: event.type === "response.cancelled" ? "cancelled" : "completed",
|
||||
...(event.responseId ? { responseId: event.responseId } : {}),
|
||||
};
|
||||
return finishResponse(outcome, "legacy").ok ? outcome : undefined;
|
||||
};
|
||||
|
||||
const flushOutput = (flush: () => void): void => {
|
||||
outputFlushGeneration += 1;
|
||||
@@ -166,6 +299,8 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
|
||||
closed = true;
|
||||
talkback?.close();
|
||||
forcedConsults.clear();
|
||||
responseOwnerTurnId = undefined;
|
||||
responseOwnerId = undefined;
|
||||
},
|
||||
createBridge(bridgeParams) {
|
||||
bridge = createRealtimeVoiceBridgeSession({
|
||||
@@ -177,18 +312,35 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
|
||||
bridgeParams.onTranscript?.(role, text, isFinal);
|
||||
},
|
||||
onEvent: (event) => {
|
||||
claimResponseEvent(event);
|
||||
const legacyOutcome = finishLegacyEvent(event);
|
||||
if (legacyOutcome) {
|
||||
bridgeParams.onResponseDone?.(legacyOutcome);
|
||||
}
|
||||
if (params.captureBridgeEvents !== false) {
|
||||
recordRealtimeVoiceBridgeEvent(bridgeEvents, event);
|
||||
}
|
||||
bridgeParams.onEvent?.(event);
|
||||
},
|
||||
onResponseDone: (outcome) => {
|
||||
if (finishResponse(outcome, "typed").ok) {
|
||||
bridgeParams.onResponseDone?.(outcome);
|
||||
}
|
||||
},
|
||||
});
|
||||
return bridge;
|
||||
},
|
||||
emit: (input) => talk.emit(input),
|
||||
ensureTurn,
|
||||
endTurn(reason = "completed") {
|
||||
talk.endTurn({ payload: params.talkPayloads.turnEnded(reason) });
|
||||
const result = talk.endTurn({ payload: params.talkPayloads.turnEnded(reason) });
|
||||
if (result.ok) {
|
||||
responseOwnerTurnId = undefined;
|
||||
responseOwnerId = undefined;
|
||||
}
|
||||
},
|
||||
finishResponse(outcome) {
|
||||
return finishResponse(outcome, "typed");
|
||||
},
|
||||
finishOutputAudio(reason) {
|
||||
talk.finishOutputAudio({ payload: params.talkPayloads.outputAudioDone(reason) });
|
||||
@@ -290,5 +442,7 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
|
||||
recordTranscript: (role, text) => recordRealtimeVoiceTranscript(transcript, role, text),
|
||||
};
|
||||
|
||||
harnessResponseOwners.set(harness, { claimResponseEvent, finishLegacyEvent });
|
||||
|
||||
return harness;
|
||||
}
|
||||
|
||||
@@ -31,6 +31,33 @@ function expectBridgeRequest(
|
||||
}
|
||||
|
||||
describe("realtime voice bridge session runtime", () => {
|
||||
it("keeps response outcomes separate from session errors", () => {
|
||||
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
|
||||
const onResponseDone = vi.fn();
|
||||
const onError = vi.fn();
|
||||
const provider: RealtimeVoiceProviderPlugin = {
|
||||
id: "test",
|
||||
label: "Test",
|
||||
isConfigured: () => true,
|
||||
createBridge: (request) => {
|
||||
callbacks = request;
|
||||
return makeBridge();
|
||||
},
|
||||
};
|
||||
createRealtimeVoiceBridgeSession({
|
||||
provider,
|
||||
providerConfig: {},
|
||||
audioSink: { sendAudio: vi.fn() },
|
||||
onResponseDone,
|
||||
onError,
|
||||
});
|
||||
const outcome = { status: "failed", message: "response failed" } as const;
|
||||
|
||||
callbacks?.onResponseDone?.(outcome);
|
||||
|
||||
expect(onResponseDone).toHaveBeenCalledWith(outcome);
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
it("routes provider output through an open audio sink", () => {
|
||||
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
|
||||
const bridge = makeBridge();
|
||||
|
||||
@@ -9,6 +9,7 @@ import type {
|
||||
RealtimeVoiceCloseReason,
|
||||
RealtimeVoiceBridgeEvent,
|
||||
RealtimeVoiceProviderConfig,
|
||||
RealtimeVoiceResponseOutcome,
|
||||
RealtimeVoiceRole,
|
||||
RealtimeVoiceTool,
|
||||
RealtimeVoiceToolCallEvent,
|
||||
@@ -69,6 +70,7 @@ export type RealtimeVoiceBridgeSessionParams = {
|
||||
tools?: RealtimeVoiceTool[];
|
||||
onTranscript?: (role: RealtimeVoiceRole, text: string, isFinal: boolean) => void;
|
||||
onEvent?: (event: RealtimeVoiceBridgeEvent) => void;
|
||||
onResponseDone?: (outcome: RealtimeVoiceResponseOutcome) => void;
|
||||
onToolCall?: (
|
||||
event: RealtimeVoiceToolCallEvent,
|
||||
session: RealtimeVoiceBridgeSession,
|
||||
@@ -195,6 +197,7 @@ export function createRealtimeVoiceBridgeSession(
|
||||
},
|
||||
onTranscript: params.onTranscript,
|
||||
onEvent: params.onEvent,
|
||||
onResponseDone: params.onResponseDone,
|
||||
onToolCall: (event) => {
|
||||
if (!bridgeRef.current || !isAdmitting()) {
|
||||
return;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
// Control UI chat module owns low-level WebRTC offer and media-message helpers.
|
||||
import { normalizeRealtimeVoiceResponseOutcome } from "../../../../src/talk/provider-types.js";
|
||||
import type { RealtimeTalkWebRtcSdpSessionResult } from "./realtime-talk-shared.ts";
|
||||
import type { RealtimeTalkVideoFrame } from "./realtime-talk-video.ts";
|
||||
|
||||
@@ -17,8 +18,10 @@ export type RealtimeServerEvent = {
|
||||
arguments?: string;
|
||||
error?: unknown;
|
||||
response?: {
|
||||
id?: string;
|
||||
status?: string;
|
||||
status_details?: unknown;
|
||||
output?: unknown[];
|
||||
};
|
||||
item?: {
|
||||
id?: string;
|
||||
@@ -32,6 +35,56 @@ export type RealtimeServerEvent = {
|
||||
};
|
||||
};
|
||||
|
||||
export class RealtimeTalkResponseOutcomeOwner {
|
||||
private activeResponseId: string | undefined;
|
||||
private unkeyedSettled = false;
|
||||
private readonly settledResponseIds = new Set<string>();
|
||||
|
||||
constructor(private readonly maxSettledResponses: number) {}
|
||||
|
||||
start(responseId: string | undefined): void {
|
||||
this.activeResponseId = responseId;
|
||||
this.unkeyedSettled = false;
|
||||
}
|
||||
|
||||
finish(event: RealtimeServerEvent) {
|
||||
const outcome =
|
||||
event.type === "response.cancelled"
|
||||
? ({
|
||||
status: "cancelled",
|
||||
...(event.response?.id ? { responseId: event.response.id } : {}),
|
||||
} as const)
|
||||
: normalizeRealtimeVoiceResponseOutcome({
|
||||
providerLabel: "OpenAI realtime voice",
|
||||
response: event.response,
|
||||
});
|
||||
if (
|
||||
(outcome.responseId && this.settledResponseIds.has(outcome.responseId)) ||
|
||||
(!outcome.responseId && this.unkeyedSettled) ||
|
||||
(outcome.responseId &&
|
||||
this.activeResponseId !== undefined &&
|
||||
outcome.responseId !== this.activeResponseId)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const overflow =
|
||||
outcome.responseId !== undefined && this.settledResponseIds.size >= this.maxSettledResponses;
|
||||
if (outcome.responseId && !overflow) {
|
||||
this.settledResponseIds.add(outcome.responseId);
|
||||
} else if (!outcome.responseId) {
|
||||
this.unkeyedSettled = true;
|
||||
}
|
||||
this.activeResponseId = undefined;
|
||||
return { outcome, overflow };
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
this.activeResponseId = undefined;
|
||||
this.unkeyedSettled = false;
|
||||
this.settledResponseIds.clear();
|
||||
}
|
||||
}
|
||||
|
||||
type PendingOfferRequest = {
|
||||
controller: AbortController;
|
||||
timeout: ReturnType<typeof globalThis.setTimeout>;
|
||||
|
||||
@@ -61,6 +61,7 @@ function dispatchDescribeViewToolCall(
|
||||
data: JSON.stringify({
|
||||
type: "response.done",
|
||||
response: {
|
||||
id: `response-${ids.callId}`,
|
||||
status: "completed",
|
||||
output: [
|
||||
{
|
||||
|
||||
@@ -579,13 +579,13 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {
|
||||
|
||||
await transport.start();
|
||||
const peer = FakePeerConnection.instances[0];
|
||||
for (const type of [
|
||||
"input_audio_buffer.speech_started",
|
||||
"input_audio_buffer.speech_stopped",
|
||||
"response.created",
|
||||
"response.done",
|
||||
for (const event of [
|
||||
{ type: "input_audio_buffer.speech_started" },
|
||||
{ type: "input_audio_buffer.speech_stopped" },
|
||||
{ type: "response.created", response: { id: "response-1" } },
|
||||
{ type: "response.done", response: { id: "response-1", status: "completed" } },
|
||||
]) {
|
||||
peer?.channel.dispatchEvent(new MessageEvent("message", { data: JSON.stringify({ type }) }));
|
||||
peer?.channel.dispatchEvent(new MessageEvent("message", { data: JSON.stringify(event) }));
|
||||
}
|
||||
|
||||
expect(onStatus).toHaveBeenCalledWith("listening", "Speech detected");
|
||||
@@ -605,6 +605,49 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {
|
||||
transport.stop();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["cancelled", "turn.cancelled"],
|
||||
["failed", "turn.ended"],
|
||||
["incomplete", "turn.ended"],
|
||||
] as const)("keeps browser Talk reusable after a %s response", async (status, terminalType) => {
|
||||
stubAnswerSdpFetch();
|
||||
const onStatus = vi.fn();
|
||||
const onTalkEvent = vi.fn();
|
||||
const transport = createOpenAiTransport({}, { onStatus, onTalkEvent });
|
||||
await transport.start();
|
||||
const peer = FakePeerConnection.instances[0];
|
||||
const response = {
|
||||
id: "response-1",
|
||||
status,
|
||||
...(status === "failed"
|
||||
? { status_details: { error: { code: "provider_error" } } }
|
||||
: status === "incomplete"
|
||||
? { status_details: { reason: "max_output_tokens" } }
|
||||
: { status_details: { reason: "client_cancelled" } }),
|
||||
};
|
||||
dispatchRealtimeEvent(peer, { type: "response.created", response: { id: "response-1" } });
|
||||
dispatchRealtimeEvent(peer, { type: "response.done", response });
|
||||
dispatchRealtimeEvent(peer, { type: "response.done", response });
|
||||
dispatchRealtimeEvent(peer, { type: "response.created", response: { id: "response-2" } });
|
||||
dispatchRealtimeEvent(peer, {
|
||||
type: "response.done",
|
||||
response: { id: "response-2", status: "completed" },
|
||||
});
|
||||
|
||||
const terminalEvents = onTalkEvent.mock.calls
|
||||
.map(([event]) => event)
|
||||
.filter((event) => event.type === "turn.ended" || event.type === "turn.cancelled");
|
||||
expect(terminalEvents).toHaveLength(2);
|
||||
expect(terminalEvents[0]?.type).toBe(terminalType);
|
||||
expect(
|
||||
onTalkEvent.mock.calls
|
||||
.map(([event]) => event.type)
|
||||
.filter((type) => type === "session.error"),
|
||||
).toHaveLength(status === "cancelled" ? 0 : 1);
|
||||
expect(onStatus).toHaveBeenLastCalledWith("listening", undefined);
|
||||
transport.stop();
|
||||
});
|
||||
|
||||
it("emits common Talk transcript events from the OpenAI data channel", async () => {
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
import { captureRealtimeTalkVideoFrame } from "./realtime-talk-video.ts";
|
||||
import {
|
||||
RealtimeTalkWebRtcOfferExchange,
|
||||
RealtimeTalkResponseOutcomeOwner,
|
||||
realtimeTalkDataChannelMaxMessageSize,
|
||||
realtimeTalkImageEvent,
|
||||
type RealtimeServerEvent,
|
||||
@@ -49,6 +50,9 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
|
||||
private responseActive = false;
|
||||
private responseCreateInFlight = false;
|
||||
private responseCreatePending = false;
|
||||
private readonly responseOutcomes = new RealtimeTalkResponseOutcomeOwner(
|
||||
MAX_COMPLETED_TOOL_CALL_IDS,
|
||||
);
|
||||
private readonly completedToolCallIds = new Set<string>();
|
||||
private readonly offerExchange = new RealtimeTalkWebRtcOfferExchange();
|
||||
private mediaSetupController: AbortController | null = null;
|
||||
@@ -275,6 +279,7 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
|
||||
}
|
||||
this.consultAbortControllers.clear();
|
||||
this.completedToolCallIds.clear();
|
||||
this.responseOutcomes.reset();
|
||||
this.responseActive = false;
|
||||
this.responseCreateInFlight = false;
|
||||
this.responseCreatePending = false;
|
||||
@@ -392,30 +397,51 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
|
||||
case "response.created":
|
||||
this.responseActive = true;
|
||||
this.responseCreateInFlight = false;
|
||||
this.responseOutcomes.start(event.response?.id);
|
||||
this.ctx.callbacks.onStatus?.("thinking", "Generating response");
|
||||
return;
|
||||
case "response.cancelled":
|
||||
case "response.done":
|
||||
if (event.type === "response.done") {
|
||||
this.handleCompletedResponse(event);
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
case "response.done": {
|
||||
const terminal = this.responseOutcomes.finish(event);
|
||||
if (!terminal) {
|
||||
return;
|
||||
}
|
||||
const { outcome } = terminal;
|
||||
try {
|
||||
if (outcome.status === "completed") {
|
||||
this.handleCompletedResponse(event);
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (outcome.status === "failed" || outcome.status === "incomplete") {
|
||||
this.ctx.callbacks.onStatus?.("error", outcome.message);
|
||||
this.emitTalkEvent({
|
||||
type: "session.error",
|
||||
final: true,
|
||||
payload: outcome,
|
||||
});
|
||||
} else {
|
||||
this.ctx.callbacks.onStatus?.(
|
||||
"listening",
|
||||
outcome.status === "cancelled" ? "Response cancelled" : undefined,
|
||||
);
|
||||
}
|
||||
this.emitTalkEvent({
|
||||
type: outcome.status === "cancelled" ? "turn.cancelled" : "turn.ended",
|
||||
final: true,
|
||||
payload: outcome,
|
||||
});
|
||||
} finally {
|
||||
if (terminal.overflow) {
|
||||
this.failConnection("Realtime response session limit exceeded");
|
||||
}
|
||||
this.responseActive = false;
|
||||
this.responseCreateInFlight = false;
|
||||
this.flushPendingResponseCreate();
|
||||
}
|
||||
this.responseActive = false;
|
||||
this.responseCreateInFlight = false;
|
||||
this.ctx.callbacks.onStatus?.("listening", this.extractResponseStatus(event));
|
||||
this.emitTalkEvent({
|
||||
type: "turn.ended",
|
||||
final: true,
|
||||
payload: {
|
||||
status:
|
||||
event.response?.status ??
|
||||
(event.type === "response.cancelled" ? "cancelled" : "completed"),
|
||||
},
|
||||
});
|
||||
this.flushPendingResponseCreate();
|
||||
return;
|
||||
}
|
||||
case "error":
|
||||
this.responseCreateInFlight = false;
|
||||
this.ctx.callbacks.onStatus?.("error", this.extractErrorDetail(event.error));
|
||||
@@ -429,11 +455,6 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
|
||||
}
|
||||
}
|
||||
|
||||
private extractResponseStatus(event: RealtimeServerEvent): string | undefined {
|
||||
const status = event.response?.status;
|
||||
return status && status !== "completed" ? `Response ${status}` : undefined;
|
||||
}
|
||||
|
||||
private emitAssistantTranscript(event: RealtimeServerEvent, final: boolean): void {
|
||||
const text = final ? (event.transcript ?? event.text) : event.delta;
|
||||
if (!text) {
|
||||
|
||||
Reference in New Issue
Block a user