refactor(talk): internalize authoritative relay events

Co-authored-by: Dallin Romney <dallinromney@gmail.com>
Punchcard-Session: golden-meadow-cedar-dv

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

@@ -26,6 +26,7 @@ import {
   MAX_AUDIO_BASE64_BYTES,
   MAX_RELAY_SESSIONS_GLOBAL,
   MAX_RELAY_SESSIONS_PER_CONN,
+  broadcastRelayTurnStarted,
   broadcastToOwner,
   drainingRelaySessions,
   ensureRelayTurn,
@@ -186,18 +187,20 @@ export function sendTalkRealtimeRelayAudio(params: {
   }
   const session = getRelaySession(params.relaySessionId, params.connId);
   const audio = decodeTalkRelayAudioBase64(params.audioBase64, "Realtime relay");
-  const turnId = ensureRelayTurn(session);
-  session.bridge.sendAudio(audio);
+  const recorded = session.harness.recordInputAudio(audio);
+  if (!recorded) {
+    return;
+  }
+  broadcastRelayTurnStarted(session, recorded.turn.event);
   broadcastToOwner(session.context, session.connId, {
     relaySessionId: session.id,
     type: "inputAudio",
     byteLength: audio.byteLength,
-    talkEvent: session.harness.talk.emit({
-      type: "input.audio.delta",
-      turnId,
-      payload: { byteLength: audio.byteLength },
-    }),
+    talkEvent: recorded.inputAudioDelta,
   });
+  // Publish the recorded input before provider code can synchronously re-enter
+  // with output events, preserving the harness's authoritative sequence order.
+  session.bridge.sendAudio(audio);
   if (typeof params.timestamp === "number" && Number.isFinite(params.timestamp)) {
     session.bridge.setMediaTimestamp(params.timestamp);
   }

src/gateway/talk-realtime-relay-session-create.ts

@@ -8,8 +8,8 @@ import {
   REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ,
   type RealtimeVoiceCloseReason,
 } from "../talk/provider-types.js";
-import { createRealtimeVoiceSessionHarness } from "../talk/realtime-session-harness.js";
-import type { TalkEventInput } from "../talk/talk-session-controller.js";
+import { createRealtimeVoiceEventCapturingSessionHarness } from "../talk/realtime-session-harness.js";
+import type { TalkEvent, TalkEventInput } from "../talk/talk-session-controller.js";
 import { VOICE_TRANSCRIPT_QUEUE_POLICY } from "../talk/voice-transcript.js";
 import {
   createTalkClientAgentConsultRunner,
@@ -41,6 +41,7 @@ import {
   RELAY_SESSION_TTL_MS,
   RELAY_TRANSCRIPT_ECHO_LOOKBACK_MS,
   adoptRelayProviderToolCallId,
+  broadcastRelayTurnStarted,
   broadcastToOwner,
   ensureRelayTurn,
   relaySessions,
@@ -78,7 +79,7 @@ export function createTalkRealtimeRelaySession(
   if (expiresAtMs === undefined) {
     throw new Error("Realtime relay session expiry is outside the supported Date range");
   }
-  const harness = createRealtimeVoiceSessionHarness({
+  const harness = createRealtimeVoiceEventCapturingSessionHarness({
     talk: {
       sessionId: relaySessionId,
       mode: "realtime",
@@ -94,16 +95,19 @@ export function createTalkRealtimeRelaySession(
       inputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
       outputAudioStarted: () => ({}),
       outputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
-      outputAudioDone: (reason) => ({ reason }),
+      outputAudioDone: (reason, details) =>
+        details?.markName ? { markName: details.markName } : { reason },
     },
     transcriptLookbackMs: RELAY_TRANSCRIPT_ECHO_LOOKBACK_MS,
     captureBridgeEvents: false,
   });
-  const emit = (event: TalkRealtimeRelayEventPayload, talkEvent?: TalkEventInput) =>
+  const broadcastEvent = (event: TalkRealtimeRelayEventPayload, talkEvent?: TalkEvent) =>
     broadcastToOwner(params.context, params.connId, {
       ...event,
-      ...(talkEvent ? { talkEvent: harness.emit(talkEvent) } : {}),
+      ...(talkEvent ? { talkEvent } : {}),
     });
+  const emit = (event: TalkRealtimeRelayEventPayload, talkEvent?: TalkEventInput) =>
+    broadcastEvent(event, talkEvent ? harness.emit(talkEvent) : undefined);
   let currentOutputItemId: string | undefined;
   let currentOutputResponseId: string | undefined;
   let ready = false;
@@ -217,8 +221,12 @@ export function createTalkRealtimeRelaySession(
         if (!relay) {
           return;
         }
-        const turnId = ensureRelayTurn(relay);
-        emit(
+        const recorded = relay.harness.recordOutputAudio(audio);
+        broadcastRelayTurnStarted(relay, recorded.turn.event);
+        if (recorded.outputAudioStarted) {
+          broadcastEvent({ relaySessionId, type: "audioStarted" }, recorded.outputAudioStarted);
+        }
+        broadcastEvent(
           {
             relaySessionId,
             type: "audio",
@@ -226,11 +234,7 @@ export function createTalkRealtimeRelaySession(
             ...(currentOutputItemId ? { itemId: currentOutputItemId } : {}),
             ...(currentOutputResponseId ? { responseId: currentOutputResponseId } : {}),
           },
-          {
-            type: "output.audio.delta",
-            turnId,
-            payload: { byteLength: audio.length },
-          },
+          recorded.outputAudioDelta,
         );
       },
       clearAudio: (reason) => {
@@ -238,15 +242,9 @@ export function createTalkRealtimeRelaySession(
         if (!relay) {
           return;
         }
-        const turnId = ensureRelayTurn(relay);
-        emit(
+        broadcastEvent(
           { relaySessionId, type: "clear", ...(reason ? { reason } : {}) },
-          {
-            type: "output.audio.done",
-            turnId,
-            payload: { reason: reason ?? "clear" },
-            final: true,
-          },
+          relay.harness.finishOutputAudio(reason ?? "clear"),
         );
       },
       sendMark: (markName) => {
@@ -254,15 +252,9 @@ export function createTalkRealtimeRelaySession(
         if (!relay) {
           return;
         }
-        const turnId = ensureRelayTurn(relay);
-        emit(
+        broadcastEvent(
           { relaySessionId, type: "mark", markName },
-          {
-            type: "output.audio.done",
-            turnId,
-            payload: { markName },
-            final: true,
-          },
+          relay.harness.finishOutputAudio("mark", { markName }),
         );
       },
     },

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

@@ -10,7 +10,7 @@ import type {
   RealtimeVoiceTool,
   RealtimeVoiceToolResultOptions,
 } from "../talk/provider-types.js";
-import type { RealtimeVoiceSessionHarness } from "../talk/realtime-session-harness.js";
+import type { RealtimeVoiceEventCapturingSessionHarness } from "../talk/realtime-session-harness.js";
 import type { RealtimeVoiceBridgeSession } from "../talk/session-runtime.js";
 import type { TalkEvent } from "../talk/talk-session-controller.js";
 import type { GatewayRequestContext } from "./server-methods/shared-types.js";
@@ -28,6 +28,7 @@ export const noFallbackRelayOutputFlush = () => {};
 export type TalkRealtimeRelayEventPayload =
   | { relaySessionId: string; type: "ready" }
   | { relaySessionId: string; type: "inputAudio"; byteLength: number }
+  | { relaySessionId: string; type: "audioStarted" }
   | {
       relaySessionId: string;
       type: "audio";
@@ -88,7 +89,7 @@ export type RelaySession = {
   connId: string;
   context: GatewayRequestContext;
   bridge: RealtimeVoiceBridgeSession;
-  harness: RealtimeVoiceSessionHarness;
+  harness: RealtimeVoiceEventCapturingSessionHarness;
   sessionKey?: string;
   agentId?: string;
   expiresAtMs: number;
@@ -214,14 +215,22 @@ function relayEventDeliveryOptions(
 }
 
 export function ensureRelayTurn(session: RelaySession): string {
-  const turn = session.harness.talk.ensureTurn();
-  if (turn.event) {
-    broadcastToOwner(session.context, session.connId, {
-      relaySessionId: session.id,
-      type: "inputAudio",
-      byteLength: 0,
-      talkEvent: turn.event,
-    });
-  }
+  const turn = session.harness.ensureTurn();
+  broadcastRelayTurnStarted(session, turn.event);
   return turn.turnId;
 }
+
+export function broadcastRelayTurnStarted(
+  session: RelaySession,
+  event: TalkEvent | undefined,
+): void {
+  if (!event) {
+    return;
+  }
+  broadcastToOwner(session.context, session.connId, {
+    relaySessionId: session.id,
+    type: "inputAudio",
+    byteLength: 0,
+    talkEvent: event,
+  });
+}

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

@@ -1526,6 +1526,11 @@ describe("talk realtime gateway relay", () => {
       language: "de",
     });
     await Promise.resolve();
+    const relay = relaySessions.get(session.relaySessionId);
+    expect(relay).toBeDefined();
+    if (!relay) {
+      throw new Error("expected active relay session");
+    }
 
     const sessionFields = expectRecordFields(session, {
       provider: "relay-test",
@@ -1566,13 +1571,28 @@ describe("talk realtime gateway relay", () => {
     expectRecordFields(readyEvent, { event: "talk.event", connIds: ["conn-1"] });
     expectDelivery(readyPayload, false);
 
+    const audioStartedPayload = findEventPayload(
+      events,
+      (payload) => payload.type === "audioStarted",
+    );
+    const audioStartedEvent = expectRecordFields(audioStartedPayload.talkEvent, {
+      type: "output.audio.started",
+    });
+    expect(relay.harness.talk.recentEvents).toContain(audioStartedPayload.talkEvent);
+    expectDelivery(audioStartedPayload, false);
+
     const audioPayload = findEventPayload(events, (payload) => payload.type === "audio");
     expectRecordFields(audioPayload, {
       relaySessionId: session.relaySessionId,
       type: "audio",
       audioBase64: Buffer.from("audio-out").toString("base64"),
     });
-    expectRecordFields(audioPayload.talkEvent, { type: "output.audio.delta" });
+    const audioEvent = expectRecordFields(audioPayload.talkEvent, { type: "output.audio.delta" });
+    if (typeof audioStartedEvent.seq !== "number" || typeof audioEvent.seq !== "number") {
+      throw new Error("Expected sequenced relay audio events");
+    }
+    expect(audioEvent.seq).toBe(audioStartedEvent.seq + 1);
+    expect(relay.harness.talk.recentEvents).toContain(audioPayload.talkEvent);
     expectDelivery(audioPayload, true);
 
     const markPayload = findEventPayload(events, (payload) => payload.type === "mark");
@@ -1581,6 +1601,12 @@ describe("talk realtime gateway relay", () => {
       type: "mark",
       markName: "mark-1",
     });
+    expectRecordFields(markPayload.talkEvent, {
+      type: "output.audio.done",
+      payload: { markName: "mark-1" },
+      final: true,
+    });
+    expect(relay.harness.talk.recentEvents).toContain(markPayload.talkEvent);
     expectDelivery(markPayload, false);
 
     const partialTranscript = findEventPayload(
@@ -1718,6 +1744,7 @@ describe("talk realtime gateway relay", () => {
       byteLength: Buffer.from("audio-in").byteLength,
     });
     expectRecordFields(inputAudioPayload.talkEvent, { type: "input.audio.delta" });
+    expect(relay.harness.talk.recentEvents).toContain(inputAudioPayload.talkEvent);
     expectDelivery(inputAudioPayload, true);
 
     const clearPayload = findEventPayload(events, (payload) => payload.type === "clear");
@@ -1786,6 +1813,54 @@ describe("talk realtime gateway relay", () => {
     expectDelivery(closePayload, false);
   });
 
+  it("publishes recorded input before synchronous provider output", () => {
+    const events: Array<{ payload: Record<string, unknown> }> = [];
+    const provider: RealtimeVoiceProviderPlugin = {
+      id: "relay-test",
+      label: "Relay Test",
+      isConfigured: () => true,
+      createBridge: (request) => ({
+        connect: vi.fn(async () => undefined),
+        sendAudio: vi.fn(() => request.onAudio(Buffer.from("provider-output"))),
+        setMediaTimestamp: vi.fn(),
+        handleBargeIn: vi.fn(),
+        submitToolResult: vi.fn(),
+        acknowledgeMark: vi.fn(),
+        close: vi.fn(),
+        isConnected: vi.fn(() => true),
+      }),
+    };
+    const session = createTalkRealtimeRelaySession({
+      context: {
+        broadcastToConnIds: (_event: string, payload: Record<string, unknown>) => {
+          events.push({ payload });
+        },
+      } as never,
+      connId: "conn-1",
+      provider,
+      providerConfig: {},
+      instructions: "brief",
+      tools: [],
+    });
+
+    sendTalkRealtimeRelayAudio({
+      relaySessionId: session.relaySessionId,
+      connId: "conn-1",
+      audioBase64: Buffer.from("browser-input").toString("base64"),
+    });
+
+    const sequencedEvents = events
+      .map(({ payload }) => payload.talkEvent)
+      .filter((event): event is Record<string, unknown> => Boolean(event));
+    expect(sequencedEvents.map((event) => event.type)).toEqual([
+      "turn.started",
+      "input.audio.delta",
+      "output.audio.started",
+      "output.audio.delta",
+    ]);
+    expect(sequencedEvents.map((event) => event.seq)).toEqual([1, 2, 3, 4]);
+  });
+
   it("emits generic issue details when relay connect fails", async () => {
     const provider: RealtimeVoiceProviderPlugin = {
       id: "openai",

src/plugin-sdk/realtime-voice.test.ts

@@ -1,9 +1,12 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, expectTypeOf, it, vi } from "vitest";
+import * as realtimeVoiceSdk from "./realtime-voice.js";
 import {
   createRealtimeVoiceAudioQueue,
+  createRealtimeVoiceSessionHarness,
   normalizeRealtimeVoiceResponseOutcome,
   RealtimeVoiceSessionLifecycle,
   type RealtimeVoiceSessionConnection,
+  type RealtimeVoiceSessionHarness,
 } from "./realtime-voice.js";
 
 function connectLifecycle(
@@ -237,6 +240,42 @@ describe("RealtimeVoiceSessionLifecycle", () => {
   });
 });
 
+describe("realtime voice session harness SDK contract", () => {
+  it("keeps legacy helper results for plugin consumers", () => {
+    const harness: RealtimeVoiceSessionHarness = createRealtimeVoiceSessionHarness({
+      talk: {
+        sessionId: "plugin-session",
+        mode: "realtime",
+        transport: "gateway-relay",
+        brain: "agent-consult",
+        provider: "test",
+      },
+      talkPayloads: {
+        turnStarted: () => ({}),
+        turnEnded: (reason) => ({ reason }),
+        inputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
+        outputAudioStarted: () => ({}),
+        outputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
+        outputAudioDone: (reason) => ({ reason }),
+      },
+    });
+
+    expect(harness.ensureTurn()).toBe("turn-1");
+    expect(harness.recordInputAudio(Buffer.from([1]))).toBe(true);
+    expect(harness.recordOutputAudio(Buffer.from([2]))).toBeUndefined();
+    expect(harness.finishOutputAudio("completed")).toBeUndefined();
+    expect(harness.endTurn("completed")).toBeUndefined();
+    expectTypeOf<RealtimeVoiceSessionHarness["ensureTurn"]>().returns.toEqualTypeOf<string>();
+    expectTypeOf<
+      RealtimeVoiceSessionHarness["recordInputAudio"]
+    >().returns.toEqualTypeOf<boolean>();
+    expectTypeOf<RealtimeVoiceSessionHarness["recordOutputAudio"]>().returns.toEqualTypeOf<void>();
+    expectTypeOf<RealtimeVoiceSessionHarness["finishOutputAudio"]>().returns.toEqualTypeOf<void>();
+    expectTypeOf<RealtimeVoiceSessionHarness["endTurn"]>().returns.toEqualTypeOf<void>();
+    expect("createRealtimeVoiceEventCapturingSessionHarness" in realtimeVoiceSdk).toBe(false);
+  });
+});
+
 describe("normalizeRealtimeVoiceResponseOutcome", () => {
   it.each([
     [

src/talk/realtime-session-harness.test.ts

@@ -2,31 +2,38 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { RealtimeVoiceProviderPlugin } from "../plugins/types.js";
 import type { RealtimeVoiceBridge } from "./provider-types.js";
-import { createRealtimeVoiceSessionHarness } from "./realtime-session-harness.js";
+import {
+  createRealtimeVoiceEventCapturingSessionHarness,
+  createRealtimeVoiceSessionHarness,
+} from "./realtime-session-harness.js";
 
 afterEach(() => {
   vi.useRealTimers();
 });
 
+const defaultHarnessParams: Parameters<typeof createRealtimeVoiceSessionHarness>[0] = {
+  talk: {
+    sessionId: "test-session",
+    mode: "realtime",
+    transport: "gateway-relay",
+    brain: "agent-consult",
+    provider: "test",
+  },
+  talkPayloads: {
+    turnStarted: () => ({ surface: "test" }),
+    turnEnded: (reason) => ({ reason }),
+    inputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
+    outputAudioStarted: () => ({ surface: "test" }),
+    outputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
+    outputAudioDone: (reason) => ({ reason }),
+  },
+};
+
 function createHarness(
   overrides: Partial<Parameters<typeof createRealtimeVoiceSessionHarness>[0]> = {},
 ) {
   return createRealtimeVoiceSessionHarness({
-    talk: {
-      sessionId: "test-session",
-      mode: "realtime",
-      transport: "gateway-relay",
-      brain: "agent-consult",
-      provider: "test",
-    },
-    talkPayloads: {
-      turnStarted: () => ({ surface: "test" }),
-      turnEnded: (reason) => ({ reason }),
-      inputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
-      outputAudioStarted: () => ({ surface: "test" }),
-      outputAudioDelta: (audio) => ({ byteLength: audio.byteLength }),
-      outputAudioDone: (reason) => ({ reason }),
-    },
+    ...defaultHarnessParams,
     ...overrides,
   });
 }
@@ -137,13 +144,14 @@ describe("realtime voice session harness", () => {
       1,
     );
   });
-  it("keeps shared Talk events ordered across input, output, and turn completion", () => {
+  it("keeps legacy helpers compatible while ordering Talk events", () => {
     const harness = createHarness();
 
+    expect(harness.ensureTurn()).toBe("turn-1");
     expect(harness.recordInputAudio(Buffer.from([1, 2]))).toBe(true);
-    harness.recordOutputAudio(Buffer.from([3, 4, 5]));
-    harness.finishOutputAudio("response.done");
-    harness.endTurn("response.done");
+    expect(harness.recordOutputAudio(Buffer.from([3, 4, 5]))).toBeUndefined();
+    expect(harness.finishOutputAudio("response.done")).toBeUndefined();
+    expect(harness.endTurn("response.done")).toBeUndefined();
 
     expect(harness.talk.recentEvents.map((event) => event.type)).toEqual([
       "turn.started",
@@ -156,6 +164,41 @@ describe("realtime voice session harness", () => {
     expect(harness.talk.recentEvents.map((event) => event.seq)).toEqual([1, 2, 3, 4, 5, 6]);
   });
 
+  it("returns the exact Talk events emitted by audio helpers", () => {
+    const harness = createRealtimeVoiceEventCapturingSessionHarness({
+      ...defaultHarnessParams,
+      talkPayloads: {
+        ...defaultHarnessParams.talkPayloads,
+        outputAudioDone: (reason, details) =>
+          details?.markName ? { markName: details.markName } : { reason },
+      },
+    });
+
+    const input = harness.recordInputAudio(Buffer.from([1, 2]));
+    const output = harness.recordOutputAudio(Buffer.from([3, 4, 5]));
+    const done = harness.finishOutputAudio("mark", { markName: "played" });
+
+    expect(input).toMatchObject({
+      turn: { turnId: "turn-1", event: { type: "turn.started" } },
+      inputAudioDelta: { type: "input.audio.delta", turnId: "turn-1" },
+    });
+    expect(output).toMatchObject({
+      turn: { turnId: "turn-1" },
+      outputAudioStarted: { type: "output.audio.started", turnId: "turn-1" },
+      outputAudioDelta: { type: "output.audio.delta", turnId: "turn-1" },
+    });
+    expect(done).toMatchObject({
+      type: "output.audio.done",
+      turnId: "turn-1",
+      payload: { markName: "played" },
+    });
+    expect(input?.turn.event).toBe(harness.talk.recentEvents[0]);
+    expect(input?.inputAudioDelta).toBe(harness.talk.recentEvents[1]);
+    expect(output.outputAudioStarted).toBe(harness.talk.recentEvents[2]);
+    expect(output.outputAudioDelta).toBe(harness.talk.recentEvents[3]);
+    expect(done).toBe(harness.talk.recentEvents[4]);
+  });
+
   it("honors a caller-specific recent Talk event limit", () => {
     const harness = createHarness({
       talk: {

src/talk/realtime-session-harness.ts

@@ -38,6 +38,7 @@ import {
 import type { TalkEvent, TalkEventInput } from "./talk-events.js";
 import {
   createTalkSessionController,
+  type TalkEnsureTurnResult,
   type TalkSessionController,
   type TalkSessionControllerParams,
   type TalkTurnResult,
@@ -74,6 +75,17 @@ type RealtimeVoiceSessionHarnessTalkPayloads = {
   outputAudioDone: (reason: string) => unknown;
 };
 
+type RealtimeVoiceOutputAudioDoneDetails = {
+  markName?: string;
+};
+
+type RealtimeVoiceEventCapturingTalkPayloads = Omit<
+  RealtimeVoiceSessionHarnessTalkPayloads,
+  "outputAudioDone"
+> & {
+  outputAudioDone: (reason: string, details?: RealtimeVoiceOutputAudioDoneDetails) => unknown;
+};
+
 type RealtimeVoiceSessionHarnessEchoSuppression = {
   bytesPerMs: number;
   tailMs: number;
@@ -103,7 +115,18 @@ type RealtimeVoiceSessionHarnessHealth = ReturnType<typeof getRealtimeVoiceTrans
     }>;
   };
 
-export type RealtimeVoiceSessionHarness<TForcedConsultContext = unknown> = {
+type RealtimeVoiceInputAudioEvents = {
+  inputAudioDelta: TalkEvent;
+  turn: TalkEnsureTurnResult;
+};
+
+type RealtimeVoiceOutputAudioEvents = {
+  outputAudioDelta: TalkEvent;
+  outputAudioStarted?: TalkEvent;
+  turn: TalkEnsureTurnResult;
+};
+
+type RealtimeVoiceSessionHarnessBase<TForcedConsultContext> = {
   readonly forcedConsults: RealtimeVoiceForcedConsultCoordinator<TForcedConsultContext>;
   readonly outputActivity: RealtimeVoiceOutputActivityTracker;
   readonly talk: TalkSessionController;
@@ -112,10 +135,7 @@ export type RealtimeVoiceSessionHarness<TForcedConsultContext = unknown> = {
   close(): void;
   createBridge(params: RealtimeVoiceBridgeSessionParams): RealtimeVoiceBridgeSession;
   emit<TPayload>(input: TalkEventInput<TPayload>): TalkEvent<TPayload>;
-  ensureTurn(): string;
-  endTurn(reason?: string): void;
   finishResponse(outcome: RealtimeVoiceResponseOutcome): TalkTurnResult;
-  finishOutputAudio(reason: string): void;
   flushOutput(flush: () => void): void;
   getHealth(params: {
     providerConnected: boolean;
@@ -124,12 +144,54 @@ export type RealtimeVoiceSessionHarness<TForcedConsultContext = unknown> = {
   handleBargeIn(options: RealtimeVoiceBargeInOptions, flushOutput: () => void): void;
   isLikelyAssistantEchoTranscript(text: string): boolean;
   isOutputPlaybackWindowActive(): boolean;
-  recordInputAudio(audio: Buffer): boolean;
-  recordOutputAudio(audio: Buffer, activity?: RealtimeVoiceOutputActivityDelta): void;
   recordTranscript(role: RealtimeVoiceRole, text: string): RealtimeVoiceTranscriptEntry;
 };
 
-export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknown>(params: {
+type RealtimeVoiceSessionHarnessMethods = {
+  ensureTurn(): string;
+  endTurn(reason?: string): void;
+  finishOutputAudio(reason: string): void;
+  recordInputAudio(audio: Buffer): boolean;
+  recordOutputAudio(audio: Buffer, activity?: RealtimeVoiceOutputActivityDelta): void;
+};
+
+type RealtimeVoiceEventCapturingSessionHarnessMethods = {
+  ensureTurn(): TalkEnsureTurnResult;
+  endTurn(reason?: string): TalkTurnResult;
+  finishOutputAudio(
+    reason: string,
+    details?: RealtimeVoiceOutputAudioDoneDetails,
+  ): TalkEvent | undefined;
+  recordInputAudio(audio: Buffer): RealtimeVoiceInputAudioEvents | undefined;
+  recordOutputAudio(
+    audio: Buffer,
+    activity?: RealtimeVoiceOutputActivityDelta,
+  ): RealtimeVoiceOutputAudioEvents;
+};
+
+export type RealtimeVoiceSessionHarness<TForcedConsultContext = unknown> =
+  RealtimeVoiceSessionHarnessBase<TForcedConsultContext> & RealtimeVoiceSessionHarnessMethods;
+
+export type RealtimeVoiceEventCapturingSessionHarness<TForcedConsultContext = unknown> =
+  RealtimeVoiceSessionHarnessBase<TForcedConsultContext> &
+    RealtimeVoiceEventCapturingSessionHarnessMethods;
+
+type RealtimeVoiceSessionHarnessImplementation<TForcedConsultContext> =
+  RealtimeVoiceSessionHarnessBase<TForcedConsultContext> & {
+    ensureTurn(): string | TalkEnsureTurnResult;
+    endTurn(reason?: string): TalkTurnResult | undefined;
+    finishOutputAudio(
+      reason: string,
+      details?: RealtimeVoiceOutputAudioDoneDetails,
+    ): TalkEvent | undefined;
+    recordInputAudio(audio: Buffer): boolean | RealtimeVoiceInputAudioEvents | undefined;
+    recordOutputAudio(
+      audio: Buffer,
+      activity?: RealtimeVoiceOutputActivityDelta,
+    ): RealtimeVoiceOutputAudioEvents | undefined;
+  };
+
+type RealtimeVoiceSessionHarnessParams = {
   talk: TalkSessionControllerParams;
   talkPayloads: RealtimeVoiceSessionHarnessTalkPayloads;
   onTalkEvent?: (event: TalkEvent) => void;
@@ -138,7 +200,38 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
   echoSuppression?: RealtimeVoiceSessionHarnessEchoSuppression;
   transcriptLookbackMs?: number;
   captureBridgeEvents?: boolean;
-}): RealtimeVoiceSessionHarness<TForcedConsultContext> {
+};
+
+type RealtimeVoiceEventCapturingSessionHarnessParams = Omit<
+  RealtimeVoiceSessionHarnessParams,
+  "talkPayloads"
+> & {
+  talkPayloads: RealtimeVoiceEventCapturingTalkPayloads;
+};
+
+export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknown>(
+  params: RealtimeVoiceSessionHarnessParams,
+): RealtimeVoiceSessionHarness<TForcedConsultContext> {
+  return createRealtimeVoiceSessionHarnessImplementation(
+    params,
+    false,
+  ) as RealtimeVoiceSessionHarness<TForcedConsultContext>;
+}
+
+/** Gateway-internal factory that returns the exact Talk events emitted by each helper. */
+export function createRealtimeVoiceEventCapturingSessionHarness<TForcedConsultContext = unknown>(
+  params: RealtimeVoiceEventCapturingSessionHarnessParams,
+): RealtimeVoiceEventCapturingSessionHarness<TForcedConsultContext> {
+  return createRealtimeVoiceSessionHarnessImplementation(
+    params,
+    true,
+  ) as RealtimeVoiceEventCapturingSessionHarness<TForcedConsultContext>;
+}
+
+function createRealtimeVoiceSessionHarnessImplementation<TForcedConsultContext = unknown>(
+  params: RealtimeVoiceEventCapturingSessionHarnessParams,
+  captureEvents: boolean,
+): RealtimeVoiceSessionHarnessImplementation<TForcedConsultContext> {
   let closed = false;
   let bridge: RealtimeVoiceBridgeSession | undefined;
   let lastInputAt: string | undefined;
@@ -178,10 +271,10 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
       })
     : undefined;
 
-  const ensureTurn = () => {
-    const turnId = talk.ensureTurn({ payload: params.talkPayloads.turnStarted() }).turnId;
-    responseOwnerTurnId ??= turnId;
-    return turnId;
+  const ensureTurnWithEvents = () => {
+    const result = talk.ensureTurn({ payload: params.talkPayloads.turnStarted() });
+    responseOwnerTurnId ??= result.turnId;
+    return result;
   };
 
   const rememberSettledResponse = (responseId: string | undefined): void => {
@@ -202,7 +295,7 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
     if (event.direction !== "server" || event.type !== "response.created") {
       return;
     }
-    responseOwnerTurnId = ensureTurn();
+    responseOwnerTurnId = ensureTurnWithEvents().turnId;
     responseOwnerId = event.responseId;
     suppressNextUnkeyedLegacyTerminal = false;
   };
@@ -286,7 +379,7 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
     flush();
   };
 
-  const harness: RealtimeVoiceSessionHarness<TForcedConsultContext> = {
+  const harness: RealtimeVoiceSessionHarnessImplementation<TForcedConsultContext> = {
     forcedConsults,
     outputActivity,
     talk,
@@ -331,19 +424,26 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
       return bridge;
     },
     emit: (input) => talk.emit(input),
-    ensureTurn,
+    ensureTurn() {
+      const result = ensureTurnWithEvents();
+      return captureEvents ? result : result.turnId;
+    },
     endTurn(reason = "completed") {
       const result = talk.endTurn({ payload: params.talkPayloads.turnEnded(reason) });
       if (result.ok) {
         responseOwnerTurnId = undefined;
         responseOwnerId = undefined;
       }
+      return captureEvents ? result : undefined;
     },
     finishResponse(outcome) {
       return finishResponse(outcome, "typed");
     },
-    finishOutputAudio(reason) {
-      talk.finishOutputAudio({ payload: params.talkPayloads.outputAudioDone(reason) });
+    finishOutputAudio(reason, details) {
+      const result = talk.finishOutputAudio({
+        payload: params.talkPayloads.outputAudioDone(reason, details),
+      });
+      return captureEvents ? result : undefined;
     },
     flushOutput,
     getHealth(healthParams) {
@@ -392,30 +492,31 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
     isOutputPlaybackWindowActive() {
       return Date.now() <= Math.max(lastOutputPlayableUntilMs, suppressInputUntilMs);
     },
-    recordInputAudio(audio) {
+    recordInputAudio(audio: Buffer) {
       if (Date.now() < suppressInputUntilMs) {
         lastSuppressedInputAt = new Date().toISOString();
         suppressedInputBytes += audio.byteLength;
-        return false;
+        return captureEvents ? undefined : false;
       }
       lastInputAt = new Date().toISOString();
       lastInputBytes += audio.byteLength;
-      harness.emit({
+      const turn = ensureTurnWithEvents();
+      const inputAudioDelta = harness.emit({
         type: "input.audio.delta",
-        turnId: ensureTurn(),
+        turnId: turn.turnId,
         payload: params.talkPayloads.inputAudioDelta(audio),
       });
-      return true;
+      return captureEvents ? { inputAudioDelta, turn } : true;
     },
-    recordOutputAudio(audio, activity = {}) {
-      const turnId = ensureTurn();
-      talk.startOutputAudio({
-        turnId,
+    recordOutputAudio(audio: Buffer, activity: RealtimeVoiceOutputActivityDelta = {}) {
+      const turn = ensureTurnWithEvents();
+      const output = talk.startOutputAudio({
+        turnId: turn.turnId,
         payload: params.talkPayloads.outputAudioStarted(),
       });
-      harness.emit({
+      const outputAudioDelta = harness.emit({
         type: "output.audio.delta",
-        turnId,
+        turnId: turn.turnId,
         payload: params.talkPayloads.outputAudioDelta(audio),
       });
       let audioMs = activity.audioMs;
@@ -438,11 +539,21 @@ export function createRealtimeVoiceSessionHarness<TForcedConsultContext = unknow
         sinkAudioBytes: activity.sinkAudioBytes ?? audio.byteLength,
       });
       lastOutputAt = new Date().toISOString();
+      return captureEvents
+        ? {
+            outputAudioDelta,
+            ...(output.event ? { outputAudioStarted: output.event } : {}),
+            turn,
+          }
+        : undefined;
     },
     recordTranscript: (role, text) => recordRealtimeVoiceTranscript(transcript, role, text),
   };
 
-  harnessResponseOwners.set(harness, { claimResponseEvent, finishLegacyEvent });
+  harnessResponseOwners.set(harness as RealtimeVoiceSessionHarness<TForcedConsultContext>, {
+    claimResponseEvent,
+    finishLegacyEvent,
+  });
 
   return harness;
 }

ui/src/pages/chat/realtime-talk-gateway-relay-types.ts

@@ -5,6 +5,7 @@ export type GatewayRelayEvent = {
   talkEvent?: RealtimeTalkEvent;
 } & (
   | { type?: "ready" }
+  | { type?: "audioStarted" }
   | { type?: "audio"; audioBase64?: string }
   | { type?: "clear"; reason?: "barge-in" }
   | { type?: "mark"; markName?: string }

ui/src/pages/chat/realtime-talk-gateway-relay.ts

@@ -316,7 +316,9 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
       switch (event.type) {
         case "ready":
           this.ctx.callbacks.onStatus?.("listening");
-          return;
+          break;
+        case "audioStarted":
+          break;
         case "audio":
           if (event.audioBase64 && !this.playbackOverflowed) {
             this.cancelRequestedForPlayback = false;