diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index 75f01fd940a5b..7b8a68ade5d30 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -504,7 +504,7 @@ extensions/google/music-generation-provider.ts 1 extensions/google/oauth-token-shared.ts 1 extensions/google/onboard.ts 3 extensions/google/provider-registration.ts 2 -extensions/google/realtime-voice-provider.ts 6 +extensions/google/realtime-voice-provider.ts 5 extensions/google/transport-stream.ts 13 extensions/google/vertex-adc.ts 4 extensions/google/video-generation-provider.ts 13 @@ -2705,6 +2705,7 @@ src/gateway/server-methods/plugin-approval.ts 2 src/gateway/server-methods/plugin-host-hooks.ts 1 src/gateway/server-methods/question.ts 2 src/gateway/server-methods/restart-request.ts 8 +src/gateway/server-methods/send.ts 1 src/gateway/server-methods/session-catalog-terminal-start.ts 1 src/gateway/server-methods/session-catalog.ts 5 src/gateway/server-methods/session-create-initial-turn.ts 2 @@ -3230,7 +3231,7 @@ src/plugins/manifest-setup-normalizers.ts 5 src/plugins/manifest.ts 2 src/plugins/marketplace.ts 6 src/plugins/memory-runtime.ts 1 -src/plugins/native-module-require.ts 2 +src/plugins/native-module-require.ts 3 src/plugins/official-external-provider-endpoints.ts 1 src/plugins/openai-compatible-embedding-provider.ts 2 src/plugins/plugin-command-dispatch-contract.ts 1 diff --git a/docs/providers/google.md b/docs/providers/google.md index 4e34824bbe282..8c9444f696249 100644 --- a/docs/providers/google.md +++ b/docs/providers/google.md @@ -431,6 +431,19 @@ nearest supported level, while `-1` leaves Google's default in place. See the [Gemini Live capability comparison](https://ai.google.dev/gemini-api/docs/live-api/capabilities). + +Gemini 3.8 Live (`gemini-3.8-live`) keeps the async function-calling contract and +rejects any thinking config, so OpenClaw sends none for it. Gemini 3.8 Live Extended +Thinking (`gemini-3.8-live-extended-thinking`) requires `NON_BLOCKING` tools, rejects +function response scheduling, and abandons a call after an interim response, so OpenClaw +sends one final result per agent consult without a "working" interim. Configure its +reasoning depth with `thinkingLevel` (`low`, `medium`, or `high`; `minimal` maps to +`low`), or a positive `thinkingBudget` mapped to the nearest level. On this model an +explicit stop or barge-in interrupts generation through a client-content turn; other +Gemini Live models interrupt only through server-side voice activity detection. See the +[Gemini 3.8 Live thinking guide](https://ai.google.dev/gemini-api/docs/live-api/thinking). + + Control UI Talk supports Google Live browser sessions with constrained one-use tokens. In Video Talk, the browser sends bounded JPEG frames directly to diff --git a/extensions/google/realtime-voice-model-contract.ts b/extensions/google/realtime-voice-model-contract.ts new file mode 100644 index 0000000000000..5ffd36d884b0f --- /dev/null +++ b/extensions/google/realtime-voice-model-contract.ts @@ -0,0 +1,104 @@ +import { type ThinkingConfig, ThinkingLevel } from "@google/genai"; +import { resolveGoogleGemini3ThinkingLevel } from "./thinking-api.js"; + +/** The thinking-related provider settings the Live model contracts read. */ +export type GoogleLiveThinkingSettings = { + thinkingLevel?: "minimal" | "low" | "medium" | "high"; + thinkingBudget?: number; +}; + +function normalizeGoogleLiveModelId(model: string): string { + return model.startsWith("models/") ? model.slice("models/".length) : model; +} + +// Gemini 3.1 Live replaces client-content text and async tools with realtime text +// and sequential function responses; explicit older models keep their prior contract. +export function isGemini31LiveModel(model: string): boolean { + const modelId = normalizeGoogleLiveModelId(model); + return modelId.startsWith("gemini-3.1-") && modelId.includes("-live"); +} + +// Gemini 3.8 Live (`gemini-3.8-live`) keeps the async tool contract but closes the +// session (1007) on any thinking config. Gemini 3.8 Live Extended Thinking +// (`gemini-3.8-live-extended-thinking`) requires NON_BLOCKING tools, closes the session +// (1007) on function response scheduling, drops a call after a `willContinue` interim +// instead of waiting for the final result, and takes `thinkingLevel` low|medium|high. +// Both verified against the live API on 2026-09-19. +export function isGemini38LiveModel(model: string): boolean { + return normalizeGoogleLiveModelId(model).startsWith("gemini-3.8-live"); +} + +export function isGemini38LiveExtendedThinkingModel(model: string): boolean { + const modelId = normalizeGoogleLiveModelId(model); + return modelId.startsWith("gemini-3.8-live") && modelId.includes("extended-thinking"); +} + +export function supportsAsyncFunctionCalling(model: string): boolean { + return !isGemini31LiveModel(model); +} + +// Continuation means an interim `willContinue` response scheduled WHEN_IDLE before the +// final result. Extended Thinking rejects the scheduling and abandons the call after the +// interim, so it gets one final response per call like Gemini 3.1 Live. +export function modelSupportsToolResultContinuation(model: string): boolean { + return supportsAsyncFunctionCalling(model) && !isGemini38LiveExtendedThinkingModel(model); +} + +// Gemini Live has no response.cancel. Extended Thinking documents that a client-content +// turn with turnComplete interrupts active generation; the server then emits +// `interrupted` and `turnComplete`. Other Live models only interrupt through server-side +// VAD on input audio. +export function supportsClientContentInterrupt(model: string): boolean { + return isGemini38LiveExtendedThinkingModel(model); +} + +function resolveGemini38LiveExtendedThinkingLevel( + config: GoogleLiveThinkingSettings, +): ThinkingLevel | undefined { + switch (config.thinkingLevel) { + case "minimal": // MINIMAL is rejected by this model; LOW is the nearest supported level. + case "low": + return ThinkingLevel.LOW; + case "medium": + return ThinkingLevel.MEDIUM; + case "high": + return ThinkingLevel.HIGH; + default: + break; + } + if (typeof config.thinkingBudget !== "number" || config.thinkingBudget < 0) { + return undefined; + } + if (config.thinkingBudget <= 2048) { + return ThinkingLevel.LOW; + } + if (config.thinkingBudget <= 8192) { + return ThinkingLevel.MEDIUM; + } + return ThinkingLevel.HIGH; +} + +export function buildThinkingConfig( + config: GoogleLiveThinkingSettings, + model: string, +): ThinkingConfig | undefined { + if (isGemini31LiveModel(model)) { + const thinkingLevel = resolveGoogleGemini3ThinkingLevel({ + modelId: model, + thinkingLevel: config.thinkingLevel, + thinkingBudget: config.thinkingBudget, + }); + return thinkingLevel ? { thinkingLevel: ThinkingLevel[thinkingLevel] } : undefined; + } + if (isGemini38LiveModel(model)) { + if (!isGemini38LiveExtendedThinkingModel(model)) { + return undefined; + } + const thinkingLevel = resolveGemini38LiveExtendedThinkingLevel(config); + return thinkingLevel ? { thinkingLevel } : undefined; + } + if (typeof config.thinkingBudget === "number") { + return { thinkingBudget: config.thinkingBudget }; + } + return undefined; +} diff --git a/extensions/google/realtime-voice-provider.gemini38.test.ts b/extensions/google/realtime-voice-provider.gemini38.test.ts new file mode 100644 index 0000000000000..e0e6fc14fda81 --- /dev/null +++ b/extensions/google/realtime-voice-provider.gemini38.test.ts @@ -0,0 +1,242 @@ +// Google tests cover the Gemini 3.8 Live model contracts of the realtime voice provider. +import type { RealtimeVoiceTool } from "openclaw/plugin-sdk/realtime-voice"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + buildThinkingConfig, + isGemini38LiveExtendedThinkingModel, + isGemini38LiveModel, + modelSupportsToolResultContinuation, + supportsClientContentInterrupt, +} from "./realtime-voice-model-contract.js"; +import { buildGoogleRealtimeVoiceProvider } from "./realtime-voice-provider.js"; + +type MockGoogleLiveConnectParams = { + model: string; + config: Record; + callbacks: { + onopen: () => void; + onmessage: (message: Record) => void; + onerror: (event: { error?: unknown; message?: string }) => void; + onclose: (event?: { code?: number; reason?: string; wasClean?: boolean }) => void; + }; +}; + +const { connectMock, createGoogleGenAIMock, session } = vi.hoisted(() => { + const sessionValue = { + close: vi.fn(), + sendClientContent: vi.fn(), + sendRealtimeInput: vi.fn(), + sendToolResponse: vi.fn(), + }; + const connectMockLocal = vi.fn(async (_params: MockGoogleLiveConnectParams) => sessionValue); + const createGoogleGenAIMockLocal = vi.fn(() => ({ + authTokens: { create: vi.fn(async () => ({ name: "auth_tokens/browser-session" })) }, + live: { connect: connectMockLocal }, + })); + return { + connectMock: connectMockLocal, + createGoogleGenAIMock: createGoogleGenAIMockLocal, + session: sessionValue, + }; +}); + +vi.mock("./google-genai-runtime.js", () => ({ + createGoogleGenAI: createGoogleGenAIMock, +})); + +const ENV_KEYS = ["GEMINI_API_KEY", "GOOGLE_API_KEY"] as const; +let envSnapshot: Partial>; + +function lastConnectParams(): MockGoogleLiveConnectParams { + const params = connectMock.mock.calls.at(-1)?.[0]; + if (!params) { + throw new Error("expected google live connect call"); + } + return params; +} + +function consultDeclaration(params: MockGoogleLiveConnectParams) { + const config = params.config as { + tools?: Array<{ functionDeclarations?: Array<{ behavior?: string; name?: string }> }>; + }; + return config.tools?.[0]?.functionDeclarations?.[0]; +} + +function createRealtimeTool(name: string): RealtimeVoiceTool { + return { + type: "function", + name, + description: "Contract test tool", + parameters: { type: "object", properties: {} }, + }; +} + +type GoogleLiveBridgeParams = Parameters< + ReturnType["createBridge"] +>[0]; + +function createGoogleLiveBridge(params: Partial = {}) { + const { providerConfig, ...callbacks } = params; + return buildGoogleRealtimeVoiceProvider().createBridge({ + providerConfig: { apiKey: "gemini-key", ...providerConfig }, + onAudio: vi.fn(), + onClearAudio: vi.fn(), + ...callbacks, + }); +} + +async function openConfiguredBridge(params: Partial = {}) { + const bridge = createGoogleLiveBridge(params); + await bridge.connect(); + lastConnectParams().callbacks.onopen(); + lastConnectParams().callbacks.onmessage({ setupComplete: { sessionId: "session-1" } }); + return bridge; +} + +describe("Gemini 3.8 Live model contracts", () => { + it("classifies the 3.8 Live model ids", () => { + expect(isGemini38LiveModel("gemini-3.8-live")).toBe(true); + expect(isGemini38LiveModel("models/gemini-3.8-live-extended-thinking")).toBe(true); + expect(isGemini38LiveExtendedThinkingModel("gemini-3.8-live")).toBe(false); + expect(isGemini38LiveExtendedThinkingModel("gemini-3.8-live-extended-thinking")).toBe(true); + expect(isGemini38LiveModel("gemini-3.1-flash-live-preview")).toBe(false); + }); + + it("reserves interim tool responses and client-turn interrupts for the right models", () => { + expect(modelSupportsToolResultContinuation("gemini-3.8-live")).toBe(true); + expect(modelSupportsToolResultContinuation("gemini-3.8-live-extended-thinking")).toBe(false); + expect(modelSupportsToolResultContinuation("gemini-3.1-flash-live-preview")).toBe(false); + expect(supportsClientContentInterrupt("gemini-3.8-live-extended-thinking")).toBe(true); + expect(supportsClientContentInterrupt("gemini-3.8-live")).toBe(false); + }); + + it("builds thinking config per model family", () => { + expect(buildThinkingConfig({ thinkingLevel: "high" }, "gemini-3.8-live")).toBeUndefined(); + expect(buildThinkingConfig({ thinkingBudget: 512 }, "gemini-3.8-live")).toBeUndefined(); + expect(buildThinkingConfig({}, "gemini-3.8-live-extended-thinking")).toBeUndefined(); + expect( + buildThinkingConfig({ thinkingLevel: "minimal" }, "gemini-3.8-live-extended-thinking"), + ).toEqual({ thinkingLevel: "LOW" }); + expect( + buildThinkingConfig({ thinkingBudget: 4_096 }, "gemini-3.8-live-extended-thinking"), + ).toEqual({ thinkingLevel: "MEDIUM" }); + expect( + buildThinkingConfig({ thinkingBudget: -1 }, "gemini-3.8-live-extended-thinking"), + ).toBeUndefined(); + }); +}); + +describe("buildGoogleRealtimeVoiceProvider with Gemini 3.8 Live", () => { + beforeEach(() => { + envSnapshot = Object.fromEntries(ENV_KEYS.map((key) => [key, process.env[key]])); + connectMock.mockClear(); + createGoogleGenAIMock.mockClear(); + session.close.mockClear(); + session.sendClientContent.mockClear(); + session.sendRealtimeInput.mockClear(); + session.sendToolResponse.mockClear(); + delete process.env.GEMINI_API_KEY; + delete process.env.GOOGLE_API_KEY; + }); + + afterEach(() => { + vi.restoreAllMocks(); + for (const key of ENV_KEYS) { + const value = envSnapshot[key]; + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + }); + + it("uses the Gemini 3.8 Live Extended Thinking contract", async () => { + const onError = vi.fn(); + const bridge = createGoogleLiveBridge({ + providerConfig: { + model: "gemini-3.8-live-extended-thinking", + thinkingLevel: "high", + thinkingBudget: 8_193, + }, + tools: [createRealtimeTool("openclaw_agent_consult")], + onToolCall: vi.fn(), + onError, + }); + + // Extended Thinking rejects function response scheduling (close 1007) and abandons a + // call after a `willContinue` interim, so the relay must send one final response. + expect(bridge.supportsToolResultContinuation).toBe(false); + await bridge.connect(); + + const params = lastConnectParams(); + expect(params.model).toBe("gemini-3.8-live-extended-thinking"); + expect(params.config.thinkingConfig).toEqual({ thinkingLevel: "HIGH" }); + expect(consultDeclaration(params)).toMatchObject({ + name: "openclaw_agent_consult", + behavior: "NON_BLOCKING", + }); + + params.callbacks.onmessage({ + setupComplete: { sessionId: "session-1" }, + toolCall: { + functionCalls: [ + { id: "consult-call", name: "openclaw_agent_consult", args: { prompt: "hi" } }, + ], + }, + }); + expect(() => + bridge.submitToolResult("consult-call", { status: "working" }, { willContinue: true }), + ).toThrow("does not support continuing tool responses"); + expect(session.sendToolResponse).not.toHaveBeenCalled(); + + void bridge.submitToolResult("consult-call", { text: "The meeting starts at 3." }); + expect(session.sendToolResponse).toHaveBeenCalledWith({ + functionResponses: [ + { + id: "consult-call", + name: "openclaw_agent_consult", + response: { text: "The meeting starts at 3." }, + }, + ], + }); + }); + + it("omits thinking config for Gemini 3.8 Live and keeps its async consult contract", async () => { + const bridge = createGoogleLiveBridge({ + providerConfig: { model: "gemini-3.8-live", thinkingLevel: "high", thinkingBudget: 8_193 }, + tools: [createRealtimeTool("openclaw_agent_consult")], + onToolCall: vi.fn(), + }); + + expect(bridge.supportsToolResultContinuation).toBe(true); + await bridge.connect(); + + const params = lastConnectParams(); + expect(params.model).toBe("gemini-3.8-live"); + expect(params.config).not.toHaveProperty("thinkingConfig"); + expect(consultDeclaration(params)).toMatchObject({ + name: "openclaw_agent_consult", + behavior: "NON_BLOCKING", + }); + }); + + it("interrupts Gemini 3.8 Live Extended Thinking output with an empty client turn on barge-in", async () => { + const bridge = await openConfiguredBridge({ + providerConfig: { model: "gemini-3.8-live-extended-thinking" }, + }); + + bridge.handleBargeIn?.({ audioPlaybackActive: false }); + expect(session.sendClientContent).not.toHaveBeenCalled(); + + bridge.handleBargeIn?.({ audioPlaybackActive: true }); + expect(session.sendClientContent).toHaveBeenCalledWith({ turnComplete: true }); + }); + + it("leaves barge-in to server-side VAD on other Gemini Live models", async () => { + const bridge = await openConfiguredBridge({ providerConfig: { model: "gemini-3.8-live" } }); + + bridge.handleBargeIn?.({ audioPlaybackActive: true, force: true }); + expect(session.sendClientContent).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/google/realtime-voice-provider.ts b/extensions/google/realtime-voice-provider.ts index 15a55ae0d4e04..f5dbc7c17140f 100644 --- a/extensions/google/realtime-voice-provider.ts +++ b/extensions/google/realtime-voice-provider.ts @@ -14,7 +14,6 @@ import { type RealtimeInputConfig, type Session, StartSensitivity, - type ThinkingConfig, TurnCoverage, } from "@google/genai"; import { @@ -33,6 +32,7 @@ import type { RealtimeVoiceRole, RealtimeVoiceTool, RealtimeVoiceToolResultOptions, + RealtimeVoiceBargeInOptions, } from "openclaw/plugin-sdk/realtime-voice"; import { convertPcmToMulaw8k, @@ -60,7 +60,13 @@ import { GOOGLE_REALTIME_DEFAULT_MODEL, GOOGLE_REALTIME_VOICE_METADATA, } from "./realtime-voice-metadata.js"; -import { resolveGoogleGemini3ThinkingLevel } from "./thinking-api.js"; +import { + buildThinkingConfig, + isGemini31LiveModel, + modelSupportsToolResultContinuation, + supportsAsyncFunctionCalling, + supportsClientContentInterrupt, +} from "./realtime-voice-model-contract.js"; const GOOGLE_REALTIME_DEFAULT_VOICE = "Kore"; const GOOGLE_REALTIME_DEFAULT_API_VERSION = "v1beta"; @@ -241,37 +247,6 @@ function resolveEnvApiKey(): string | undefined { return trimToUndefined(process.env.GEMINI_API_KEY) ?? trimToUndefined(process.env.GOOGLE_API_KEY); } -// Gemini 3.1 Live replaces client-content text and async tools with realtime text -// and sequential function responses; explicit older models keep their prior contract. -function isGemini31LiveModel(model: string): boolean { - const modelId = model.startsWith("models/") ? model.slice("models/".length) : model; - return modelId.startsWith("gemini-3.1-") && modelId.includes("-live"); -} - -function supportsAsyncFunctionCalling(model: string): boolean { - return !isGemini31LiveModel(model); -} - -function buildThinkingConfig( - config: GoogleRealtimeLiveConfig, - model: string, -): ThinkingConfig | undefined { - if (isGemini31LiveModel(model)) { - const thinkingLevel = resolveGoogleGemini3ThinkingLevel({ - modelId: model, - thinkingLevel: config.thinkingLevel, - thinkingBudget: config.thinkingBudget, - }); - return thinkingLevel - ? { thinkingLevel: thinkingLevel as ThinkingConfig["thinkingLevel"] } - : undefined; - } - if (typeof config.thinkingBudget === "number") { - return { thinkingBudget: config.thinkingBudget }; - } - return undefined; -} - function buildRealtimeInputConfig( config: GoogleRealtimeLiveConfig, ): RealtimeInputConfig | undefined { @@ -483,7 +458,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { constructor(private readonly config: GoogleRealtimeVoiceBridgeConfig) { this.audioFormat = config.audioFormat ?? REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ; this.model = config.model ?? GOOGLE_REALTIME_DEFAULT_MODEL; - this.supportsToolResultContinuation = supportsAsyncFunctionCalling(this.model); + this.supportsToolResultContinuation = modelSupportsToolResultContinuation(this.model); } async connect(): Promise { @@ -675,6 +650,28 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { setMediaTimestamp(_ts: number): void {} + handleBargeIn(options?: RealtimeVoiceBargeInOptions): void { + if (!supportsClientContentInterrupt(this.model)) { + return; + } + if (!this.session || !this.connected || !this.sessionConfigured) { + return; + } + if (options?.audioPlaybackActive !== true && options?.force !== true) { + return; + } + try { + // An empty client turn: `{"clientContent":{"turnComplete":true}}`. The server answers + // with `interrupted` (cleared audio) and `turnComplete` (response done) and starts no + // new generation. + this.session.sendClientContent({ turnComplete: true }); + } catch (error) { + this.config.onError?.( + error instanceof Error ? error : new Error("Google Live barge-in interrupt failed"), + ); + } + } + sendUserMessage(text: string): void { const normalized = text.trim(); if (!normalized || !this.session || !this.connected || !this.sessionConfigured) { diff --git a/src/gateway/talk/relay/agent-consult.test.ts b/src/gateway/talk/relay/agent-consult.test.ts new file mode 100644 index 0000000000000..e1c048ff85522 --- /dev/null +++ b/src/gateway/talk/relay/agent-consult.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it, vi } from "vitest"; +import type { RealtimeVoiceAgentConsultRunner } from "../../../talk/provider-types.js"; +import { bindTalkRealtimeRelayAgentConsult } from "./agent-consult.js"; +import type { RelaySession } from "./state.js"; + +function createRunPrompt(run: () => Promise<{ text: string }>) { + return Object.assign(vi.fn(run), { + adoptCompletionClaims: vi.fn(), + claimAppend: vi.fn(() => true), + claimFailureAppend: vi.fn(() => true), + }); +} + +describe("bindTalkRealtimeRelayAgentConsult", () => { + it("holds assistant transcript appends for the whole consult and releases them once it settles", async () => { + const relay = {} as RelaySession; + let resolveRun!: (value: { text: string }) => void; + const pendingRun = new Promise<{ text: string }>((resolve) => { + resolveRun = resolve; + }); + const runAgentConsult = bindTalkRealtimeRelayAgentConsult( + createRunPrompt(() => pendingRun) as never, + () => relay, + async () => {}, + ); + + const run = runAgentConsult({ prompt: "what is on my calendar" } as never); + await vi.waitFor(() => expect(relay.assistantTranscriptHold?.depth).toBe(1)); + resolveRun({ text: "done" }); + await expect(run).resolves.toEqual({ text: "done" }); + expect(relay.assistantTranscriptHold).toBeUndefined(); + }); + + it("releases the hold when the consult fails", async () => { + const relay = {} as RelaySession; + const runAgentConsult = bindTalkRealtimeRelayAgentConsult( + createRunPrompt(async () => { + throw new Error("consult failed"); + }) as never, + () => relay, + async () => {}, + ); + + await expect(runAgentConsult({ prompt: "again" } as never)).rejects.toThrow("consult failed"); + expect(relay.assistantTranscriptHold).toBeUndefined(); + }); + + it("does not run a consult once the relay is gone", async () => { + const runPrompt = createRunPrompt(async () => ({ text: "done" })); + const runAgentConsult = bindTalkRealtimeRelayAgentConsult( + runPrompt as never, + () => undefined, + async () => {}, + ); + + await expect(runAgentConsult({ prompt: "hi" } as never)).rejects.toThrow( + "Realtime gateway-relay session is closed", + ); + expect(runPrompt).not.toHaveBeenCalled(); + }); +}); diff --git a/src/gateway/talk/relay/agent-consult.ts b/src/gateway/talk/relay/agent-consult.ts index a0855297f3c03..0fcfd0d969ebd 100644 --- a/src/gateway/talk/relay/agent-consult.ts +++ b/src/gateway/talk/relay/agent-consult.ts @@ -1,5 +1,7 @@ import type { RealtimeVoiceAgentConsultRunner } from "../../../talk/provider-types.js"; import type { TalkAgentConsultRequest } from "../client-agent-consult.types.js"; +import type { RelaySession } from "./state.js"; +import { beginRelayAssistantTranscriptHold } from "./voice.js"; type RelayAgentConsultRunner = RealtimeVoiceAgentConsultRunner & { adoptCompletionClaims: () => void; @@ -11,9 +13,10 @@ type RelayAgentConsultRunner = RealtimeVoiceAgentConsultRunner & { export function bindTalkRealtimeRelayAgentConsult( runPrompt: RelayAgentConsultRunner, - isCurrent: () => boolean, + getRelay: () => RelaySession | undefined, waitForTranscript: (signal?: AbortSignal) => Promise, ) { + const isCurrent = () => getRelay() !== undefined; const runAgentConsult = async (request: TalkAgentConsultRequest) => { if (!isCurrent()) { throw new Error("Realtime gateway-relay session is closed"); @@ -22,7 +25,14 @@ export function bindTalkRealtimeRelayAgentConsult( if (!isCurrent()) { throw new Error("Realtime gateway-relay session is closed"); } - return await runPrompt(request); + // Voice finals that land while the consult's keyed user turn awaits adoption would + // move the session anchor and fail the run; hold their session appends until it settles. + const release = beginRelayAssistantTranscriptHold(getRelay()); + try { + return await runPrompt(request); + } finally { + release(); + } }; const steer = runPrompt.steer; const lifecycleMethods = { @@ -48,7 +58,12 @@ export function bindTalkRealtimeRelayAgentConsult( if (!isCurrent()) { throw new Error("Realtime relay session is no longer active"); } - return await steer(request); + const release = beginRelayAssistantTranscriptHold(getRelay()); + try { + return await steer(request); + } finally { + release(); + } }, } : {}), diff --git a/src/gateway/talk/relay/cancellation-deadline.ts b/src/gateway/talk/relay/cancellation-deadline.ts new file mode 100644 index 0000000000000..ea4188472ec65 --- /dev/null +++ b/src/gateway/talk/relay/cancellation-deadline.ts @@ -0,0 +1,43 @@ +import { relaySessions, type RelaySession } from "./state.js"; + +/** How long a cancelled turn waits for the provider to confirm the cancelled response. */ +export const TURN_BOUND_CANCELLATION_DRAIN_MS = 1_000; +/** Upper bound for discarding a stale generation whose provider never reports it done. */ +const STALE_OUTPUT_DISCARD_MAX_MS = 30_000; + +/** + * Bounds a cancellation drain. When the provider never confirms the cancelled response, + * the relay completes the cancellation locally instead of closing the session. + * + * Gemini Live has no response.cancel: its bridge interrupts through input audio + * (server-side VAD) or, on Extended Thinking, a client turn, and only reports the response + * done at its own turnComplete. Closing at the deadline ended every Google relay call on + * the first stop or barge-in. Completing locally releases microphone audio and tool + * results, and the stale generation's remaining output is discarded until the provider + * reports it done or a replacement response starts. + */ +export function scheduleRelayCancellationDeadline( + session: RelaySession, + params: { turnId: string; reason: string; terminalEpoch: number }, +): void { + setTimeout(() => { + if ( + relaySessions.get(session.id) !== session || + session.toolResultEpoch !== params.terminalEpoch || + session.outputOwnership.phase !== "cancelling" + ) { + return; + } + if (!session.outputOwnership.completeCancellationLocally()) { + return; + } + session.context.logGateway.warn( + `talk relay: provider did not confirm output cancellation within ${TURN_BOUND_CANCELLATION_DRAIN_MS}ms; keeping the session open and discarding stale output (reason=${params.reason}, turnId=${params.turnId})`, + ); + setTimeout(() => { + if (relaySessions.get(session.id) === session && session.outputOwnership.discarding) { + session.outputOwnership.discarding = false; + } + }, STALE_OUTPUT_DISCARD_MAX_MS).unref?.(); + }, TURN_BOUND_CANCELLATION_DRAIN_MS).unref?.(); +} diff --git a/src/gateway/talk/relay/cancellation-recovery.test.ts b/src/gateway/talk/relay/cancellation-recovery.test.ts new file mode 100644 index 0000000000000..f6ab0e93a1af2 --- /dev/null +++ b/src/gateway/talk/relay/cancellation-recovery.test.ts @@ -0,0 +1,253 @@ +/** + * Tests relay output cancellation when the provider cannot confirm the cancelled response. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createDeferred } from "../../../../test/helpers/promise.js"; +import type { OpenClawConfig } from "../../../config/types.js"; +import type { RealtimeVoiceProviderPlugin } from "../../../plugins/types.js"; +import { resetClientVoiceConfirmationStateForTest } from "../../../talk/client-voice-confirmation.test-support.js"; +import { ensureClientVoiceAgentSessionEntry } from "../../../talk/client-voice-session.js"; +import { clientVoiceSessionTesting } from "../../../talk/client-voice-session.test-support.js"; +import { resolveRealtimeVoiceProviderCapabilities } from "../../../talk/provider-resolver.js"; +import type { + RealtimeVoiceBridge, + RealtimeVoiceBridgeCreateRequest, +} from "../../../talk/provider-types.js"; +import { + createOpenClawTestState, + type OpenClawTestState, +} from "../../../test-utils/openclaw-test-state.js"; +import { prepareTalkSessionTarget } from "../session-target.js"; +import { + cancelTalkRealtimeRelayTurn, + createTalkRealtimeRelaySession, + sendTalkRealtimeRelayAudio, + stopTalkRealtimeRelaySession, +} from "./index.js"; +import { drainingRelaySessions, relaySessions } from "./state.js"; + +const activeRelaySessions = new Map(); + +function makeRelayTransport(overrides: Partial = {}) { + return { + connect: vi.fn(async () => undefined), + sendAudio: vi.fn(), + setMediaTimestamp: vi.fn(), + handleBargeIn: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + isConnected: vi.fn(() => true), + ...overrides, + }; +} + +function createRelayFixture(transportOverrides: Partial = {}) { + let request: RealtimeVoiceBridgeCreateRequest | undefined; + const transport = makeRelayTransport(transportOverrides); + const provider: RealtimeVoiceProviderPlugin = { + id: "relay-test", + label: "Relay Test", + isConfigured: () => true, + createBridge: (bridgeRequest) => { + request = bridgeRequest; + return transport; + }, + }; + const broadcastToConnIds = vi.fn(); + const warn = vi.fn(); + const cfg = { agents: { entries: { main: { default: true } } } } as OpenClawConfig; + const capabilities = resolveRealtimeVoiceProviderCapabilities({ + provider, + providerConfig: {}, + cfg, + surface: "gateway-relay", + }); + const session = createTalkRealtimeRelaySession({ + context: { + broadcastToConnIds, + broadcast: vi.fn(), + logGateway: { warn }, + chatAbortControllers: new Map(), + } as never, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + controlSource: capabilities?.handlesAgentConsult === true ? "delegation" : "transcript", + capabilities, + cfg, + sessionTarget: prepareTalkSessionTarget(cfg, "agent:main:main"), + }); + activeRelaySessions.set(session.relaySessionId, "conn-1"); + const relay = relaySessions.get(session.relaySessionId); + if (!request || !relay) { + throw new Error("expected the relay to create its bridge"); + } + const payloadsOfType = (type: string) => + broadcastToConnIds.mock.calls + .map(([, payload]) => payload) + .filter( + (payload): payload is Record => + typeof payload === "object" && + payload !== null && + (payload as Record).type === type, + ); + return { + relaySessionId: session.relaySessionId, + relay, + request, + transport, + warn, + payloadsOfType, + }; +} + +function ensureActiveRelayTurnId(relaySessionId: string): string { + const relay = relaySessions.get(relaySessionId); + if (!relay) { + throw new Error(`Missing relay test session ${relaySessionId}`); + } + if (!relay.harness.talk.activeTurnId) { + relay.harness.talk.startTurn({ turnId: "turn-1" }); + } + return relay.harness.talk.activeTurnId ?? "turn-1"; +} + +describe("talk realtime relay cancellation recovery", () => { + let testState: OpenClawTestState | undefined; + + beforeEach(async () => { + testState = await createOpenClawTestState({ + label: "talk-realtime-relay-cancellation", + scenario: "minimal", + }); + await ensureClientVoiceAgentSessionEntry({ agentId: "main", sessionKey: "agent:main:main" }); + }); + + afterEach(async () => { + try { + for (const [relaySessionId, connId] of activeRelaySessions) { + try { + await stopTalkRealtimeRelaySession({ relaySessionId, connId }); + } catch (error) { + if ( + !(error instanceof Error) || + !error.message.includes("Unknown realtime relay session") + ) { + throw error; + } + } + } + await Promise.all( + [...drainingRelaySessions].map( + (session) => + session.closing?.completion ?? session.voiceSessionClose ?? Promise.resolve(), + ), + ); + } finally { + activeRelaySessions.clear(); + vi.useRealTimers(); + clientVoiceSessionTesting.reset(); + resetClientVoiceConfirmationStateForTest(); + await testState?.cleanup(); + testState = undefined; + } + }); + + it("keeps a stalled turn-bound cancellation open after its drain deadline and discards the stale generation", async () => { + vi.useFakeTimers(); + const pending = createDeferred(); + const fixture = createRelayFixture({ submitToolResult: vi.fn(() => pending.promise) }); + const { relaySessionId, relay, request, transport, payloadsOfType } = fixture; + + let cancellationSettled = false; + const cancellation = cancelTalkRealtimeRelayTurn({ + relaySessionId, + connId: "conn-1", + reason: "android-stop-tts", + turnId: ensureActiveRelayTurnId(relaySessionId), + }); + void cancellation.then(() => (cancellationSettled = true)); + const pendingAudio = Promise.resolve( + sendTalkRealtimeRelayAudio({ relaySessionId, connId: "conn-1", audioBase64: "AQI=" }), + ); + let audioSettled = false; + void pendingAudio.then( + () => (audioSettled = true), + () => (audioSettled = true), + ); + await vi.advanceTimersByTimeAsync(999); + expect(relaySessions.has(relaySessionId)).toBe(true); + expect(cancellationSettled).toBe(false); + expect(audioSettled).toBe(false); + expect(transport.sendAudio).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); + // The session stays open, microphone audio flows again, and the stall is logged. + await expect(pendingAudio).resolves.toBeUndefined(); + expect(transport.sendAudio).toHaveBeenCalledOnce(); + expect(relaySessions.has(relaySessionId)).toBe(true); + expect(transport.close).not.toHaveBeenCalled(); + expect(fixture.warn).toHaveBeenCalledWith( + expect.stringContaining("did not confirm output cancellation"), + ); + + // Output from the interrupted generation is dropped until the provider reports it done. + const audioBefore = payloadsOfType("audio").length; + const transcriptsBefore = payloadsOfType("transcript").length; + request.onAudio(Buffer.from("stale audio")); + request.onTranscript?.("assistant", "stale words", true); + request.onToolCall?.({ + itemId: "stale-item", + callId: "stale-call", + name: "custom_tool", + args: {}, + }); + expect(payloadsOfType("audio")).toHaveLength(audioBefore); + expect(payloadsOfType("transcript")).toHaveLength(transcriptsBefore); + expect(payloadsOfType("toolCall")).toHaveLength(0); + expect(relay.outputOwnership.discarding).toBe(true); + + request.onResponseDone?.({ status: "cancelled" }); + expect(relay.outputOwnership.discarding).toBe(false); + // The phone captures continuously; the next microphone frame re-arms a turn for the reply. + await sendTalkRealtimeRelayAudio({ relaySessionId, connId: "conn-1", audioBase64: "AQI=" }); + request.onAudio(Buffer.from("fresh audio")); + expect(payloadsOfType("audio")).toHaveLength(audioBefore + 1); + expect(relaySessions.has(relaySessionId)).toBe(true); + pending.resolve(); + }); + + it("keeps an exact-response relay open when cancellation is never confirmed", async () => { + vi.useFakeTimers(); + const { relaySessionId, relay, request, transport, payloadsOfType } = createRelayFixture(); + request.onEvent?.({ direction: "server", type: "response.created", responseId: "response-1" }); + + let cancellationSettled = false; + const cancellation = cancelTalkRealtimeRelayTurn({ + relaySessionId, + connId: "conn-1", + turnId: ensureActiveRelayTurnId(relaySessionId), + }); + void cancellation.then(() => (cancellationSettled = true)); + await vi.advanceTimersByTimeAsync(999); + expect(relaySessions.has(relaySessionId)).toBe(true); + expect(cancellationSettled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); + expect(relaySessions.has(relaySessionId)).toBe(true); + expect(transport.close).not.toHaveBeenCalled(); + + // The stale response's late audio is discarded until a replacement response starts. + const before = payloadsOfType("audio").length; + request.onAudio(Buffer.from("stale audio")); + expect(payloadsOfType("audio")).toHaveLength(before); + relay.harness.talk.startTurn({ turnId: "turn-next" }); + request.onEvent?.({ direction: "server", type: "response.created", responseId: "response-2" }); + request.onAudio(Buffer.from("fresh audio")); + expect(payloadsOfType("audio")).toHaveLength(before + 1); + }); +}); diff --git a/src/gateway/talk/relay/index.test.ts b/src/gateway/talk/relay/index.test.ts index 09890b1c475c0..86c090701eaa0 100644 --- a/src/gateway/talk/relay/index.test.ts +++ b/src/gateway/talk/relay/index.test.ts @@ -74,6 +74,7 @@ import { drainingRelaySessions, relaySessions } from "./state.js"; import { MAX_RELAY_TOOL_CALL_IDENTITIES } from "./tool-call-ledger.js"; const activeRelaySessions = new Map(); +const relayStub = {} as never; const tempDirs = useAutoCleanupTempDirTracker(afterEach); const RELAY_AUTH_ERROR = "Realtime provider authentication failed. Check the provider credentials and try again."; @@ -212,7 +213,7 @@ describe("talk realtime gateway relay", () => { let current = true; const runAgentConsult = bindTalkRealtimeRelayAgentConsult( runPrompt as never, - () => current, + () => (current ? relayStub : undefined), async () => {}, ); ( @@ -246,7 +247,7 @@ describe("talk realtime gateway relay", () => { ); const runAgentConsult = bindTalkRealtimeRelayAgentConsult( runPrompt as never, - () => true, + () => relayStub, async () => {}, ); @@ -4294,38 +4295,6 @@ describe("talk realtime gateway relay", () => { }, ); - it("closes an exact-response relay when cancellation is never confirmed", async () => { - vi.useFakeTimers(); - let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; - const close = vi.fn(); - const provider = createIdleRelayProvider(); - provider.createBridge = (request) => { - bridgeRequest = request; - return makeRelayTransport({ close }); - }; - const { session } = createAbortableRelayRunFixture(provider); - bridgeRequest?.onEvent?.({ - direction: "server", - type: "response.created", - responseId: "response-1", - }); - - let cancellationSettled = false; - const cancellation = cancelTalkRealtimeRelayTurn({ - relaySessionId: session.relaySessionId, - connId: "conn-1", - turnId: ensureActiveRelayTurnId(session.relaySessionId), - }); - void cancellation.then(() => (cancellationSettled = true)); - await vi.advanceTimersByTimeAsync(999); - expect(relaySessions.has(session.relaySessionId)).toBe(true); - expect(cancellationSettled).toBe(false); - await vi.advanceTimersByTimeAsync(1); - await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); - expect(relaySessions.has(session.relaySessionId)).toBe(false); - expect(close).toHaveBeenCalledOnce(); - }); - it("fails visibly when a replacement response starts before cancellation confirms", async () => { let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; const close = vi.fn(); @@ -4366,46 +4335,6 @@ describe("talk realtime gateway relay", () => { ).toHaveLength(1); }); - it("closes a stalled turn-bound cancellation after its drain deadline", async () => { - vi.useFakeTimers(); - const pending = createDeferred(); - const close = vi.fn(); - const provider = createIdleRelayProvider(); - provider.createBridge = () => - makeRelayTransport({ close, submitToolResult: vi.fn(() => pending.promise) }); - const { session } = createAbortableRelayRunFixture(provider); - - let cancellationSettled = false; - const cancellation = cancelTalkRealtimeRelayTurn({ - relaySessionId: session.relaySessionId, - connId: "conn-1", - turnId: ensureActiveRelayTurnId(session.relaySessionId), - }); - void cancellation.then(() => (cancellationSettled = true)); - const pendingAudio = Promise.resolve( - sendTalkRealtimeRelayAudio({ - relaySessionId: session.relaySessionId, - connId: "conn-1", - audioBase64: "AQI=", - }), - ); - let audioSettled = false; - void pendingAudio.then( - () => (audioSettled = true), - () => (audioSettled = true), - ); - await vi.advanceTimersByTimeAsync(999); - expect(relaySessions.has(session.relaySessionId)).toBe(true); - expect(cancellationSettled).toBe(false); - expect(audioSettled).toBe(false); - await vi.advanceTimersByTimeAsync(1); - await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); - await expect(pendingAudio).rejects.toThrow("Unknown realtime relay session"); - expect(relaySessions.has(session.relaySessionId)).toBe(false); - expect(close).toHaveBeenCalledOnce(); - pending.resolve(); - }); - it("provider close owns turn-bound drain teardown and late drain cannot touch a successor", async () => { vi.useFakeTimers(); const pending = createDeferred(); diff --git a/src/gateway/talk/relay/operations.ts b/src/gateway/talk/relay/operations.ts index fca2f4ca91b1a..3b5b0f32bd4eb 100644 --- a/src/gateway/talk/relay/operations.ts +++ b/src/gateway/talk/relay/operations.ts @@ -27,6 +27,7 @@ import { registerTalkVoiceSession, unregisterTalkVoiceSession, } from "../voice-selection.js"; +import { scheduleRelayCancellationDeadline } from "./cancellation-deadline.js"; import { submitForcedTalkRealtimeRelayToolResult, submitRelayAgentControlProviderResults, @@ -52,9 +53,11 @@ import { resolveRelayProviderToolCallId, type RelaySession, } from "./state.js"; -import { closeRelayVoiceSession, ensureRelayVoiceSession } from "./voice.js"; - -const TURN_BOUND_CANCELLATION_DRAIN_MS = 1_000; +import { + closeRelayVoiceSession, + ensureRelayVoiceSession, + releaseRelayAssistantTranscriptHold, +} from "./voice.js"; export function adoptTalkRealtimeRelaySession( session: RelaySession, @@ -138,6 +141,7 @@ export function closeRelaySession( } const closing: NonNullable = { reason }; session.closing = closing; + releaseRelayAssistantTranscriptHold(session); const disposition = options?.disposition ?? (isTalkVoiceSessionReplacing(session.id, session.connId, session.sessionTarget.agentId) @@ -655,15 +659,7 @@ export async function cancelTalkRealtimeRelayTurn(params: { const cancellationDrained = (session.outputOwnership.drain = createDeferredCore()); retireRelayAgentRuns(session, reason); cancelRelayTurn(session, turnId, reason); - setTimeout(() => { - if ( - relaySessions.get(session.id) === session && - session.toolResultEpoch === terminalEpoch && - session.outputOwnership.phase === "cancelling" - ) { - void closeRelaySession(session, "completed"); - } - }, TURN_BOUND_CANCELLATION_DRAIN_MS).unref?.(); + scheduleRelayCancellationDeadline(session, { turnId, reason, terminalEpoch }); void Promise.allSettled( [...rootCallIds].map(async (callId) => { await submitTalkRealtimeRelayToolResult({ diff --git a/src/gateway/talk/relay/session-create.ts b/src/gateway/talk/relay/session-create.ts index f3b1275071470..36c50b1ef2811 100644 --- a/src/gateway/talk/relay/session-create.ts +++ b/src/gateway/talk/relay/session-create.ts @@ -174,7 +174,7 @@ export function createTalkRealtimeRelaySession( }); const runAgentConsult = bindTalkRealtimeRelayAgentConsult( consultRunner.runPrompt, - () => getActiveRelay() !== undefined, + getActiveRelay, (signal) => confirmationReadiness.wait(signal), ); const runControl = createTalkRealtimeRunControlOwner({ @@ -241,7 +241,7 @@ export function createTalkRealtimeRelaySession( audioSink: { isOpen: () => Boolean(getActiveRelay()), sendAudio: (audio) => { - if (!getActiveRelay() || outputOwnership.phase === "cancelling") { + if (!getActiveRelay() || outputOwnership.suppressingOutput) { return; } const outputTurnId = outputOwnership.resolve(true); @@ -419,7 +419,7 @@ export function createTalkRealtimeRelaySession( if (!relay || relay.voiceSessionClose) { return; } - if (!relay.closing && role === "assistant" && outputOwnership.phase === "cancelling") { + if (!relay.closing && role === "assistant" && outputOwnership.suppressingOutput) { return; } if (!relay.closing && role === "user" && !final) { @@ -470,7 +470,7 @@ export function createTalkRealtimeRelaySession( }, onToolCall: (toolCall) => { const relay = getActiveRelay(); - if (!relay || outputOwnership.phase === "cancelling") { + if (!relay || outputOwnership.suppressingOutput) { return; } const outputTurnId = outputOwnership.resolve(true); diff --git a/src/gateway/talk/relay/state.ts b/src/gateway/talk/relay/state.ts index bd4d37f023cb9..740adc815a37e 100644 --- a/src/gateway/talk/relay/state.ts +++ b/src/gateway/talk/relay/state.ts @@ -98,6 +98,12 @@ export class TalkRealtimeRelayOutputOwnership { turnId?: string; responseId?: string; drain?: { promise: Promise; resolve: () => void }; + /** + * Set when the provider never confirmed a cancelled response: the relay released the + * drain itself and drops the interrupted generation's remaining output until the + * provider reports that response done (or a replacement response starts). + */ + discarding = false; constructor( private readonly activeTurnId: () => string | undefined, @@ -105,8 +111,17 @@ export class TalkRealtimeRelayOutputOwnership { private readonly fail: (message: string) => void, ) {} + /** Output is refused while a cancellation drains or a stale generation is discarded. */ + get suppressingOutput(): boolean { + return this.phase === "cancelling" || this.discarding; + } + responseCreated(responseId: string | undefined): boolean { const normalizedResponseId = responseId?.trim(); + if (this.discarding && normalizedResponseId && normalizedResponseId !== this.responseId) { + // A replacement response supersedes the stale generation being discarded. + this.discarding = false; + } if (this.phase === "unowned") { Object.assign(this, { mode: normalizedResponseId ? ("exact-response" as const) : ("turn-bound" as const), @@ -129,6 +144,9 @@ export class TalkRealtimeRelayOutputOwnership { } resolve(claim: boolean): string | undefined { + if (this.discarding) { + return undefined; + } const activeTurnId = this.activeTurnId(); if ( this.phase !== "cancelling" && @@ -148,6 +166,11 @@ export class TalkRealtimeRelayOutputOwnership { } finish(responseId: string | undefined, cancellationEvent = false) { + if (this.discarding) { + // The stale generation finally ended; nothing was owned in the meantime. + this.discarding = false; + return "cancelled"; + } const cancelled = this.phase === "cancelling"; if ( (cancellationEvent && !cancelled) || @@ -161,6 +184,20 @@ export class TalkRealtimeRelayOutputOwnership { return cancelled ? "cancelled" : "completed"; } + /** + * Completes a pending cancellation without provider confirmation. Returns false when no + * cancellation is pending. + */ + completeCancellationLocally(): boolean { + if (this.phase !== "cancelling") { + return false; + } + this.discarding = true; + this.drain?.resolve(); + Object.assign(this, { phase: "unowned" as const, turnId: undefined, responseId: undefined }); + return true; + } + bind(provider: RelayProvider, runAgentConsult: RealtimeVoiceAgentConsultRunner): RelayProvider { return { ...provider, @@ -218,6 +255,8 @@ export type RelaySession = { voiceSessionCreated: boolean; voiceTranscriptSeq: number; voiceTranscriptQueue: BoundedSerialQueue; + /** Assistant finals held back from the session while an agent consult is admitted. */ + assistantTranscriptHold?: { depth: number; held: string[] }; confirmationReadiness: ReturnType; voiceSessionClose?: Promise; closing?: { reason: "completed" | "error"; completion?: Promise }; diff --git a/src/gateway/talk/relay/voice.test.ts b/src/gateway/talk/relay/voice.test.ts index 18ff2ee5db30b..dd4bcb820dd9a 100644 --- a/src/gateway/talk/relay/voice.test.ts +++ b/src/gateway/talk/relay/voice.test.ts @@ -2,7 +2,11 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { createClientVoiceConfirmationReadiness } from "../../../talk/client-voice-confirmation-readiness.js"; import { VOICE_TRANSCRIPT_QUEUE_POLICY } from "../../../talk/voice-transcript.js"; import type { RelaySession } from "./state.js"; -import { closeRelayVoiceSession, enqueueRelayVoiceTranscript } from "./voice.js"; +import { + beginRelayAssistantTranscriptHold, + closeRelayVoiceSession, + enqueueRelayVoiceTranscript, +} from "./voice.js"; const voiceSessionMocks = vi.hoisted(() => ({ appendRelayVoiceTranscript: vi.fn(), @@ -60,6 +64,39 @@ describe("realtime relay voice transcript persistence", () => { voiceSessionMocks.createOrResumeClientVoiceSession.mockReset(); }); + it("holds assistant finals while a consult is admitted and appends them in order afterwards", async () => { + voiceSessionMocks.appendRelayVoiceTranscript.mockResolvedValue(undefined); + const { session } = createRelaySession(); + const appended = () => + voiceSessionMocks.appendRelayVoiceTranscript.mock.calls.map(([params]) => [ + (params as { role: string }).role, + (params as { text: string }).text, + ]); + const release = beginRelayAssistantTranscriptHold(session); + const nested = beginRelayAssistantTranscriptHold(session); + + expect(enqueueRelayVoiceTranscript(session, "assistant", "Let me check.")).toBe(true); + expect(enqueueRelayVoiceTranscript(session, "user", "thanks")).toBe(true); + expect(enqueueRelayVoiceTranscript(session, "assistant", "One moment.")).toBe(true); + await session.voiceTranscriptQueue.flush(); + // User finals keep flowing; only assistant finals wait. + expect(appended()).toEqual([["user", "thanks"]]); + + nested(); + await session.voiceTranscriptQueue.flush(); + expect(appended()).toEqual([["user", "thanks"]]); + + release(); + release(); + await session.voiceTranscriptQueue.flush(); + expect(appended()).toEqual([ + ["user", "thanks"], + ["assistant", "Let me check."], + ["assistant", "One moment."], + ]); + expect(session.assistantTranscriptHold).toBeUndefined(); + }); + it("bounds stalled finals, drains the accepted prefix, and closes once", async () => { const firstAppend = deferred(); voiceSessionMocks.appendRelayVoiceTranscript.mockImplementation( diff --git a/src/gateway/talk/relay/voice.ts b/src/gateway/talk/relay/voice.ts index 978ed95d70e3d..f11a581a5d829 100644 --- a/src/gateway/talk/relay/voice.ts +++ b/src/gateway/talk/relay/voice.ts @@ -50,6 +50,10 @@ export function enqueueRelayVoiceTranscript( if (!normalizedText) { return true; } + if (role === "assistant" && session.assistantTranscriptHold && !session.closing) { + session.assistantTranscriptHold.held.push(normalizedText); + return true; + } if (!ensureRelayVoiceSession(session)) { session.confirmationReadiness.fail(new Error("Realtime voice session could not be recorded")); return true; @@ -104,6 +108,46 @@ export function enqueueRelayVoiceTranscript( return true; } +/** + * Holds assistant transcript session appends while an agent consult is admitted. A spoken + * filler persisted between the consult's keyed user turn and its adoption moves the + * session's current-turn anchor and fails the run ("keyed user is outside the current + * turn"). Clients, echo tracking, and spoken run control still see transcripts at once; + * only the durable append waits. Holds nest; the last release appends the held finals in + * their original order. + */ +export function beginRelayAssistantTranscriptHold(session: RelaySession | undefined): () => void { + if (!session) { + return () => {}; + } + const hold = (session.assistantTranscriptHold ??= { depth: 0, held: [] }); + hold.depth += 1; + let released = false; + return () => { + if (released) { + return; + } + released = true; + hold.depth -= 1; + if (hold.depth > 0 || session.assistantTranscriptHold !== hold) { + return; + } + releaseRelayAssistantTranscriptHold(session); + }; +} + +/** Appends every held assistant final now, in order, and clears the hold. */ +export function releaseRelayAssistantTranscriptHold(session: RelaySession): void { + const hold = session.assistantTranscriptHold; + if (!hold) { + return; + } + session.assistantTranscriptHold = undefined; + for (const text of hold.held) { + enqueueRelayVoiceTranscript(session, "assistant", text); + } +} + export function closeRelayVoiceSession(session: RelaySession): Promise { if (session.voiceSessionClose) { return session.voiceSessionClose;