fix(talk): require successful transcript close barriers

src/gateway/talk-realtime-relay-voice.test.ts

@@ -108,6 +108,29 @@ describe("realtime relay voice transcript persistence", () => {
     expect(enqueueRelayVoiceTranscript(session, "user", "too late")).toBe(false);
   });
 
+  it("does not close the durable record after an accepted transcript fails", async () => {
+    vi.useFakeTimers();
+    try {
+      voiceSessionMocks.appendRelayVoiceTranscript.mockRejectedValue(
+        new Error("transcript write failed"),
+      );
+      const { session } = createRelaySession();
+
+      expect(enqueueRelayVoiceTranscript(session, "user", "persist me")).toBe(true);
+      const close = closeRelayVoiceSession(session);
+      await vi.runAllTimersAsync();
+      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"),
+      );
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("normalizes the bounded pre-bind transcript buffer", () => {
     const { session } = createRelaySession();
     session.sessionKey = undefined;

src/gateway/talk-realtime-relay-voice.ts

@@ -144,7 +144,7 @@ export function closeRelayVoiceSession(session: RelaySession): Promise<void> {
   }
   const sessionKey = session.sessionKey;
   session.voiceSessionClose = session.voiceTranscriptQueue
-    .flush()
+    .flush({ requireSuccess: true })
     .then(async () => {
       const config = session.voiceConfig ?? session.context.getRuntimeConfig();
       await closeClientVoiceSession({

src/shared/bounded-serial-queue.test.ts

@@ -126,6 +126,22 @@ describe("BoundedSerialQueue", () => {
     }
   });
 
+  it("can require every task in the accepted prefix to succeed", async () => {
+    const failure = new Error("persistence failed");
+    const queue = new BoundedSerialQueue({ maxPendingCount: 1, maxPendingWeight: 1 });
+    const task = queue.enqueue(async () => {
+      throw failure;
+    });
+    const ordinaryFlush = queue.flush();
+    const strictFlush = queue.flush({ requireSuccess: true });
+
+    await expect(ordinaryFlush).resolves.toBeUndefined();
+    await expect(strictFlush).rejects.toBe(failure);
+    if (task.accepted) {
+      await expect(task.completion).rejects.toBe(failure);
+    }
+  });
+
   it("seals idempotently while preserving accepted work", async () => {
     const first = deferred();
     const queue = new BoundedSerialQueue({ maxPendingCount: 1, maxPendingWeight: 1 });

src/shared/bounded-serial-queue.ts

@@ -21,6 +21,8 @@ export class BoundedSerialQueue {
   private active = false;
   private sealed = false;
   private overflowed = false;
+  private failed = false;
+  private firstFailure: unknown;
   private settledPrefix: Promise<void> = Promise.resolve();
 
   constructor(
@@ -104,9 +106,18 @@ export class BoundedSerialQueue {
    *
    * Later admissions do not extend this barrier, which keeps consult flushes
    * finite while close can seal first to drain the entire accepted prefix.
+   * Close owners can require that prefix to have completed without failures.
    */
-  flush(): Promise<void> {
-    return this.settledPrefix;
+  flush(options: { requireSuccess?: boolean } = {}): Promise<void> {
+    const prefix = this.settledPrefix;
+    if (options.requireSuccess !== true) {
+      return prefix;
+    }
+    return prefix.then(() => {
+      if (this.failed) {
+        throw this.firstFailure;
+      }
+    });
   }
 
   private startTask(task: BoundedSerialQueueTask): void {
@@ -117,6 +128,10 @@ export class BoundedSerialQueue {
     try {
       task.resolve(await task.run());
     } catch (error) {
+      if (!this.failed) {
+        this.failed = true;
+        this.firstFailure = error;
+      }
       task.reject(error);
     } finally {
       const next = this.pending.shift();

src/talk/client-voice-session.test.ts

@@ -403,6 +403,75 @@ describe("client voice session", () => {
     );
   });
 
+  it("keeps the session open when an accepted transcript fails during close", async () => {
+    await seedSession("agent:main:main");
+    const voiceSessionId = createOrResumeClientVoiceSession({
+      agentId: "main",
+      sessionKey: "agent:main:main",
+      origin: "client",
+      voiceSessionId: "voice-close-after-failure",
+    });
+    const transcriptWrite = createDeferred();
+    const failure = new Error("transcript write failed");
+    const actualAppend = sessionAccessorMocks.actualAppendTranscriptMessage!;
+    sessionAccessorMocks.appendTranscriptMessage.mockImplementationOnce(async () => {
+      await transcriptWrite.promise;
+      throw failure;
+    });
+    const append = appendClientVoiceTranscript({
+      agentId: "main",
+      sessionKey: "agent:main:main",
+      voiceSessionId,
+      entryId: "retryable",
+      role: "user",
+      text: "persist me",
+    });
+    const appendResult = append.then(
+      () => undefined,
+      (error: unknown) => error,
+    );
+    await vi.waitFor(() =>
+      expect(sessionAccessorMocks.appendTranscriptMessage).toHaveBeenCalledOnce(),
+    );
+    const close = closeClientVoiceSession({
+      agentId: "main",
+      sessionKey: "agent:main:main",
+      voiceSessionId,
+      config: {},
+      now: 42,
+    });
+    const closeResult = close.then(
+      () => undefined,
+      (error: unknown) => error,
+    );
+
+    transcriptWrite.resolve();
+    expect(await appendResult).toBe(failure);
+    expect(await closeResult).toBe(failure);
+    expect(clientVoiceSessionTesting.readRecord("main", voiceSessionId)?.status).toBe("open");
+
+    sessionAccessorMocks.appendTranscriptMessage.mockImplementation(actualAppend);
+    await appendClientVoiceTranscript({
+      agentId: "main",
+      sessionKey: "agent:main:main",
+      voiceSessionId,
+      entryId: "retryable",
+      role: "user",
+      text: "persist me",
+    });
+    await closeClientVoiceSession({
+      agentId: "main",
+      sessionKey: "agent:main:main",
+      voiceSessionId,
+      config: {},
+      now: 99,
+    });
+    expect(clientVoiceSessionTesting.readRecord("main", voiceSessionId)).toMatchObject({
+      status: "closed",
+      closedAt: 99,
+    });
+  });
+
   it("bounds stalled transcript operations and closes after the accepted prefix", async () => {
     await seedSession("agent:main:main");
     const voiceSessionId = createOrResumeClientVoiceSession({

src/talk/voice-transcript.ts

@@ -106,7 +106,7 @@ class VoiceTranscriptOperationRegistry {
     if (!owner.closePromise) {
       // Seal synchronously so no transcript can enter behind the close barrier.
       owner.queue.seal();
-      owner.closePromise = owner.queue.flush().then(operation);
+      owner.closePromise = owner.queue.flush({ requireSuccess: true }).then(operation);
     }
     try {
       await owner.closePromise;