diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json index 37311b5fa037..65a27da36ca6 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json @@ -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"} diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json index fc3664f07489..b858637685a4 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json @@ -1 +1 @@ -{"contentHash":"02d7c4b0ccf053d70fcdae12cd41e6028a025ac4464d9ad01996d825f5558c6f","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} +{"contentHash":"a806b806c516e82741cf2731d0d0375736eda65dcf130f0a66f3a631b6b1b482","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-core.json b/docs/.generated/plugin-sdk-api-baseline/channel-core.json index 7c994f73594f..3456659edd9c 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-core.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-core.json @@ -1 +1 @@ -{"contentHash":"78ece577895a6a5cc4a5bbe4e2791d72f28ce7f53a349b6faa6eaf1e3bc90c72","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} +{"contentHash":"60730746ea4ad9d8aef60a7f4d4aecd7cd28ae5d28b21e44646b3fe1398e6c95","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json index b4ae54b4200c..7b0c8c32a5ca 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json @@ -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"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-message.json b/docs/.generated/plugin-sdk-api-baseline/channel-message.json index 92e00f489550..c7b39954c9e5 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-message.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-message.json @@ -1 +1 @@ -{"contentHash":"f7460f72ff844417e3ba3ca35fedb6b9d9f6f9d2cd1c4bb692bf3c7a8cd986fe","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} +{"contentHash":"f185748f1394f1d64653b21889954229bf0049e174526554e52c8e0bfc2c7dc5","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json index e421fde49065..0a1285782833 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json @@ -1 +1 @@ -{"contentHash":"6ddb91ab448c9471d4b96e86a687927ec237aa1fda9ac8d242caecf2863c12d4","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} +{"contentHash":"59cfee1296e172c295a550d4adbfff8776f10b73e07902394a68aa143efecabc","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json index b785909b2ecf..238879f79200 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json @@ -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"} diff --git a/docs/.generated/plugin-sdk-api-baseline/core.json b/docs/.generated/plugin-sdk-api-baseline/core.json index 25a8050c7e69..e75d0359d584 100644 --- a/docs/.generated/plugin-sdk-api-baseline/core.json +++ b/docs/.generated/plugin-sdk-api-baseline/core.json @@ -1 +1 @@ -{"contentHash":"b06e4402306dbb88968f7f6e03ea05ac04d9b07be18c3d6bd17657ce50db7543","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} +{"contentHash":"589c2e34ab4910b95df172434a94e7a74838280c31f0955319fd696be3f3bc78","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/discord.json b/docs/.generated/plugin-sdk-api-baseline/discord.json index 8a668609c946..42ed4d5a1c06 100644 --- a/docs/.generated/plugin-sdk-api-baseline/discord.json +++ b/docs/.generated/plugin-sdk-api-baseline/discord.json @@ -1 +1 @@ -{"contentHash":"34565f9ed0312937609380d0f4b1ebfb4955a79d0b1fc3644d7c25f143bec5f4","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} +{"contentHash":"55b33df13c3362c01f13580f9903900075a2bb50b05cae56533a3ce9a5170639","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} diff --git a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json index 41e4aa737a52..83719337658d 100644 --- a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json +++ b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json @@ -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"} diff --git a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json index 5c3f067ec52d..b1b7979b206f 100644 --- a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json @@ -1 +1 @@ -{"contentHash":"9c9caf0ad1b5ac92dd564c9a98b1f302855a5506f347a38da114804589bac187","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} +{"contentHash":"8d9f9721597b619189cbaa30747e10d2fb8adcf8e8404880b85683fb8fcebd47","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json index 2d332c85afb5..980e7fd34bb4 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json @@ -1 +1 @@ -{"contentHash":"79c5b931cceb7c1be3c95b9ca6e0362bdae5c6d6ee5e5aed97fcbdd9ea1062c1","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} +{"contentHash":"b25cfc5913de6cc44719995ec258226c7cb9fd1623afc277d75a5bc2e9870ccf","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json index 19d710eff4d4..742d19f8bfd6 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json @@ -1 +1 @@ -{"contentHash":"bc0d9ea4642d666045fe90464c8e5e34c0afc049d20d05823a5ddb4b52ae1c10","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} +{"contentHash":"0a01c88224f2d5dfd489f7478940032838e8246c9b00ff94fb6bcaf8d6bf84f5","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json index deabac351c92..41777f20eb3c 100644 --- a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json @@ -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"} diff --git a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json index a109659f37d7..dcf6b2700e4d 100644 --- a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json +++ b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json @@ -1 +1 @@ -{"contentHash":"b6c16456c26e0be9f02d0e420ee6054a48f529a9288c66e0e31e16b1367b6d5e","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} +{"contentHash":"92132ddbd26e3d81621f59004521f3875158b4704ee0b9e39f4798766f229702","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} diff --git a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json index 510ec354ddda..decd9ed10322 100644 --- a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json +++ b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json @@ -1 +1 @@ -{"contentHash":"1a2a81855860352fe59cf3162489fa1a93907c16b75d68460b610777e29120a0","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} +{"contentHash":"50ffb05c64c0f2898b3543900132261e152eda3999462154360efc3db7194359","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} diff --git a/extensions/discord/src/voice/manager.e2e.test-support.ts b/extensions/discord/src/voice/manager.e2e.test-support.ts index 736fdf97df96..ee212da44148 100644 --- a/extensions/discord/src/voice/manager.e2e.test-support.ts +++ b/extensions/discord/src/voice/manager.e2e.test-support.ts @@ -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, diff --git a/extensions/discord/src/voice/manager.e2e.test.ts b/extensions/discord/src/voice/manager.e2e.test.ts index 25648598cd8c..bcb0b382a185 100644 --- a/extensions/discord/src/voice/manager.e2e.test.ts +++ b/extensions/discord/src/voice/manager.e2e.test.ts @@ -286,7 +286,48 @@ vi.mock("openclaw/plugin-sdk/realtime-voice", async () => { return { ...harness, createBridge: (bridgeParams: Parameters[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[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(); diff --git a/extensions/discord/src/voice/realtime.ts b/extensions/discord/src/voice/realtime.ts index ed18f91ef12d..b6617a2e8b54 100644 --- a/extensions/discord/src/voice/realtime.ts +++ b/extensions/discord/src/voice/realtime.ts @@ -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, diff --git a/extensions/openai/realtime-voice-provider.live.test.ts b/extensions/openai/realtime-voice-provider.live.test.ts index 539de439389e..bfed35f60563 100644 --- a/extensions/openai/realtime-voice-provider.live.test.ts +++ b/extensions/openai/realtime-voice-provider.live.test.ts @@ -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((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; diff --git a/extensions/openai/realtime-voice-provider.ts b/extensions/openai/realtime-voice-provider.ts index 80e958ec4270..ee61e1b9385b 100644 --- a/extensions/openai/realtime-voice-provider.ts +++ b/extensions/openai/realtime-voice-provider.ts @@ -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, diff --git a/extensions/openai/realtime-voice-terminal-outcomes.test.ts b/extensions/openai/realtime-voice-terminal-outcomes.test.ts new file mode 100644 index 000000000000..33be996f8f95 --- /dev/null +++ b/extensions/openai/realtime-voice-terminal-outcomes.test.ts @@ -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((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +async function waitFor(promise: Promise, label: string): Promise { + let timeout: ReturnType | undefined; + try { + await Promise.race([ + promise, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error(`timed out waiting for ${label}`)), 2_000); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + +async function capture( + terminalEvent: Record, + options: { completeFollowup?: boolean; queueFollowup?: boolean; throwCallback?: boolean } = {}, +): Promise { + 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(); + 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((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((resolve) => { + wss.close(() => resolve()); + }); + await new Promise((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); + }); +}); diff --git a/extensions/voice-call/src/webhook/realtime-handler.test.ts b/extensions/voice-call/src/webhook/realtime-handler.test.ts index bb295b46cc67..96a094725f0f 100644 --- a/extensions/voice-call/src/webhook/realtime-handler.test.ts +++ b/extensions/voice-call/src/webhook/realtime-handler.test.ts @@ -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")); diff --git a/extensions/voice-call/src/webhook/realtime-handler.ts b/extensions/voice-call/src/webhook/realtime-handler.ts index 44855f01a671..f84ec179dd70 100644 --- a/extensions/voice-call/src/webhook/realtime-handler.ts +++ b/extensions/voice-call/src/webhook/realtime-handler.ts @@ -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", diff --git a/extensions/xai/lazy-capability-providers.ts b/extensions/xai/lazy-capability-providers.ts index 559242887cb5..522189afd520 100644 --- a/extensions/xai/lazy-capability-providers.ts +++ b/extensions/xai/lazy-capability-providers.ts @@ -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) } : {}), diff --git a/extensions/xai/realtime-voice-events.ts b/extensions/xai/realtime-voice-events.ts index eda0dba0e8ac..1962eb4a2e02 100644 --- a/extensions/xai/realtime-voice-events.ts +++ b/extensions/xai/realtime-voice-events.ts @@ -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(); private inputTranscriptReplacements = new Map(); 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 = ""; diff --git a/extensions/xai/realtime-voice-provider.test.ts b/extensions/xai/realtime-voice-provider.test.ts index fd3ecad23c0d..7c558b08cd22 100644 --- a/extensions/xai/realtime-voice-provider.test.ts +++ b/extensions/xai/realtime-voice-provider.test.ts @@ -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); diff --git a/extensions/xai/realtime-voice-terminal-outcomes.test.ts b/extensions/xai/realtime-voice-terminal-outcomes.test.ts index 675b0ab9a5fd..4a79a616d3d2 100644 --- a/extensions/xai/realtime-voice-terminal-outcomes.test.ts +++ b/extensions/xai/realtime-voice-terminal-outcomes.test.ts @@ -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) => void; + throwOnResponseDone?: boolean; }; async function waitForFixtureEvent(promise: Promise, label: string): Promise { @@ -35,7 +39,7 @@ async function captureRealtimeOutcome( options: CaptureRealtimeOutcomeOptions = {}, ): Promise { 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((resolve) => { markServerEventHandled = resolve; @@ -44,6 +48,10 @@ async function captureRealtimeOutcome( const responseCreatedHandled = new Promise((resolve) => { markResponseCreatedHandled = resolve; }); + let markQueuedResponseCompleted: () => void = () => {}; + const queuedResponseCompleted = new Promise((resolve) => { + markQueuedResponseCompleted = resolve; + }); const server = createServer(); const sockets = new Set(); 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] }); + } }); }); diff --git a/src/gateway/server-methods/talk-client.ts b/src/gateway/server-methods/talk-client.ts index 4c476cb0a47b..b84d45eeafea 100644 --- a/src/gateway/server-methods/talk-client.ts +++ b/src/gateway/server-methods/talk-client.ts @@ -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 = { diff --git a/src/gateway/talk-client-gateway-control.test.ts b/src/gateway/talk-client-gateway-control.test.ts index e90ef92a281c..75b5743c4fbb 100644 --- a/src/gateway/talk-client-gateway-control.test.ts +++ b/src/gateway/talk-client-gateway-control.test.ts @@ -14,7 +14,78 @@ function deferred() { 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), diff --git a/src/gateway/talk-client-gateway-control.ts b/src/gateway/talk-client-gateway-control.ts index e61255355ee9..2b5345b2e5a1 100644 --- a/src/gateway/talk-client-gateway-control.ts +++ b/src/gateway/talk-client-gateway-control.ts @@ -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; 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; - warn: (message: string) => void; }): GatewayControlOwner { let bridge: RealtimeVoiceBridge | undefined; let closeProvider: (() => Promise) | undefined; @@ -316,6 +322,33 @@ export function createTalkClientGatewayControlOwner(params: { const entryPrefix = `gateway-${randomUUID()}`; const consultQueue = createRealtimeControlQueue(); const consultControllers = new Map(); + 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 => { 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); } diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index 5ad9cea2ae9c..d0b88b719421 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -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) => { diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index 112b8b11c76b..6483f11b95e9 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -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); + const firstTalkEvents = firstPayloads + .map((payload) => payload.talkEvent) + .filter((event): event is Record => 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).type === "audioDone", + ), + ).toHaveLength(2); + expect(relaySessions.has(session.relaySessionId)).toBe(true); + expect(close).not.toHaveBeenCalled(); + }); + afterEach(async () => { for (const [relaySessionId, connId] of activeRelaySessions) { try { diff --git a/src/meeting-bot/realtime-engine-support.ts b/src/meeting-bot/realtime-engine-support.ts index 312a6024c872..d70cf8586a23 100644 --- a/src/meeting-bot/realtime-engine-support.ts +++ b/src/meeting-bot/realtime-engine-support.ts @@ -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; + 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 }; +} diff --git a/src/meeting-bot/realtime-engine.test.ts b/src/meeting-bot/realtime-engine.test.ts index 8506099f23f9..a0b2f578bca5 100644 --- a/src/meeting-bot/realtime-engine.test.ts +++ b/src/meeting-bot/realtime-engine.test.ts @@ -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 { diff --git a/src/meeting-bot/realtime-engine.ts b/src/meeting-bot/realtime-engine.ts index a12b29aba6b8..18288202bb17 100644 --- a/src/meeting-bot/realtime-engine.ts +++ b/src/meeting-bot/realtime-engine.ts @@ -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, diff --git a/src/plugin-sdk/realtime-voice.test.ts b/src/plugin-sdk/realtime-voice.test.ts index 4fb73be7b16d..60501b77a067 100644 --- a/src/plugin-sdk/realtime-voice.test.ts +++ b/src/plugin-sdk/realtime-voice.test.ts @@ -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"); diff --git a/src/plugin-sdk/realtime-voice.ts b/src/plugin-sdk/realtime-voice.ts index 63d6c9bf340c..3fc1626c3d12 100644 --- a/src/plugin-sdk/realtime-voice.ts +++ b/src/plugin-sdk/realtime-voice.ts @@ -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"; diff --git a/src/talk/provider-types.ts b/src/talk/provider-types.ts index 4b7c2ddc9b55..c2c40fc61c20 100644 --- a/src/talk/provider-types.ts +++ b/src/talk/provider-types.ts @@ -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; diff --git a/src/talk/realtime-session-harness.test.ts b/src/talk/realtime-session-harness.test.ts index 3769f050ff22..1c6e33043a97 100644 --- a/src/talk/realtime-session-harness.test.ts +++ b/src/talk/realtime-session-harness.test.ts @@ -45,6 +45,98 @@ function makeBridge(overrides: Partial = {}): 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[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[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(); diff --git a/src/talk/realtime-session-harness.ts b/src/talk/realtime-session-harness.ts index 94041a5bdafc..742ab5d7c376 100644 --- a/src/talk/realtime-session-harness.ts +++ b/src/talk/realtime-session-harness.ts @@ -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 = { emit(input: TalkEventInput): TalkEvent; 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(); + const settledResponseIdOrder: string[] = []; const transcript: RealtimeVoiceTranscriptEntry[] = []; const bridgeEvents: RealtimeVoiceBridgeEventLogEntry[] = []; const outputActivity = createRealtimeVoiceOutputActivityTracker(); @@ -144,7 +178,106 @@ export function createRealtimeVoiceSessionHarness 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 { + 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 recordRealtimeVoiceTranscript(transcript, role, text), }; + harnessResponseOwners.set(harness, { claimResponseEvent, finishLegacyEvent }); + return harness; } diff --git a/src/talk/session-runtime.test.ts b/src/talk/session-runtime.test.ts index bf5f58f878cd..3903eeee0cdf 100644 --- a/src/talk/session-runtime.test.ts +++ b/src/talk/session-runtime.test.ts @@ -31,6 +31,33 @@ function expectBridgeRequest( } describe("realtime voice bridge session runtime", () => { + it("keeps response outcomes separate from session errors", () => { + let callbacks: Parameters[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[0] | undefined; const bridge = makeBridge(); diff --git a/src/talk/session-runtime.ts b/src/talk/session-runtime.ts index 68b9a5c28593..43ed0ed60f3d 100644 --- a/src/talk/session-runtime.ts +++ b/src/talk/session-runtime.ts @@ -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; diff --git a/ui/src/pages/chat/realtime-talk-webrtc-support.ts b/ui/src/pages/chat/realtime-talk-webrtc-support.ts index 0c991070e1b2..298f91fe1163 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc-support.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc-support.ts @@ -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(); + + 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; diff --git a/ui/src/pages/chat/realtime-talk-webrtc-video.test.ts b/ui/src/pages/chat/realtime-talk-webrtc-video.test.ts index 091e810afe99..7193d483fd5c 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc-video.test.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc-video.test.ts @@ -61,6 +61,7 @@ function dispatchDescribeViewToolCall( data: JSON.stringify({ type: "response.done", response: { + id: `response-${ids.callId}`, status: "completed", output: [ { diff --git a/ui/src/pages/chat/realtime-talk-webrtc.test.ts b/ui/src/pages/chat/realtime-talk-webrtc.test.ts index 8119cc05ba90..12cabe58852b 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc.test.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc.test.ts @@ -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", diff --git a/ui/src/pages/chat/realtime-talk-webrtc.ts b/ui/src/pages/chat/realtime-talk-webrtc.ts index 66b948005bea..69250017b9e5 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc.ts @@ -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(); 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) {