fix(talk): bound relay transcript persistence

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<void> {
   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. */

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;

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<void>;
+  voiceTranscriptQueue: BoundedSerialQueue;
+  voiceSessionClose?: Promise<void>;
+  failVoiceTranscriptPersistence: (message: string) => void;
   pendingVoiceTranscripts: Array<{ role: "user" | "assistant"; text: string }>;
 };
 

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<void>; resolve: () => void } {
+  let resolve!: () => void;
+  const promise = new Promise<void>((accept) => {
+    resolve = accept;
+  });
+  return { promise, resolve };
+}
+
+function createRelaySession(): {
+  session: RelaySession;
+  failVoiceTranscriptPersistence: ReturnType<typeof vi.fn>;
+} {
+  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,
+    );
+  });
+});

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<void> {
+  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;
 }

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<void>((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<string>) => {
+            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<string, unknown> =>
+            typeof payload === "object" &&
+            payload !== null &&
+            (payload as Record<string, unknown>).type === "error",
+        );
+      const closePayloads = events
+        .map((entry) => entry.payload)
+        .filter(
+          (payload): payload is Record<string, unknown> =>
+            typeof payload === "object" &&
+            payload !== null &&
+            (payload as Record<string, unknown>).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(