diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts index 8f791f39608b..bbbc20356de7 100644 --- a/src/gateway/talk-realtime-relay-operations.ts +++ b/src/gateway/talk-realtime-relay-operations.ts @@ -100,7 +100,7 @@ export function closeRelaySession(session: RelaySession, reason: "completed" | " } finally { // Provider teardown may throw, but the relay must still reach its durable // voice and owner-visible terminal state before that error is surfaced. - closeRelayVoiceSession(session); + void closeRelayVoiceSession(session); broadcastToOwner(session.context, session.connId, { relaySessionId: session.id, type: "close", @@ -351,7 +351,7 @@ export async function flushTalkRealtimeRelayVoiceWrites(params: { connId: string; }): Promise { const session = getRelaySession(params.relaySessionId, params.connId); - await session.voiceTranscriptWrites; + await session.voiceTranscriptQueue.flush(); } /** Applies realtime voice-control text to the active agent-consult chat run. */ diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index fc75eefc87b5..110ed9cb3891 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -17,6 +17,7 @@ import { } from "../talk/provider-types.js"; import { createRealtimeVoiceSessionHarness } from "../talk/realtime-session-harness.js"; import type { TalkEventInput } from "../talk/talk-session-controller.js"; +import { VOICE_TRANSCRIPT_QUEUE_POLICY } from "../talk/voice-transcript.js"; import { registerChatAbortController } from "./chat-abort.js"; import { buildAlreadyDeliveredToolResult, @@ -105,6 +106,7 @@ export function createTalkRealtimeRelaySession( let currentOutputResponseId: string | undefined; let ready = false; let failureEmitted = false; + let transcriptPersistenceFailed = false; const constructionTerminal: { current?: { kind: "error"; error: Error } | { kind: "close"; reason: RealtimeVoiceCloseReason }; } = {}; @@ -292,10 +294,10 @@ export function createTalkRealtimeRelaySession( if (!relay) { return; } - const turnId = ensureRelayTurn(relay); - if (final) { - enqueueRelayVoiceTranscript(relay, role, text); + if (final && !enqueueRelayVoiceTranscript(relay, role, text)) { + return; } + const turnId = ensureRelayTurn(relay); const eventType = role === "assistant" ? final @@ -453,7 +455,7 @@ export function createTalkRealtimeRelaySession( forgetUnifiedTalkSession(relaySessionId); clearTimeout(active.cleanupTimer); abortRelayAgentRuns(active, "relay-closed"); - closeRelayVoiceSession(active); + void closeRelayVoiceSession(active); if (!ready && !failureEmitted) { const issue = realtimeRelayIssue({ message: "Realtime provider closed before the session became ready.", @@ -489,6 +491,25 @@ export function createTalkRealtimeRelaySession( throw new Error(`Realtime provider closed during session creation: ${earlyTerminal.reason}`); } const initialSessionKey = params.sessionKey?.trim() || undefined; + const failVoiceTranscriptPersistence = (message: string) => { + const active = relaySessions.get(relaySessionId); + if (!active || transcriptPersistenceFailed) { + return; + } + transcriptPersistenceFailed = true; + if (!failureEmitted) { + failureEmitted = true; + emit( + { relaySessionId, type: "error", message }, + { + type: "session.error", + payload: { message }, + final: true, + }, + ); + } + closeRelaySession(active, "error"); + }; const relay: RelaySession = { id: relaySessionId, connId: params.connId, @@ -525,7 +546,8 @@ export function createTalkRealtimeRelaySession( ...(params.cfg ? { voiceConfig: params.cfg } : {}), voiceSessionCreated: false, voiceTranscriptSeq: 0, - voiceTranscriptWrites: Promise.resolve(), + voiceTranscriptQueue: VOICE_TRANSCRIPT_QUEUE_POLICY.createQueue(), + failVoiceTranscriptPersistence, pendingVoiceTranscripts: [], }; relayRef.current = relay; diff --git a/src/gateway/talk-realtime-relay-state.ts b/src/gateway/talk-realtime-relay-state.ts index 0568d8b55b54..6e98cea8498a 100644 --- a/src/gateway/talk-realtime-relay-state.ts +++ b/src/gateway/talk-realtime-relay-state.ts @@ -1,5 +1,6 @@ import type { OpenClawConfig } from "../config/types.js"; import type { RealtimeVoiceProviderPlugin } from "../plugins/types.js"; +import type { BoundedSerialQueue } from "../shared/bounded-serial-queue.js"; import type { RealtimeVoiceAgentControlResult } from "../talk/agent-run-control.js"; import type { RealtimeVoiceBrowserAudioContract, @@ -112,7 +113,9 @@ export type RelaySession = { voiceConfig?: OpenClawConfig; voiceSessionCreated: boolean; voiceTranscriptSeq: number; - voiceTranscriptWrites: Promise; + voiceTranscriptQueue: BoundedSerialQueue; + voiceSessionClose?: Promise; + failVoiceTranscriptPersistence: (message: string) => void; pendingVoiceTranscripts: Array<{ role: "user" | "assistant"; text: string }>; }; diff --git a/src/gateway/talk-realtime-relay-voice.test.ts b/src/gateway/talk-realtime-relay-voice.test.ts new file mode 100644 index 000000000000..f3c55dcc041b --- /dev/null +++ b/src/gateway/talk-realtime-relay-voice.test.ts @@ -0,0 +1,121 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { VOICE_TRANSCRIPT_QUEUE_POLICY } from "../talk/voice-transcript.js"; +import type { RelaySession } from "./talk-realtime-relay-state.js"; +import { + closeRelayVoiceSession, + enqueueRelayVoiceTranscript, +} from "./talk-realtime-relay-voice.js"; + +const voiceSessionMocks = vi.hoisted(() => ({ + appendRelayVoiceTranscript: vi.fn(), + closeClientVoiceSession: vi.fn(), + createOrResumeClientVoiceSession: vi.fn(), +})); + +vi.mock("../talk/client-voice-session.js", () => voiceSessionMocks); + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((accept) => { + resolve = accept; + }); + return { promise, resolve }; +} + +function createRelaySession(): { + session: RelaySession; + failVoiceTranscriptPersistence: ReturnType; +} { + const failVoiceTranscriptPersistence = vi.fn(() => { + void closeRelayVoiceSession(session); + }); + const session = { + id: "relay-voice-bounded", + sessionKey: "agent:main:main", + agentId: "main", + provider: "openai", + context: { + getRuntimeConfig: () => ({}), + logGateway: { warn: vi.fn() }, + }, + voiceSessionCreated: false, + voiceTranscriptSeq: 0, + voiceTranscriptQueue: VOICE_TRANSCRIPT_QUEUE_POLICY.createQueue(), + failVoiceTranscriptPersistence, + pendingVoiceTranscripts: [], + } as unknown as RelaySession; + return { session, failVoiceTranscriptPersistence }; +} + +describe("realtime relay voice transcript persistence", () => { + beforeEach(() => { + voiceSessionMocks.appendRelayVoiceTranscript.mockReset(); + voiceSessionMocks.closeClientVoiceSession.mockReset().mockResolvedValue(undefined); + voiceSessionMocks.createOrResumeClientVoiceSession.mockReset(); + }); + + it("bounds stalled finals, drains the accepted prefix, and closes once", async () => { + const firstAppend = deferred(); + voiceSessionMocks.appendRelayVoiceTranscript.mockImplementation( + async ({ entryId }: { entryId: string }) => { + if (entryId === "1") { + await firstAppend.promise; + } + }, + ); + const { session, failVoiceTranscriptPersistence } = createRelaySession(); + let accepted = 0; + + for (let index = 0; index < 10_000; index += 1) { + if ( + enqueueRelayVoiceTranscript( + session, + index % 2 === 0 ? "user" : "assistant", + ` ${"x".repeat(9_000)} `, + ) + ) { + accepted += 1; + } + } + + expect(accepted).toBe(41); + expect(voiceSessionMocks.appendRelayVoiceTranscript).toHaveBeenCalledOnce(); + expect(failVoiceTranscriptPersistence).toHaveBeenCalledOnce(); + const close = session.voiceSessionClose; + expect(close).toBeDefined(); + expect(closeRelayVoiceSession(session)).toBe(close); + expect(voiceSessionMocks.closeClientVoiceSession).not.toHaveBeenCalled(); + + firstAppend.resolve(); + await close; + + expect(voiceSessionMocks.appendRelayVoiceTranscript).toHaveBeenCalledTimes(41); + expect( + voiceSessionMocks.appendRelayVoiceTranscript.mock.calls.map( + ([params]) => (params as { entryId: string }).entryId, + ), + ).toEqual(Array.from({ length: 41 }, (_, index) => String(index + 1))); + expect( + voiceSessionMocks.appendRelayVoiceTranscript.mock.calls.every( + ([params]) => (params as { text: string }).text.length === 8_000, + ), + ).toBe(true); + expect(voiceSessionMocks.closeClientVoiceSession).toHaveBeenCalledOnce(); + expect(enqueueRelayVoiceTranscript(session, "user", "too late")).toBe(false); + }); + + it("normalizes the bounded pre-bind transcript buffer", () => { + const { session } = createRelaySession(); + session.sessionKey = undefined; + session.agentId = undefined; + + for (let index = 0; index < 100; index += 1) { + expect(enqueueRelayVoiceTranscript(session, "user", ` ${"x".repeat(9_000)} `)).toBe(true); + } + + expect(session.pendingVoiceTranscripts).toHaveLength(40); + expect(session.pendingVoiceTranscripts.every((entry) => entry.text.length === 8_000)).toBe( + true, + ); + }); +}); diff --git a/src/gateway/talk-realtime-relay-voice.ts b/src/gateway/talk-realtime-relay-voice.ts index d7e12ffd0c6b..444854265824 100644 --- a/src/gateway/talk-realtime-relay-voice.ts +++ b/src/gateway/talk-realtime-relay-voice.ts @@ -5,6 +5,10 @@ import { closeClientVoiceSession, createOrResumeClientVoiceSession, } from "../talk/client-voice-session.js"; +import { + normalizeVoiceTranscriptText, + VOICE_TRANSCRIPT_QUEUE_POLICY, +} from "../talk/voice-transcript.js"; import type { RelaySession } from "./talk-realtime-relay-state.js"; const RELAY_TRANSCRIPT_RETRY_DELAYS_MS = [0, 500, 2_000] as const; @@ -63,31 +67,30 @@ export function ensureRelayVoiceSession(session: RelaySession): boolean { } } -const MAX_PENDING_VOICE_TRANSCRIPTS = 40; - export function enqueueRelayVoiceTranscript( session: RelaySession, role: "user" | "assistant", text: string, -): void { +): boolean { + const normalizedText = normalizeVoiceTranscriptText(text); if (!session.sessionKey) { // Lazy-bound relays hear audio before talk.client.toolCall supplies the session // key; buffer bounded finals so the call's opening turns survive the binding. // Never-binding callers accept best-effort loss: they had no persistence before. - session.pendingVoiceTranscripts.push({ role, text }); - if (session.pendingVoiceTranscripts.length > MAX_PENDING_VOICE_TRANSCRIPTS) { + session.pendingVoiceTranscripts.push({ role, text: normalizedText }); + if (session.pendingVoiceTranscripts.length > VOICE_TRANSCRIPT_QUEUE_POLICY.maxPendingCount) { session.pendingVoiceTranscripts.shift(); } - return; + return true; } if (!ensureRelayVoiceSession(session)) { - return; + return true; } - session.voiceTranscriptSeq += 1; - const entryId = String(session.voiceTranscriptSeq); + const transcriptSeq = session.voiceTranscriptSeq + 1; + const entryId = String(transcriptSeq); const sessionKey = session.sessionKey; - session.voiceTranscriptWrites = session.voiceTranscriptWrites - .then(async () => { + const admission = session.voiceTranscriptQueue.enqueue( + async () => { let lastError: unknown; for (const delayMs of RELAY_TRANSCRIPT_RETRY_DELAYS_MS) { if (delayMs > 0) { @@ -102,7 +105,7 @@ export function enqueueRelayVoiceTranscript( voiceSessionId: session.id, entryId, role, - text, + text: normalizedText, ...(session.voiceConfig ? { config: session.voiceConfig } : {}), }); return; @@ -111,18 +114,34 @@ export function enqueueRelayVoiceTranscript( } } throw lastError; - }) - .catch((error: unknown) => { - logRelayVoiceFailure(session, "realtime relay transcript append failed", error); - }); + }, + { weight: normalizedText.length }, + ); + if (!admission.accepted) { + if (admission.reason === "overflow") { + session.failVoiceTranscriptPersistence(VOICE_TRANSCRIPT_QUEUE_POLICY.overflowMessage); + } + return false; + } + session.voiceTranscriptSeq = transcriptSeq; + void admission.completion.catch((error: unknown) => { + logRelayVoiceFailure(session, "realtime relay transcript append failed", error); + }); + return true; } -export function closeRelayVoiceSession(session: RelaySession): void { +export function closeRelayVoiceSession(session: RelaySession): Promise { + if (session.voiceSessionClose) { + return session.voiceSessionClose; + } + session.voiceTranscriptQueue.seal(); if (!session.sessionKey || !ensureRelayVoiceSession(session)) { - return; + session.voiceSessionClose = Promise.resolve(); + return session.voiceSessionClose; } const sessionKey = session.sessionKey; - void session.voiceTranscriptWrites + session.voiceSessionClose = session.voiceTranscriptQueue + .flush() .then(async () => { const config = session.voiceConfig ?? session.context.getRuntimeConfig(); await closeClientVoiceSession({ @@ -135,4 +154,5 @@ export function closeRelayVoiceSession(session: RelaySession): void { .catch((error: unknown) => { logRelayVoiceFailure(session, "realtime relay voice session close failed", error); }); + return session.voiceSessionClose; } diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index d715b63c274d..653031a04a52 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -23,6 +23,7 @@ import type { } from "../talk/provider-types.js"; import { captureEnv, setTestEnvValue } from "../test-utils/env.js"; import { createChatRunState } from "./server-chat-state.js"; +import { relaySessions } from "./talk-realtime-relay-state.js"; import { acknowledgeTalkRealtimeRelayMark, cancelTalkRealtimeRelayTurn, @@ -421,6 +422,102 @@ describe("talk realtime gateway relay", () => { } }); + it("emits one terminal error and close when transcript persistence overflows", async () => { + const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]); + const tempDir = await fs.realpath(tempDirs.make("openclaw-relay-voice-overflow-")); + setTestEnvValue("OPENCLAW_STATE_DIR", tempDir); + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const bridgeClose = vi.fn(); + const events: Array<{ event: string; payload: unknown; connIds: string[] }> = []; + let releaseQueue!: () => void; + const queueBlocked = new Promise((resolve) => { + releaseQueue = resolve; + }); + try { + await replaceSessionEntry( + { agentId: "main", sessionKey: "agent:main:main" }, + { sessionId: "relay-voice-overflow-session", updatedAt: Date.now() }, + ); + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + bridgeRequest = request; + return { + ...createIdleRelayProvider().createBridge(request), + close: bridgeClose, + } as RealtimeVoiceBridge; + }; + const session = createTalkRealtimeRelaySession({ + context: { + broadcastToConnIds: (event: string, payload: unknown, connIds: ReadonlySet) => { + events.push({ event, payload, connIds: [...connIds] }); + }, + chatAbortControllers: new Map(), + getRuntimeConfig: () => ({}), + logGateway: { warn: vi.fn() }, + } as never, + connId: "conn-voice-overflow", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + sessionKey: "agent:main:main", + }); + const relay = relaySessions.get(session.relaySessionId); + if (!relay) { + throw new Error("expected active relay"); + } + relay.voiceTranscriptQueue.enqueue(async () => await queueBlocked); + + for (let index = 0; index < 10_000; index += 1) { + bridgeRequest?.onTranscript?.("user", `message ${index}`, true); + } + + const errorPayloads = events + .map((entry) => entry.payload) + .filter( + (payload): payload is Record => + typeof payload === "object" && + payload !== null && + (payload as Record).type === "error", + ); + const closePayloads = events + .map((entry) => entry.payload) + .filter( + (payload): payload is Record => + typeof payload === "object" && + payload !== null && + (payload as Record).type === "close", + ); + expect(errorPayloads).toHaveLength(1); + expect(errorPayloads[0]).toMatchObject({ + relaySessionId: session.relaySessionId, + type: "error", + message: expect.stringContaining("persistence could not keep up"), + }); + expect(errorPayloads[0]).not.toHaveProperty("code"); + expect(closePayloads).toEqual([ + expect.objectContaining({ + relaySessionId: session.relaySessionId, + type: "close", + reason: "error", + }), + ]); + expect(bridgeClose).toHaveBeenCalledOnce(); + expect(relaySessions.has(session.relaySessionId)).toBe(false); + + releaseQueue(); + await relay.voiceSessionClose; + expect(clientVoiceSessionTesting.readRecord("main", session.relaySessionId)?.status).toBe( + "closed", + ); + } finally { + releaseQueue(); + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + envSnapshot.restore(); + } + }); + it("creates the relay voice record before binding a transcript-free consult", async () => { const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]); const tempDir = await fs.realpath(