diff --git a/src/gateway/talk-realtime-relay-voice.test.ts b/src/gateway/talk-realtime-relay-voice.test.ts index 1fbbb107a29b..4da5a072c839 100644 --- a/src/gateway/talk-realtime-relay-voice.test.ts +++ b/src/gateway/talk-realtime-relay-voice.test.ts @@ -8,7 +8,7 @@ import { const voiceSessionMocks = vi.hoisted(() => ({ appendRelayVoiceTranscript: vi.fn(), - closeClientVoiceSession: vi.fn(), + closeRelayVoiceSessionRecord: vi.fn(), createOrResumeClientVoiceSession: vi.fn(), })); @@ -50,7 +50,7 @@ function createRelaySession(): { describe("realtime relay voice transcript persistence", () => { beforeEach(() => { voiceSessionMocks.appendRelayVoiceTranscript.mockReset(); - voiceSessionMocks.closeClientVoiceSession.mockReset().mockResolvedValue(undefined); + voiceSessionMocks.closeRelayVoiceSessionRecord.mockReset().mockResolvedValue(undefined); voiceSessionMocks.createOrResumeClientVoiceSession.mockReset(); }); @@ -88,7 +88,7 @@ describe("realtime relay voice transcript persistence", () => { const close = session.voiceSessionClose; expect(close).toBeDefined(); expect(closeRelayVoiceSession(session)).toBe(close); - expect(voiceSessionMocks.closeClientVoiceSession).not.toHaveBeenCalled(); + expect(voiceSessionMocks.closeRelayVoiceSessionRecord).not.toHaveBeenCalled(); firstAppend.resolve(); await close; @@ -104,11 +104,11 @@ describe("realtime relay voice transcript persistence", () => { ([params]) => (params as { text: string }).text.length === 8_000, ), ).toBe(true); - expect(voiceSessionMocks.closeClientVoiceSession).toHaveBeenCalledOnce(); + expect(voiceSessionMocks.closeRelayVoiceSessionRecord).toHaveBeenCalledOnce(); expect(enqueueRelayVoiceTranscript(session, "user", "too late")).toBe(false); }); - it("does not close the durable record after an accepted transcript fails", async () => { + it("terminally closes the durable record after bounded transcript retries fail", async () => { vi.useFakeTimers(); try { voiceSessionMocks.appendRelayVoiceTranscript.mockRejectedValue( @@ -122,9 +122,9 @@ describe("realtime relay voice transcript persistence", () => { await close; expect(voiceSessionMocks.appendRelayVoiceTranscript).toHaveBeenCalledTimes(3); - expect(voiceSessionMocks.closeClientVoiceSession).not.toHaveBeenCalled(); - expect(session.context.logGateway?.warn).toHaveBeenCalledWith( - expect.stringContaining("realtime relay voice session close failed"), + expect(voiceSessionMocks.closeRelayVoiceSessionRecord).toHaveBeenCalledOnce(); + expect(session.context.logGateway?.warn).toHaveBeenCalledExactlyOnceWith( + expect.stringContaining("realtime relay transcript append failed"), ); } finally { vi.useRealTimers(); diff --git a/src/gateway/talk-realtime-relay-voice.ts b/src/gateway/talk-realtime-relay-voice.ts index 8e936497984a..ac571e49c0e6 100644 --- a/src/gateway/talk-realtime-relay-voice.ts +++ b/src/gateway/talk-realtime-relay-voice.ts @@ -2,7 +2,7 @@ import { formatErrorMessage } from "../infra/errors.js"; import { resolveTalkSessionAgentId } from "../talk/agent-target.js"; import { appendRelayVoiceTranscript, - closeClientVoiceSession, + closeRelayVoiceSessionRecord, createOrResumeClientVoiceSession, } from "../talk/client-voice-session.js"; import { @@ -144,10 +144,10 @@ export function closeRelayVoiceSession(session: RelaySession): Promise { } const sessionKey = session.sessionKey; session.voiceSessionClose = session.voiceTranscriptQueue - .flush({ requireSuccess: true }) + .flush() .then(async () => { const config = session.voiceConfig ?? session.context.getRuntimeConfig(); - await closeClientVoiceSession({ + await closeRelayVoiceSessionRecord({ agentId: resolveRelayAgentId(session, sessionKey), sessionKey, voiceSessionId: session.id, diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index ca6dc76e849f..dd3b867e9ea4 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -759,6 +759,7 @@ describe("talk realtime gateway relay", () => { "closed", ), ); + expect(warn).toHaveBeenCalledTimes(1); } finally { closeOpenClawAgentDatabasesForTest(); closeOpenClawStateDatabaseForTest(); diff --git a/src/talk/client-voice-session.test.ts b/src/talk/client-voice-session.test.ts index b36af95ca1cd..f0fd83fcdcc2 100644 --- a/src/talk/client-voice-session.test.ts +++ b/src/talk/client-voice-session.test.ts @@ -22,7 +22,9 @@ import { import { resetClientVoiceConfirmationStateForTest } from "./client-voice-confirmation.test-support.js"; import { appendClientVoiceTranscript, + appendRelayVoiceTranscript, closeClientVoiceSession, + closeRelayVoiceSessionRecord, closeStaleClientVoiceSessions, createOrResumeClientVoiceSession, ensureClientVoiceAgentSessionEntry, @@ -522,6 +524,45 @@ describe("client voice session", () => { ).toEqual([]); }); + it("terminally closes relay sessions while retaining unresolved transcript identity", async () => { + await seedSession("agent:main:main"); + const voiceSessionId = createOrResumeClientVoiceSession({ + agentId: "main", + sessionKey: "agent:main:main", + origin: "relay", + }); + sessionAccessorMocks.appendTranscriptMessage.mockRejectedValueOnce( + new Error("transcript write failed"), + ); + + await expect( + appendRelayVoiceTranscript({ + agentId: "main", + sessionKey: "agent:main:main", + voiceSessionId, + entryId: "relay-entry-1", + role: "user", + text: "persist me", + }), + ).rejects.toThrow("transcript write failed"); + const unresolved = clientVoiceSessionTesting.readRecord( + "main", + voiceSessionId, + )?.transcriptFailureKeys; + expect(unresolved).toEqual([expect.stringMatching(/^[0-9a-f]{64}$/)]); + + await closeRelayVoiceSessionRecord({ + agentId: "main", + sessionKey: "agent:main:main", + voiceSessionId, + config: {}, + }); + expect(clientVoiceSessionTesting.readRecord("main", voiceSessionId)).toMatchObject({ + status: "closed", + transcriptFailureKeys: unresolved, + }); + }); + it("bounds stalled transcript operations and closes after the accepted prefix", async () => { await seedSession("agent:main:main"); const voiceSessionId = createOrResumeClientVoiceSession({ diff --git a/src/talk/client-voice-session.ts b/src/talk/client-voice-session.ts index f74ca43eb05e..9d8cd2da3120 100644 --- a/src/talk/client-voice-session.ts +++ b/src/talk/client-voice-session.ts @@ -563,6 +563,7 @@ async function closeClientVoiceSessionInternal(params: { sessionKey: string; voiceSessionId: string; config: OpenClawConfig; + transcriptFailurePolicy: "require-success" | "retain-and-close"; now?: number; }): Promise { const existing = readRecord(params.agentId, params.voiceSessionId); @@ -578,9 +579,15 @@ async function closeClientVoiceSessionInternal(params: { throw new Error("voice session disappeared during close"); } assertOwnership(current, params); - if (current.transcriptFailureKeys.length > 0) { + if ( + current.transcriptFailureKeys.length > 0 && + params.transcriptFailurePolicy === "require-success" + ) { throw new Error("voice transcript persistence must be retried before close"); } + if (params.transcriptFailurePolicy === "retain-and-close" && current.origin !== "relay") { + throw new Error("only relay voice sessions may close with unresolved transcripts"); + } if (current.status === "open") { current.status = "closed"; current.closedAt = now; @@ -618,7 +625,27 @@ export async function closeClientVoiceSession(params: { config: OpenClawConfig; now?: number; }): Promise { - await closeVoiceSessionOperationOwner(params); + await closeVoiceSessionOperationOwner({ + ...params, + transcriptFailurePolicy: "require-success", + }); +} + +/** + * Terminally close a relay call after its bounded append retries settle. + * Relays have no payload replay owner after teardown, so unresolved hashes remain as audit state. + */ +export async function closeRelayVoiceSessionRecord(params: { + agentId: string; + sessionKey: string; + voiceSessionId: string; + config: OpenClawConfig; + now?: number; +}): Promise { + await closeVoiceSessionOperationOwner({ + ...params, + transcriptFailurePolicy: "retain-and-close", + }); } /** Close abandoned open calls idle for the fixed six-hour recovery window. */