From a1d14f0675ce218a74e19cf54fd72aa376cd49c5 Mon Sep 17 00:00:00 2001 From: Peng_Lx Date: Thu, 27 Aug 2026 06:04:06 +0800 Subject: [PATCH 1/8] feat(codex): add provider-native voice mode --- ...ProviderSessionStartup.integration.test.ts | 2 + apps/server/src/auth/RpcAuthorization.test.ts | 9 + apps/server/src/auth/RpcAuthorization.ts | 2 + .../Layers/CheckpointReactor.test.ts | 2 + .../Layers/ProviderCommandReactor.test.ts | 2 + .../Layers/ProviderRuntimeIngestion.test.ts | 2 + .../src/provider/Layers/CodexAdapter.test.ts | 31 +++ .../src/provider/Layers/CodexAdapter.ts | 23 ++ .../Layers/CodexSessionRuntime.test.ts | 12 + .../provider/Layers/CodexSessionRuntime.ts | 72 +++++ .../Layers/ProviderAdapterRegistry.test.ts | 2 + .../provider/Layers/ProviderService.test.ts | 91 +++++- .../src/provider/Layers/ProviderService.ts | 75 +++++ .../Layers/ProviderSessionReaper.test.ts | 2 + .../src/provider/Services/CodexAdapter.ts | 6 + .../src/provider/Services/ProviderAdapter.ts | 17 ++ .../src/provider/Services/ProviderService.ts | 11 + .../serverRuntimeStartup.reconcile.test.ts | 2 + apps/server/src/ws.ts | 31 +++ apps/web/src/components/ChatView.tsx | 10 + apps/web/src/components/chat/ChatComposer.tsx | 25 ++ .../chat/ComposerVoiceControl.test.tsx | 59 ++++ .../components/chat/ComposerVoiceControl.tsx | 102 +++++++ apps/web/src/hooks/useCodexRealtimeVoice.ts | 262 ++++++++++++++++++ docs/internals/providers.md | 15 + docs/user/providers-codex.md | 16 ++ .../src/state/threadCommands.ts | 12 + packages/contracts/src/provider.test.ts | 39 +++ packages/contracts/src/provider.ts | 34 +++ packages/contracts/src/rpc.ts | 19 ++ 30 files changed, 985 insertions(+), 2 deletions(-) create mode 100644 apps/web/src/components/chat/ComposerVoiceControl.test.tsx create mode 100644 apps/web/src/components/chat/ComposerVoiceControl.tsx create mode 100644 apps/web/src/hooks/useCodexRealtimeVoice.ts diff --git a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts index 78a33364f5a3..ffd3d79bd6e1 100644 --- a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts +++ b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts @@ -117,6 +117,8 @@ const startupDependencies = Layer.mergeAll( getInstanceInfo: () => Effect.die("unused"), rollbackConversation: () => Effect.die("unused"), uploadFeedback: () => Effect.die("unused"), + startRealtimeVoice: () => Effect.die("unused"), + stopRealtimeVoice: () => Effect.die("unused"), streamEvents: Stream.empty, }), ); diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 25971b0c0aec..5a7320e64971 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -43,6 +43,15 @@ describe("RPC authorization scopes", () => { ); }); + it("requires permission to operate on a thread for realtime voice", () => { + expect(requiredScopeForRpcMethod(WS_METHODS.providerRealtimeVoiceStart)).toBe( + AuthOrchestrationOperateScope, + ); + expect(requiredScopeForRpcMethod(WS_METHODS.providerRealtimeVoiceStop)).toBe( + AuthOrchestrationOperateScope, + ); + }); + it("reads the reviewer menu under the same scope as the pull request it belongs to", () => { // The candidate list is a read like the detail beside it, and asking somebody for a review is // a write like every other pull request operation. diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 28ceac4cec99..fecf7d7f7567 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -87,6 +87,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.attachmentsCreateUploadUrl]: AuthOrchestrationOperateScope, [WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope, [WS_METHODS.providerUploadFeedback]: AuthOrchestrationOperateScope, + [WS_METHODS.providerRealtimeVoiceStart]: AuthOrchestrationOperateScope, + [WS_METHODS.providerRealtimeVoiceStop]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope, [WS_METHODS.vcsRefreshStatus]: AuthOrchestrationReadScope, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index ca4cb7afd9ab..fdfe12e3315d 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -126,6 +126,8 @@ function createProviderServiceHarness( }), rollbackConversation, uploadFeedback: () => unsupported(), + startRealtimeVoice: () => unsupported(), + stopRealtimeVoice: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index a22a7acfb705..5ef7fa4f2f1f 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -352,6 +352,8 @@ describe("ProviderCommandReactor", () => { }, rollbackConversation: () => unsupported(), uploadFeedback: () => unsupported(), + startRealtimeVoice: () => unsupported(), + stopRealtimeVoice: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 84858b6affe9..ae1837b4f151 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -126,6 +126,8 @@ function createProviderServiceHarness() { }, rollbackConversation: () => unsupported(), uploadFeedback: () => unsupported(), + startRealtimeVoice: () => unsupported(), + stopRealtimeVoice: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 4986d02c9b67..07a9f4b54416 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -108,6 +108,10 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { Promise.resolve({ threadId: "provider-thread-1" }), ); + public readonly startRealtimeVoiceImpl = vi.fn((sdp: string) => Promise.resolve(sdp)); + + public readonly stopRealtimeVoiceImpl = vi.fn(() => Promise.resolve(undefined)); + public readonly respondToRequestImpl = vi.fn( (_requestId: ApprovalRequestId, _decision: ProviderApprovalDecision): Promise => Promise.resolve(undefined), @@ -150,6 +154,12 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { return Effect.promise(() => this.uploadFeedbackImpl(reason)); } + startRealtimeVoice(sdp: string) { + return Effect.promise(() => this.startRealtimeVoiceImpl(sdp)); + } + + stopRealtimeVoice = Effect.promise(() => this.stopRealtimeVoiceImpl()); + respondToRequest(requestId: ApprovalRequestId, decision: ProviderApprovalDecision) { return Effect.promise(() => this.respondToRequestImpl(requestId, decision)); } @@ -372,6 +382,27 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { }), ); + it.effect("routes realtime voice signaling through the active Codex runtime", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const threadId = asThreadId("thread-voice"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + const runtime = sessionRuntimeFactory.lastRuntime; + NodeAssert.ok(runtime); + + const result = yield* adapter.startRealtimeVoice({ threadId, sdp: "offer-sdp" }); + yield* adapter.stopRealtimeVoice({ threadId }); + + NodeAssert.deepStrictEqual(result, { sdp: "offer-sdp" }); + NodeAssert.deepStrictEqual(runtime.startRealtimeVoiceImpl.mock.calls, [["offer-sdp"]]); + NodeAssert.equal(runtime.stopRealtimeVoiceImpl.mock.calls.length, 1); + }), + ); + it.effect("maps codex model options before sending a turn", () => Effect.gen(function* () { const adapter = yield* CodexAdapter; diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 0f7d999662e9..adc93009911b 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1870,6 +1870,27 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ), ); + const startRealtimeVoice: CodexAdapterShape["startRealtimeVoice"] = (input) => + requireSession(input.threadId).pipe( + Effect.flatMap((session) => session.runtime.startRealtimeVoice(input.sdp)), + Effect.map((sdp) => ({ sdp })), + Effect.mapError((cause) => + cause._tag === "ProviderAdapterSessionNotFoundError" + ? cause + : mapCodexRuntimeError(input.threadId, "thread/realtime/start", cause), + ), + ); + + const stopRealtimeVoice: CodexAdapterShape["stopRealtimeVoice"] = (input) => + requireSession(input.threadId).pipe( + Effect.flatMap((session) => session.runtime.stopRealtimeVoice), + Effect.mapError((cause) => + cause._tag === "ProviderAdapterSessionNotFoundError" + ? cause + : mapCodexRuntimeError(input.threadId, "thread/realtime/stop", cause), + ), + ); + const readThread: CodexAdapterShape["readThread"] = (threadId) => requireSession(threadId).pipe( Effect.flatMap((session) => session.runtime.readThread), @@ -2005,6 +2026,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( startSession, sendTurn, interruptTurn, + startRealtimeVoice, + stopRealtimeVoice, readThread, rollbackThread, uploadFeedback, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 6a6cec5b1e61..ae69153b2b05 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -16,6 +16,7 @@ import { } from "../CodexDeveloperInstructions.ts"; import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; import { + buildCodexRealtimeStartParams, buildTurnStartParams, describeMcpElicitation, hasConfiguredMcpServer, @@ -26,6 +27,17 @@ import { } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); +describe("buildCodexRealtimeStartParams", () => { + it("uses Codex's v3 WebRTC audio transport", () => { + NodeAssert.deepStrictEqual(buildCodexRealtimeStartParams("provider-thread-1", "offer-sdp"), { + threadId: "provider-thread-1", + outputModality: "audio", + version: "v3", + transport: { type: "webrtc", sdp: "offer-sdp" }, + }); + }); +}); + describe("CodexSessionRuntimeIdentifierGenerationError", () => { it("retains identifier purpose and the random source failure", () => { const cause = new Error("random source unavailable"); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index b34067b7fb90..3f9d2f328906 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -25,6 +25,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -191,6 +192,8 @@ export interface CodexSessionRuntimeShape { input: CodexSessionRuntimeSendTurnInput, ) => Effect.Effect; readonly interruptTurn: (turnId?: TurnId) => Effect.Effect; + readonly startRealtimeVoice: (sdp: string) => Effect.Effect; + readonly stopRealtimeVoice: Effect.Effect; readonly readThread: Effect.Effect; readonly rollbackThread: ( numTurns: number, @@ -210,6 +213,15 @@ export interface CodexSessionRuntimeShape { readonly close: Effect.Effect; } +export function buildCodexRealtimeStartParams(threadId: string, sdp: string) { + return { + threadId, + outputModality: "audio", + version: "v3", + transport: { type: "webrtc", sdp }, + } as const; +} + export type CodexSessionRuntimeError = | CodexErrors.CodexAppServerError | CodexSessionRuntimePendingApprovalNotFoundError @@ -1130,6 +1142,7 @@ export const makeCodexSessionRuntime = ( const collabChildLiveTurnsRef = yield* Ref.make(new Map()); const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter(); const closedRef = yield* Ref.make(false); + const pendingRealtimeSdpRef = yield* Ref.make | null>(null); // `~` is not shell-expanded when env vars are set via // `child_process.spawn`; `expandHomePath` lets a configured @@ -1712,6 +1725,23 @@ export const makeCodexSessionRuntime = ( ), ); + yield* client.handleServerNotification("thread/realtime/sdp", (payload) => + currentSessionProviderThreadId.pipe( + Effect.flatMap((providerThreadId) => { + if (providerThreadId && payload.threadId !== providerThreadId) { + return Effect.void; + } + return Ref.get(pendingRealtimeSdpRef).pipe( + Effect.flatMap((pending) => + pending === null + ? Effect.void + : Deferred.succeed(pending, payload.sdp).pipe(Effect.asVoid), + ), + ); + }), + ), + ); + yield* client.handleServerRequest("item/commandExecution/requestApproval", (payload) => Effect.gen(function* () { const requestId = ApprovalRequestId.make(yield* randomUUIDv4("command-approval-request")); @@ -2073,6 +2103,12 @@ export const makeCodexSessionRuntime = ( } yield* settlePendingApprovals("cancel"); yield* settlePendingUserInputs({}); + const providerThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); + if (providerThreadId) { + yield* client.raw + .request("thread/realtime/stop", { threadId: providerThreadId }) + .pipe(Effect.ignore); + } yield* updateSession(sessionRef, { status: "closed", activeTurnId: undefined, @@ -2180,6 +2216,42 @@ export const makeCodexSessionRuntime = ( turnId: effectiveTurnId, }); }), + startRealtimeVoice: (sdp) => + Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + const pending = yield* Deferred.make(); + const installed = yield* Ref.modify(pendingRealtimeSdpRef, (current) => + current === null ? ([true, pending] as const) : ([false, current] as const), + ); + if (!installed) { + return yield* CodexErrors.CodexAppServerRequestError.invalidRequest( + "A Codex realtime voice session is already connecting.", + { method: "thread/realtime/start" }, + ); + } + + return yield* Effect.gen(function* () { + yield* client.raw.request( + "thread/realtime/start", + buildCodexRealtimeStartParams(providerThreadId, sdp), + ); + const answer = yield* Deferred.await(pending).pipe(Effect.timeoutOption("20 seconds")); + if (Option.isNone(answer)) { + yield* client.raw + .request("thread/realtime/stop", { threadId: providerThreadId }) + .pipe(Effect.ignore); + return yield* CodexErrors.CodexAppServerRequestError.invalidRequest( + "Codex realtime voice did not return an SDP answer.", + { method: "thread/realtime/start" }, + ); + } + return answer.value; + }).pipe(Effect.ensuring(Ref.set(pendingRealtimeSdpRef, null))); + }), + stopRealtimeVoice: Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + yield* client.raw.request("thread/realtime/stop", { threadId: providerThreadId }); + }), readThread: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; const response = yield* client.request("thread/read", { diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts index 280601275a7e..574b4ee86891 100644 --- a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts @@ -41,6 +41,8 @@ const fakeCodexAdapter: CodexAdapter.CodexAdapterShape = { readThread: vi.fn(), rollbackThread: vi.fn(), uploadFeedback: vi.fn(), + startRealtimeVoice: vi.fn(), + stopRealtimeVoice: vi.fn(), stopAll: vi.fn(), streamEvents: Stream.empty, }; diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index bd89dc4f8812..42806028807d 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -9,12 +9,13 @@ import type { ProviderSendTurnInput, ProviderSession, ProviderTurnStartResult, + ProviderRealtimeVoiceStartInput, + ProviderRealtimeVoiceStartResult, ProviderUploadFeedbackInput, ProviderUploadFeedbackResult, } from "@t3tools/contracts"; import { ApprovalRequestId, - EnvironmentId, EventId, ProviderDriverKind, ProviderInstanceId, @@ -206,6 +207,18 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { Effect.succeed({ feedbackId: `feedback-${input.threadId}` }), ); + const startRealtimeVoice = vi.fn( + ( + input: ProviderRealtimeVoiceStartInput, + ): Effect.Effect => + Effect.succeed({ sdp: `answer:${input.sdp}` }), + ); + + const stopRealtimeVoice = vi.fn( + (_input: { readonly threadId: ThreadId }): Effect.Effect => + Effect.void, + ); + const stopAll = vi.fn( (): Effect.Effect => Effect.sync(() => { @@ -228,7 +241,7 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { hasSession, readThread, rollbackThread, - ...(provider === CODEX_DRIVER ? { uploadFeedback } : {}), + ...(provider === CODEX_DRIVER ? { uploadFeedback, startRealtimeVoice, stopRealtimeVoice } : {}), stopAll, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); @@ -265,6 +278,8 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { readThread, rollbackThread, uploadFeedback, + startRealtimeVoice, + stopRealtimeVoice, stopAll, }; } @@ -1038,6 +1053,78 @@ routing.layer("ProviderServiceLive routing", (it) => { }), ); + it.effect("routes realtime voice signaling to the Codex adapter", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("thread-voice-route"); + yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + runtimeMode: "full-access", + }); + routing.codex.startRealtimeVoice.mockClear(); + routing.codex.stopRealtimeVoice.mockClear(); + + const result = yield* provider.startRealtimeVoice({ threadId, sdp: "offer-sdp" }); + yield* provider.stopRealtimeVoice({ threadId }); + + assert.deepStrictEqual(result, { sdp: "answer:offer-sdp" }); + assert.deepStrictEqual(routing.codex.startRealtimeVoice.mock.calls, [ + [{ threadId, sdp: "offer-sdp" }], + ]); + assert.deepStrictEqual(routing.codex.stopRealtimeVoice.mock.calls, [[{ threadId }]]); + }), + ); + + it.effect("recovers a stopped Codex session before starting realtime voice", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("thread-voice-recover"); + yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + cwd: "/tmp/voice-project", + runtimeMode: "full-access", + }); + yield* routing.codex.stopSession(threadId); + routing.codex.startSession.mockClear(); + routing.codex.startRealtimeVoice.mockClear(); + + const result = yield* provider.startRealtimeVoice({ threadId, sdp: "offer-sdp" }); + + assert.deepStrictEqual(result, { sdp: "answer:offer-sdp" }); + assert.equal(routing.codex.startSession.mock.calls.length, 1); + assert.deepStrictEqual(routing.codex.startRealtimeVoice.mock.calls, [ + [{ threadId, sdp: "offer-sdp" }], + ]); + }), + ); + + it.effect("rejects realtime voice for unsupported providers without restarting them", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("thread-voice-unsupported"); + yield* provider.startSession(threadId, { + provider: CLAUDE_AGENT_DRIVER, + providerInstanceId: claudeAgentInstanceId, + threadId, + runtimeMode: "full-access", + }); + yield* routing.claude.stopSession(threadId); + routing.claude.startSession.mockClear(); + + const error = yield* provider + .startRealtimeVoice({ threadId, sdp: "offer-sdp" }) + .pipe(Effect.flip); + + assert.instanceOf(error, ProviderValidationError); + assert.include(error.issue, "does not support realtime voice"); + assert.equal(routing.claude.startSession.mock.calls.length, 0); + }), + ); + it.effect("recovers a stopped Codex session before uploading feedback", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index b8cd0df539ac..ad9f6ccde779 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -14,6 +14,8 @@ import { NonNegativeInt, ThreadId, ProviderInterruptTurnInput, + ProviderRealtimeVoiceStartInput, + ProviderRealtimeVoiceStopInput, ProviderRespondToRequestInput, ProviderRespondToUserInputInput, ProviderSendTurnInput, @@ -864,6 +866,77 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); + const startRealtimeVoice: ProviderServiceMethod<"startRealtimeVoice"> = Effect.fn( + "startRealtimeVoice", + )(function* (rawInput) { + const input = yield* decodeInputOrValidationError({ + operation: "ProviderService.startRealtimeVoice", + schema: ProviderRealtimeVoiceStartInput, + payload: rawInput, + }); + let routed = yield* resolveRoutableSession({ + threadId: input.threadId, + operation: "ProviderService.startRealtimeVoice", + allowRecovery: false, + }); + if (!routed.adapter.startRealtimeVoice) { + return yield* toValidationError( + "ProviderService.startRealtimeVoice", + `Provider '${routed.adapter.provider}' does not support realtime voice.`, + ); + } + if (!routed.isActive) { + routed = yield* resolveRoutableSession({ + threadId: input.threadId, + operation: "ProviderService.startRealtimeVoice", + allowRecovery: true, + }); + } + const startVoice = routed.adapter.startRealtimeVoice; + if (!startVoice) { + return yield* toValidationError( + "ProviderService.startRealtimeVoice", + `Provider '${routed.adapter.provider}' does not support realtime voice.`, + ); + } + yield* Effect.annotateCurrentSpan({ + "provider.operation": "start-realtime-voice", + "provider.kind": routed.adapter.provider, + "provider.thread_id": input.threadId, + }); + return yield* startVoice(input); + }); + + const stopRealtimeVoice: ProviderServiceMethod<"stopRealtimeVoice"> = Effect.fn( + "stopRealtimeVoice", + )(function* (rawInput) { + const input = yield* decodeInputOrValidationError({ + operation: "ProviderService.stopRealtimeVoice", + schema: ProviderRealtimeVoiceStopInput, + payload: rawInput, + }); + const routed = yield* resolveRoutableSession({ + threadId: input.threadId, + operation: "ProviderService.stopRealtimeVoice", + allowRecovery: false, + }); + if (!routed.isActive) { + return; + } + if (!routed.adapter.stopRealtimeVoice) { + return yield* toValidationError( + "ProviderService.stopRealtimeVoice", + `Provider '${routed.adapter.provider}' does not support realtime voice.`, + ); + } + yield* Effect.annotateCurrentSpan({ + "provider.operation": "stop-realtime-voice", + "provider.kind": routed.adapter.provider, + "provider.thread_id": input.threadId, + }); + yield* routed.adapter.stopRealtimeVoice(input); + }); + const respondToRequest: ProviderServiceMethod<"respondToRequest"> = Effect.fn("respondToRequest")( function* (rawInput) { const input = yield* decodeInputOrValidationError({ @@ -1223,6 +1296,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( startSession, sendTurn, interruptTurn, + startRealtimeVoice, + stopRealtimeVoice, respondToRequest, respondToUserInput, stopSession, diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 0b1bc9e149f7..ab432b9f9c4c 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -185,6 +185,8 @@ describe("ProviderSessionReaper", () => { }, rollbackConversation: () => unsupported(), uploadFeedback: () => unsupported(), + startRealtimeVoice: () => unsupported(), + stopRealtimeVoice: () => unsupported(), streamEvents: Stream.empty, }; diff --git a/apps/server/src/provider/Services/CodexAdapter.ts b/apps/server/src/provider/Services/CodexAdapter.ts index a0d9c0c28e9e..48b1aa338521 100644 --- a/apps/server/src/provider/Services/CodexAdapter.ts +++ b/apps/server/src/provider/Services/CodexAdapter.ts @@ -17,6 +17,12 @@ import type { ProviderAdapterShape } from "./ProviderAdapter.ts"; * a branded driver kind as the nominal discriminant. */ export interface CodexAdapterShape extends ProviderAdapterShape { + readonly startRealtimeVoice: NonNullable< + ProviderAdapterShape["startRealtimeVoice"] + >; + readonly stopRealtimeVoice: NonNullable< + ProviderAdapterShape["stopRealtimeVoice"] + >; readonly uploadFeedback: NonNullable< ProviderAdapterShape["uploadFeedback"] >; diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 634745832b37..2e92c1d1fd8a 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -13,6 +13,9 @@ import type { ProviderDriverKind, ProviderUserInputAnswers, ProviderRuntimeEvent, + ProviderRealtimeVoiceStartInput, + ProviderRealtimeVoiceStartResult, + ProviderRealtimeVoiceStopInput, ProviderSendTurnInput, ProviderSession, ProviderSessionStartInput, @@ -70,6 +73,20 @@ export interface ProviderAdapterShape { */ readonly interruptTurn: (threadId: ThreadId, turnId?: TurnId) => Effect.Effect; + /** + * Start a provider-native realtime voice session when supported. + */ + readonly startRealtimeVoice?: ( + input: ProviderRealtimeVoiceStartInput, + ) => Effect.Effect; + + /** + * Stop a provider-native realtime voice session when supported. + */ + readonly stopRealtimeVoice?: ( + input: ProviderRealtimeVoiceStopInput, + ) => Effect.Effect; + /** * Respond to an interactive approval request. */ diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 545641d2e866..d78fbce31b04 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -14,6 +14,9 @@ import type { ProviderInterruptTurnInput, ProviderInstanceId, + ProviderRealtimeVoiceStartInput, + ProviderRealtimeVoiceStartResult, + ProviderRealtimeVoiceStopInput, ProviderRespondToRequestInput, ProviderRespondToUserInputInput, ProviderRuntimeEvent, @@ -60,6 +63,14 @@ export interface ProviderServiceShape { input: ProviderInterruptTurnInput, ) => Effect.Effect; + readonly startRealtimeVoice: ( + input: ProviderRealtimeVoiceStartInput, + ) => Effect.Effect; + + readonly stopRealtimeVoice: ( + input: ProviderRealtimeVoiceStopInput, + ) => Effect.Effect; + /** * Respond to a provider approval request. */ diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 485cd5bb08a4..f48863ee03a8 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -56,6 +56,8 @@ const makeProviderService = (liveThreadIds: ReadonlyArray = []) => getInstanceInfo: () => Effect.die("unused"), rollbackConversation: () => Effect.die("unused"), uploadFeedback: () => Effect.die("unused"), + startRealtimeVoice: () => Effect.die("unused"), + stopRealtimeVoice: () => Effect.die("unused"), streamEvents: Stream.empty, }) satisfies ProviderService.ProviderService["Service"]; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 226c82cdb1ac..c9cbf902bec7 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -44,6 +44,7 @@ import { ProjectSearchContentsError, ProjectSearchEntriesError, ProjectWriteFileError, + ProviderRealtimeVoiceError, ProviderUploadFeedbackError, RelayClientInstallFailedError, type RelayClientInstallProgressEvent, @@ -1603,6 +1604,36 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "provider" }, ), + [WS_METHODS.providerRealtimeVoiceStart]: (input) => + observeRpcEffect( + WS_METHODS.providerRealtimeVoiceStart, + providerService.startRealtimeVoice(input).pipe( + Effect.mapError( + (cause) => + new ProviderRealtimeVoiceError({ + threadId: input.threadId, + operation: "start", + cause, + }), + ), + ), + { "rpc.aggregate": "provider" }, + ), + [WS_METHODS.providerRealtimeVoiceStop]: (input) => + observeRpcEffect( + WS_METHODS.providerRealtimeVoiceStop, + providerService.stopRealtimeVoice(input).pipe( + Effect.mapError( + (cause) => + new ProviderRealtimeVoiceError({ + threadId: input.threadId, + operation: "stop", + cause, + }), + ), + ), + { "rpc.aggregate": "provider" }, + ), [WS_METHODS.serverUpdateProvider]: (input) => observeRpcEffect( WS_METHODS.serverUpdateProvider, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f0188af478c0..bfdecedd54b6 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -398,6 +398,7 @@ import { serverUpdateGuidance, } from "../versionSkew"; import { useAssetUrls } from "../assets/assetUrls"; +import { useCodexRealtimeVoice } from "../hooks/useCodexRealtimeVoice"; const IMAGE_ONLY_BOOTSTRAP_PROMPT = "[User attached one or more images without additional text. Respond using the conversation context and the attached image(s).]"; @@ -2313,6 +2314,14 @@ function ChatViewContent(props: ChatViewProps) { selectedProviderByThreadId ?? threadProvider, ); const selectedProvider: ProviderDriverKind = lockedProvider ?? unlockedSelectedProvider; + const codexRealtimeVoice = useCodexRealtimeVoice({ + environmentId, + threadId: routeKind === "server" ? activeThreadId : null, + enabled: + routeKind === "server" && + selectedProvider === ProviderDriverKind.make("codex") && + activeEnvironmentUnavailableState === null, + }); const phase = derivePhase(activeThread?.session ?? null); const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; const activeContextWindow = useMemo( @@ -7072,6 +7081,7 @@ function ChatViewContent(props: ChatViewProps) { activeProject?.defaultModelSelection } activeThreadModelSelection={activeThread?.modelSelection} + codexRealtimeVoice={codexRealtimeVoice} activeContextWindow={activeContextWindow} compactDisabled={compactDisabled} compactDisabledReason={compactDisabledReason} diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 6d0ca8a765ba..fe774cfe3364 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -108,6 +108,11 @@ import { shouldUseCompactComposerFooter, } from "../composerFooterLayout"; import { type ComposerPromptEditorHandle, ComposerPromptEditor } from "../ComposerPromptEditor"; +import { + type CodexRealtimeVoiceController, + supportsCodexRealtimeVoiceVersion, +} from "../../hooks/useCodexRealtimeVoice"; +import { ComposerVoiceControl } from "./ComposerVoiceControl"; import { ProviderModelPicker } from "./ProviderModelPicker"; import { type ComposerCommandItem, ComposerCommandMenu } from "./ComposerCommandMenu"; import { ComposerPendingApprovalActions } from "./ComposerPendingApprovalActions"; @@ -614,6 +619,7 @@ export interface ChatComposerProps { providerStatuses: ServerProvider[]; activeProjectDefaultModelSelection: ModelSelection | null | undefined; activeThreadModelSelection: ModelSelection | null | undefined; + codexRealtimeVoice: CodexRealtimeVoiceController; // Context window activeContextWindow: ContextWindowSnapshot | null; @@ -710,6 +716,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) providerStatuses, activeProjectDefaultModelSelection, activeThreadModelSelection, + codexRealtimeVoice, activeContextWindow, compactDisabled, compactDisabledReason, @@ -946,6 +953,9 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) () => selectedProviderEntry?.snapshot ?? null, [selectedProviderEntry], ); + const codexRealtimeVoiceVersionSupported = supportsCodexRealtimeVoiceVersion( + selectedProviderStatus?.version ?? null, + ); const selectedProviderModels = useMemo>( () => selectedProviderEntry?.models ?? [], [selectedProviderEntry], @@ -3555,6 +3565,21 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) > {showMobilePendingAnswerActions ? null : inlineTasksBadge} {showMobilePendingAnswerActions ? null : inlineStashBadge} + {selectedProvider === ProviderDriverKind.make("codex") && activeThreadId ? ( + + ) : null} = {}, +): CodexRealtimeVoiceController => ({ + supported: true, + status: "idle", + muted: false, + error: null, + start: async () => {}, + stop: async () => {}, + toggleMuted: () => {}, + ...overrides, +}); + +describe("ComposerVoiceControl", () => { + it("gates known Codex versions before realtime v3 shipped", () => { + expect(supportsCodexRealtimeVoiceVersion("0.144.0")).toBe(false); + expect(supportsCodexRealtimeVoiceVersion("codex-cli 0.145.0")).toBe(true); + expect(supportsCodexRealtimeVoiceVersion("0.146.1-alpha.1")).toBe(true); + expect(supportsCodexRealtimeVoiceVersion(null)).toBe(true); + }); + + it("renders a compact microphone action before voice starts", () => { + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain('aria-label="Talk to Codex"'); + expect(markup).toContain('data-codex-voice-state="idle"'); + }); + + it("shows mute and end actions while voice is live", () => { + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain("Voice live"); + expect(markup).toContain('aria-label="Mute microphone"'); + expect(markup).toContain('aria-label="End Codex voice"'); + }); + + it("renders the muted state without hiding the end action", () => { + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain("Muted"); + expect(markup).toContain('aria-label="Unmute microphone"'); + expect(markup).toContain('aria-label="End Codex voice"'); + }); +}); diff --git a/apps/web/src/components/chat/ComposerVoiceControl.tsx b/apps/web/src/components/chat/ComposerVoiceControl.tsx new file mode 100644 index 000000000000..9cb3d0cd828e --- /dev/null +++ b/apps/web/src/components/chat/ComposerVoiceControl.tsx @@ -0,0 +1,102 @@ +import { memo, type PointerEventHandler, type ReactElement } from "react"; +import { MicIcon, MicOffIcon, PhoneOffIcon } from "lucide-react"; + +import type { CodexRealtimeVoiceController } from "~/hooks/useCodexRealtimeVoice"; +import { cn } from "~/lib/utils"; +import { Button } from "../ui/button"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; + +const preserveComposerFocus: PointerEventHandler = (event) => { + event.preventDefault(); +}; + +function VoiceButtonTooltip(props: { readonly label: string; readonly children: ReactElement }) { + return ( + + }>{props.children} + {props.label} + + ); +} + +export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { + readonly voice: CodexRealtimeVoiceController; + readonly disabled: boolean; + readonly disabledReason?: string; +}) { + const { voice } = props; + const active = voice.status === "connecting" || voice.status === "live"; + + if (!active) { + const tooltip = !voice.supported + ? "Codex voice requires microphone and WebRTC support" + : props.disabled + ? (props.disabledReason ?? "Codex voice is unavailable while this thread connects") + : (voice.error ?? "Talk to Codex"); + return ( + + + + ); + } + + const label = + voice.status === "connecting" ? "Connecting…" : voice.muted ? "Muted" : "Voice live"; + return ( +
+
+ ); +}); diff --git a/apps/web/src/hooks/useCodexRealtimeVoice.ts b/apps/web/src/hooks/useCodexRealtimeVoice.ts new file mode 100644 index 000000000000..e6480f804a05 --- /dev/null +++ b/apps/web/src/hooks/useCodexRealtimeVoice.ts @@ -0,0 +1,262 @@ +import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { useCallback, useEffect, useRef, useState } from "react"; + +import { threadEnvironment } from "../state/threads"; +import { useAtomCommand } from "../state/use-atom-command"; + +export type CodexRealtimeVoiceStatus = "idle" | "connecting" | "live" | "error"; + +export interface CodexRealtimeVoiceController { + readonly supported: boolean; + readonly status: CodexRealtimeVoiceStatus; + readonly muted: boolean; + readonly error: string | null; + readonly start: () => Promise; + readonly stop: () => Promise; + readonly toggleMuted: () => void; +} + +interface LocalVoiceSession { + readonly peer: RTCPeerConnection; + readonly microphone: MediaStream; + readonly events: RTCDataChannel; + readonly audio: HTMLAudioElement; +} + +const ICE_GATHERING_TIMEOUT_MS = 5_000; +const MINIMUM_CODEX_REALTIME_VOICE_VERSION = [0, 145, 0] as const; + +export function supportsCodexRealtimeVoiceVersion(version: string | null): boolean { + if (!version) return true; + const match = version.match(/(\d+)\.(\d+)\.(\d+)/); + if (!match) return true; + const installed = match.slice(1).map(Number); + for (let index = 0; index < MINIMUM_CODEX_REALTIME_VOICE_VERSION.length; index += 1) { + const minimum = MINIMUM_CODEX_REALTIME_VOICE_VERSION[index]; + if (minimum === undefined) continue; + const difference = (installed[index] ?? 0) - minimum; + if (difference !== 0) return difference > 0; + } + return true; +} + +function waitForIceGathering(peer: RTCPeerConnection): Promise { + if (peer.iceGatheringState === "complete") { + return Promise.resolve(); + } + + return new Promise((resolve) => { + const finish = () => { + window.clearTimeout(timeoutId); + peer.removeEventListener("icegatheringstatechange", handleStateChange); + resolve(); + }; + const handleStateChange = () => { + if (peer.iceGatheringState === "complete") finish(); + }; + const timeoutId = window.setTimeout(finish, ICE_GATHERING_TIMEOUT_MS); + peer.addEventListener("icegatheringstatechange", handleStateChange); + }); +} + +function voiceErrorMessage(cause: unknown): string { + if (cause instanceof DOMException) { + if (cause.name === "NotAllowedError") return "Microphone access was denied."; + if (cause.name === "NotFoundError") return "No microphone was found."; + } + return "Codex voice could not connect. Try again."; +} + +function disposeLocalSession(session: LocalVoiceSession | null): void { + if (!session) return; + session.peer.onconnectionstatechange = null; + session.peer.ontrack = null; + session.events.close(); + session.microphone.getTracks().forEach((track) => track.stop()); + session.peer.close(); + session.audio.pause(); + session.audio.srcObject = null; +} + +export function useCodexRealtimeVoice(input: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId | null; + readonly enabled: boolean; +}): CodexRealtimeVoiceController { + const startRemoteVoice = useAtomCommand(threadEnvironment.startRealtimeVoice, { + reportFailure: false, + }); + const stopRemoteVoice = useAtomCommand(threadEnvironment.stopRealtimeVoice, { + reportFailure: false, + }); + const localSessionRef = useRef(null); + const remoteStartedRef = useRef(false); + const generationRef = useRef(0); + const [status, setStatus] = useState("idle"); + const [muted, setMuted] = useState(false); + const [error, setError] = useState(null); + + const supported = + typeof window !== "undefined" && + typeof RTCPeerConnection !== "undefined" && + typeof navigator.mediaDevices?.getUserMedia === "function"; + + const clearLocalSession = useCallback(() => { + const session = localSessionRef.current; + localSessionRef.current = null; + disposeLocalSession(session); + }, []); + + const stop = useCallback(async () => { + generationRef.current += 1; + const shouldStopRemote = remoteStartedRef.current || localSessionRef.current !== null; + remoteStartedRef.current = false; + clearLocalSession(); + setStatus("idle"); + setMuted(false); + setError(null); + + if (shouldStopRemote && input.threadId) { + await stopRemoteVoice({ + environmentId: input.environmentId, + input: { threadId: input.threadId }, + }); + } + }, [clearLocalSession, input.environmentId, input.threadId, stopRemoteVoice]); + + const start = useCallback(async () => { + if (!input.enabled || !input.threadId || !supported || localSessionRef.current) return; + const threadId = input.threadId; + + const generation = generationRef.current + 1; + generationRef.current = generation; + setStatus("connecting"); + setMuted(false); + setError(null); + + try { + const microphone = await navigator.mediaDevices.getUserMedia({ + audio: { + autoGainControl: true, + echoCancellation: true, + noiseSuppression: true, + }, + }); + if (generationRef.current !== generation) { + microphone.getTracks().forEach((track) => track.stop()); + return; + } + + const peer = new RTCPeerConnection(); + const events = peer.createDataChannel("oai-events"); + const audio = new Audio(); + audio.autoplay = true; + audio.setAttribute("playsinline", ""); + microphone.getAudioTracks().forEach((track) => peer.addTrack(track, microphone)); + + const session = { peer, microphone, events, audio } satisfies LocalVoiceSession; + localSessionRef.current = session; + peer.ontrack = (event) => { + audio.srcObject = event.streams[0] ?? new MediaStream([event.track]); + void audio.play().catch(() => undefined); + }; + peer.onconnectionstatechange = () => { + if (generationRef.current !== generation) return; + if (peer.connectionState === "connected") { + setStatus("live"); + return; + } + if (peer.connectionState !== "failed") return; + + generationRef.current += 1; + remoteStartedRef.current = false; + clearLocalSession(); + setStatus("error"); + setError("The Codex voice connection was lost."); + void stopRemoteVoice({ + environmentId: input.environmentId, + input: { threadId }, + }); + }; + + const offer = await peer.createOffer(); + await peer.setLocalDescription(offer); + await waitForIceGathering(peer); + const localSdp = peer.localDescription?.sdp; + if (!localSdp) throw new Error("WebRTC did not produce a local session description."); + + const result = await startRemoteVoice({ + environmentId: input.environmentId, + input: { threadId, sdp: localSdp }, + }); + if (generationRef.current !== generation) { + await stopRemoteVoice({ + environmentId: input.environmentId, + input: { threadId }, + }); + return; + } + if (result._tag !== "Success") { + throw new Error("Codex rejected the realtime voice session."); + } + + remoteStartedRef.current = true; + await peer.setRemoteDescription({ type: "answer", sdp: result.value.sdp }); + setStatus("live"); + } catch (cause) { + if (generationRef.current !== generation) return; + generationRef.current += 1; + const shouldStopRemote = remoteStartedRef.current; + remoteStartedRef.current = false; + clearLocalSession(); + setStatus("error"); + setError(voiceErrorMessage(cause)); + if (shouldStopRemote) { + void stopRemoteVoice({ + environmentId: input.environmentId, + input: { threadId }, + }); + } + } + }, [ + clearLocalSession, + input.enabled, + input.environmentId, + input.threadId, + startRemoteVoice, + stopRemoteVoice, + supported, + ]); + + const toggleMuted = useCallback(() => { + const microphoneTrack = localSessionRef.current?.microphone.getAudioTracks()[0]; + if (!microphoneTrack) return; + const nextMuted = microphoneTrack.enabled; + microphoneTrack.enabled = !nextMuted; + setMuted(nextMuted); + }, []); + + useEffect(() => { + if (!input.enabled && (localSessionRef.current || remoteStartedRef.current)) { + void stop(); + } + }, [input.enabled, stop]); + + useEffect( + () => () => { + generationRef.current += 1; + const shouldStopRemote = remoteStartedRef.current; + remoteStartedRef.current = false; + clearLocalSession(); + if (shouldStopRemote && input.threadId) { + void stopRemoteVoice({ + environmentId: input.environmentId, + input: { threadId: input.threadId }, + }); + } + }, + [clearLocalSession, input.environmentId, input.threadId, stopRemoteVoice], + ); + + return { supported, status, muted, error, start, stop, toggleMuted }; +} diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 7d68dc4efe17..44bb60ec785f 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -65,6 +65,21 @@ Provider output comes back as internal commands such as `thread.message.assistan `thread.session.set`, which clients observe through `orchestration.subscribeThread`. See [overview.md](./overview.md) for the command/event loop. +### Codex realtime voice + +Codex voice is deliberately an ephemeral provider operation rather than an orchestration command. +The web or desktop client creates a WebRTC offer with its microphone track and the `oai-events` +data channel, then calls `provider.realtimeVoice.start`. `ProviderService` routes that offer to the +active Codex adapter, recovering the bound session when needed. The Codex session runtime starts +`thread/realtime/start` with protocol v3 and returns the SDP notification as the RPC answer. Stop is +the matching `provider.realtimeVoice.stop` / `thread/realtime/stop` pair. + +Only SDP signaling crosses the T3 WebSocket. Realtime audio goes directly between the client and +OpenAI over WebRTC; it is not proxied or persisted by the T3 server. Closing the client control, +changing threads or providers, losing the peer connection, or closing the provider runtime all +stop the ephemeral session. Other provider adapters leave the optional realtime operations +unsupported, and the native mobile clients do not expose the control. + ## Server-side workers Provider work flows through three queue-backed workers. All three are built with diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index f696e8877b2e..a435fe6b25a2 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -34,6 +34,22 @@ In an existing Codex thread, send `/feedback` or `/feedback` followed by a descr issue. T3 Code uploads the thread and Codex logs to OpenAI and shows a thread ID that you can copy and share with OpenAI employees. +## Talk to Codex + +Codex threads have a microphone button beside the send button in the web and desktop apps. Select +it, allow microphone access when prompted, and start talking. The compact voice control shows when +the connection is live and lets you mute your microphone or end the conversation. + +Voice uses Codex's native realtime session for the current thread, so the conversation stays with +the same Codex account and workspace as the text chat. It requires a Codex version that supports +realtime voice, a working microphone, and WebRTC access from the client. If the microphone button +turns red, hover it for the connection or permission error and select it to retry. + +Browsers only expose microphones to secure pages, so remote web connections need HTTPS. Local +`localhost` pages and the desktop app are treated as secure contexts. + +The native iOS and Android apps do not show this control yet. + ## Approve access to other apps When a Codex tool needs access to an app such as Safari, T3 Code shows the app name and asks for diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index c540644289df..772091b8a4f6 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -210,5 +210,17 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + startRealtimeVoice: createEnvironmentRpcCommand(runtime, { + label: "environment-data:commands:thread:start-realtime-voice", + tag: WS_METHODS.providerRealtimeVoiceStart, + scheduler, + concurrency, + }), + stopRealtimeVoice: createEnvironmentRpcCommand(runtime, { + label: "environment-data:commands:thread:stop-realtime-voice", + tag: WS_METHODS.providerRealtimeVoiceStop, + scheduler, + concurrency, + }), }; } diff --git a/packages/contracts/src/provider.test.ts b/packages/contracts/src/provider.test.ts index ba7ca63745b6..e03750e22de5 100644 --- a/packages/contracts/src/provider.test.ts +++ b/packages/contracts/src/provider.test.ts @@ -4,6 +4,9 @@ import * as Schema from "effect/Schema"; import { ThreadId } from "./baseSchemas.ts"; import { ProviderEvent, + ProviderRealtimeVoiceError, + ProviderRealtimeVoiceStartInput, + ProviderRealtimeVoiceStartResult, ProviderSendTurnInput, ProviderSession, ProviderSessionStartInput, @@ -18,6 +21,12 @@ const decodeProviderSession = Schema.decodeUnknownSync(ProviderSession); const decodeProviderEvent = Schema.decodeUnknownSync(ProviderEvent); const decodeProviderUploadFeedbackInput = Schema.decodeUnknownSync(ProviderUploadFeedbackInput); const decodeProviderUploadFeedbackResult = Schema.decodeUnknownSync(ProviderUploadFeedbackResult); +const decodeProviderRealtimeVoiceStartInput = Schema.decodeUnknownSync( + ProviderRealtimeVoiceStartInput, +); +const decodeProviderRealtimeVoiceStartResult = Schema.decodeUnknownSync( + ProviderRealtimeVoiceStartResult, +); function getOptionValue( options: ReadonlyArray<{ id: string; value: unknown }> | undefined, @@ -192,6 +201,36 @@ describe("provider feedback", () => { }); }); +describe("provider realtime voice", () => { + it("accepts bounded WebRTC offer and answer descriptions", () => { + expect( + decodeProviderRealtimeVoiceStartInput({ threadId: "thread-1", sdp: "offer-sdp" }), + ).toEqual({ threadId: "thread-1", sdp: "offer-sdp" }); + expect(decodeProviderRealtimeVoiceStartResult({ sdp: "answer-sdp" })).toEqual({ + sdp: "answer-sdp", + }); + }); + + it("rejects empty session descriptions", () => { + expect(() => + decodeProviderRealtimeVoiceStartInput({ threadId: "thread-1", sdp: "" }), + ).toThrow(); + expect(() => decodeProviderRealtimeVoiceStartResult({ sdp: "" })).toThrow(); + }); + + it("does not expose upstream request details in its message", () => { + const cause = new Error("provider request secret"); + const error = new ProviderRealtimeVoiceError({ + threadId: ThreadId.make("thread-1"), + operation: "start", + cause, + }); + + expect(error.message).toBe("Failed to start realtime voice for thread thread-1."); + expect(error.message).not.toContain("provider request secret"); + }); +}); + describe("providerInstanceId routing key (slice-2 invariant)", () => { it("decodes a ProviderSessionStartInput without providerInstanceId (legacy producer)", () => { const parsed = decodeProviderSessionStartInput({ diff --git a/packages/contracts/src/provider.ts b/packages/contracts/src/provider.ts index 42a943923037..e96d4d6c7a58 100644 --- a/packages/contracts/src/provider.ts +++ b/packages/contracts/src/provider.ts @@ -121,6 +121,40 @@ export const ProviderUploadFeedbackResult = Schema.Struct({ }); export type ProviderUploadFeedbackResult = typeof ProviderUploadFeedbackResult.Type; +const RealtimeSessionDescription = Schema.String.check( + Schema.isMinLength(1), + Schema.isMaxLength(256 * 1024), +); + +export const ProviderRealtimeVoiceStartInput = Schema.Struct({ + threadId: ThreadId, + sdp: RealtimeSessionDescription, +}); +export type ProviderRealtimeVoiceStartInput = typeof ProviderRealtimeVoiceStartInput.Type; + +export const ProviderRealtimeVoiceStartResult = Schema.Struct({ + sdp: RealtimeSessionDescription, +}); +export type ProviderRealtimeVoiceStartResult = typeof ProviderRealtimeVoiceStartResult.Type; + +export const ProviderRealtimeVoiceStopInput = Schema.Struct({ + threadId: ThreadId, +}); +export type ProviderRealtimeVoiceStopInput = typeof ProviderRealtimeVoiceStopInput.Type; + +export class ProviderRealtimeVoiceError extends Schema.TaggedErrorClass()( + "ProviderRealtimeVoiceError", + { + threadId: ThreadId, + operation: Schema.Literals(["start", "stop"]), + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + return `Failed to ${this.operation} realtime voice for thread ${this.threadId}.`; + } +} + export class ProviderUploadFeedbackError extends Schema.TaggedErrorClass()( "ProviderUploadFeedbackError", { diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 14363cfedff9..25b62fdf7093 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -75,6 +75,10 @@ import { OrchestrationGetWorkflowScriptError, } from "./orchestration.ts"; import { + ProviderRealtimeVoiceError, + ProviderRealtimeVoiceStartInput, + ProviderRealtimeVoiceStartResult, + ProviderRealtimeVoiceStopInput, ProviderUploadFeedbackError, ProviderUploadFeedbackInput, ProviderUploadFeedbackResult, @@ -228,6 +232,8 @@ export const WS_METHODS = { // Provider methods providerUploadFeedback: "provider.uploadFeedback", + providerRealtimeVoiceStart: "provider.realtimeVoice.start", + providerRealtimeVoiceStop: "provider.realtimeVoice.stop", // VCS methods vcsPull: "vcs.pull", @@ -700,6 +706,17 @@ export const WsProviderUploadFeedbackRpc = Rpc.make(WS_METHODS.providerUploadFee error: Schema.Union([ProviderUploadFeedbackError, EnvironmentAuthorizationError]), }); +export const WsProviderRealtimeVoiceStartRpc = Rpc.make(WS_METHODS.providerRealtimeVoiceStart, { + payload: ProviderRealtimeVoiceStartInput, + success: ProviderRealtimeVoiceStartResult, + error: Schema.Union([ProviderRealtimeVoiceError, EnvironmentAuthorizationError]), +}); + +export const WsProviderRealtimeVoiceStopRpc = Rpc.make(WS_METHODS.providerRealtimeVoiceStop, { + payload: ProviderRealtimeVoiceStopInput, + error: Schema.Union([ProviderRealtimeVoiceError, EnvironmentAuthorizationError]), +}); + export const WsSubscribeVcsStatusRpc = Rpc.make(WS_METHODS.subscribeVcsStatus, { payload: VcsStatusInput, success: VcsStatusStreamEvent, @@ -1072,6 +1089,8 @@ export const WsRpcGroup = RpcGroup.make( WsAttachmentsCreateUploadUrlRpc, WsAttachmentsDeleteRpc, WsProviderUploadFeedbackRpc, + WsProviderRealtimeVoiceStartRpc, + WsProviderRealtimeVoiceStopRpc, WsSubscribeVcsStatusRpc, WsVcsPullRpc, WsVcsRefreshStatusRpc, From 0419745820bca26ee7cbf3488dadf7d93bb76fce Mon Sep 17 00:00:00 2001 From: Peng_Lx Date: Thu, 27 Aug 2026 06:09:04 +0800 Subject: [PATCH 2/8] fix(web): preserve Codex voice icon colors --- apps/web/src/components/chat/ComposerVoiceControl.test.tsx | 2 ++ apps/web/src/components/chat/ComposerVoiceControl.tsx | 6 +++--- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/apps/web/src/components/chat/ComposerVoiceControl.test.tsx b/apps/web/src/components/chat/ComposerVoiceControl.test.tsx index ba6f2c65b9eb..7f02c8df111d 100644 --- a/apps/web/src/components/chat/ComposerVoiceControl.test.tsx +++ b/apps/web/src/components/chat/ComposerVoiceControl.test.tsx @@ -35,6 +35,7 @@ describe("ComposerVoiceControl", () => { expect(markup).toContain('aria-label="Talk to Codex"'); expect(markup).toContain('data-codex-voice-state="idle"'); + expect(markup).toContain("[--control-icon-color:currentColor]"); }); it("shows mute and end actions while voice is live", () => { @@ -45,6 +46,7 @@ describe("ComposerVoiceControl", () => { expect(markup).toContain("Voice live"); expect(markup).toContain('aria-label="Mute microphone"'); expect(markup).toContain('aria-label="End Codex voice"'); + expect(markup.match(/\[--control-icon-color:currentColor\]/g)).toHaveLength(2); }); it("renders the muted state without hiding the end action", () => { diff --git a/apps/web/src/components/chat/ComposerVoiceControl.tsx b/apps/web/src/components/chat/ComposerVoiceControl.tsx index 9cb3d0cd828e..ac0605b48818 100644 --- a/apps/web/src/components/chat/ComposerVoiceControl.tsx +++ b/apps/web/src/components/chat/ComposerVoiceControl.tsx @@ -40,7 +40,7 @@ export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { size="icon-sm" variant="ghost" className={cn( - "rounded-full text-secondary-label hover:text-foreground", + "rounded-full text-secondary-label [--control-icon-color:currentColor] hover:text-foreground", voice.status === "error" && "text-destructive hover:text-destructive", )} disabled={props.disabled || !voice.supported} @@ -75,7 +75,7 @@ export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { type="button" size="icon-xs" variant="ghost" - className="rounded-full text-secondary-label hover:text-foreground" + className="rounded-full text-secondary-label [--control-icon-color:currentColor] hover:text-foreground" disabled={voice.status === "connecting"} onPointerDown={preserveComposerFocus} onClick={voice.toggleMuted} @@ -89,7 +89,7 @@ export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { type="button" size="icon-xs" variant="ghost" - className="rounded-full text-destructive hover:bg-destructive/10 hover:text-destructive" + className="rounded-full text-destructive [--control-icon-color:currentColor] hover:bg-destructive/10 hover:text-destructive" onPointerDown={preserveComposerFocus} onClick={() => void voice.stop()} aria-label="End Codex voice" From f6c2cad029ec78c9c1f08a08b279cf3284a67604 Mon Sep 17 00:00:00 2001 From: Peng_Lx Date: Thu, 27 Aug 2026 06:31:50 +0800 Subject: [PATCH 3/8] fix(codex): harden realtime voice lifecycle --- .../CodexRealtimeVoice.integration.test.ts | 186 +++++++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 215 +++++++++++++++--- .../testFixtures/codexCollabMockPeer.mjs | 52 +++++ apps/server/src/ws.ts | 18 +- apps/web/src/components/chat/ChatComposer.tsx | 4 +- .../chat/ComposerVoiceControl.test.tsx | 1 + .../components/chat/ComposerVoiceControl.tsx | 2 +- .../src/hooks/useCodexRealtimeVoice.test.ts | 66 ++++++ apps/web/src/hooks/useCodexRealtimeVoice.ts | 54 +++-- .../src/state/threadCommands.ts | 4 +- packages/contracts/src/provider.test.ts | 12 +- packages/contracts/src/provider.ts | 1 - 12 files changed, 553 insertions(+), 62 deletions(-) create mode 100644 apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts create mode 100644 apps/web/src/hooks/useCodexRealtimeVoice.test.ts diff --git a/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts b/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts new file mode 100644 index 000000000000..8d59c045c5a4 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts @@ -0,0 +1,186 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFS from "node:fs"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import { assert, describe } from "vite-plus/test"; + +import wireFixture from "../testFixtures/codexMultiAgentWire.json" with { type: "json" }; +import { + CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError, + CodexSessionRuntimeRealtimeVoiceNegotiationError, + CodexSessionRuntimeRealtimeVoiceStoppedError, + makeCodexSessionRuntime, + type CodexSessionRuntimeError, +} from "./CodexSessionRuntime.ts"; + +const peerPath = NodePath.join(import.meta.dirname, "../testFixtures/codexCollabMockPeer.sh"); + +function runtimeTest( + name: string, + script: Record, + run: ( + runtime: Effect.Success>, + scriptPath: string, + ) => Effect.Effect, +) { + it.live(name, () => + Effect.gen(function* () { + const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-codex-voice-")); + const scriptPath = NodePath.join(directory, "script.json"); + NodeFS.writeFileSync( + scriptPath, + // @effect-diagnostics-next-line preferSchemaOverJson:off + JSON.stringify({ rootThreadId: wireFixture.rootThreadId, notifications: [], ...script }), + "utf8", + ); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(directory, { force: true, recursive: true })), + ); + + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make(`thread-${name}`), + binaryPath: peerPath, + cwd: "/tmp", + runtimeMode: "full-access", + environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath }, + realtimeVoiceNegotiationTimeoutMs: 500, + realtimeVoiceStopTimeoutMs: 200, + }); + yield* Effect.addFinalizer(() => runtime.close); + yield* runtime.start(); + yield* run(runtime, scriptPath); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +} + +describe("CodexSessionRuntime realtime voice", () => { + runtimeTest( + "stops an in-flight negotiation and permits an immediate restart", + { + realtimeStarts: [{ started: true }, { sdp: "answer-sdp" }], + }, + (runtime) => + Effect.gen(function* () { + const started = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const firstStart = yield* runtime.startRealtimeVoice("first-offer").pipe(Effect.forkScoped); + yield* Fiber.join(started); + + yield* runtime.stopRealtimeVoice; + const stoppedExit = yield* Fiber.await(firstStart); + assert.isTrue(Exit.isFailure(stoppedExit)); + if (Exit.isFailure(stoppedExit)) { + assert.instanceOf( + Cause.squash(stoppedExit.cause), + CodexSessionRuntimeRealtimeVoiceStoppedError, + ); + } + + const answer = yield* runtime.startRealtimeVoice("second-offer"); + assert.equal(answer, "answer-sdp"); + }), + ); + + runtimeTest( + "stops the provider when an in-flight start is interrupted", + { realtimeStarts: [{ hangRequest: true, started: true }] }, + (runtime, scriptPath) => + Effect.gen(function* () { + const started = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const start = yield* runtime.startRealtimeVoice("offer-sdp").pipe(Effect.forkScoped); + yield* Fiber.join(started); + + yield* Fiber.interrupt(start); + + const stops = NodeFS.readFileSync(`${scriptPath}.realtime-stops`, "utf8"); + assert.include(stops, wireFixture.rootThreadId); + }), + ); + + runtimeTest( + "fails immediately when Codex rejects negotiation", + { realtimeStarts: [{ error: "mock negotiation rejected" }] }, + (runtime) => + Effect.gen(function* () { + const exit = yield* Effect.exit(runtime.startRealtimeVoice("offer-sdp")); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause); + assert.instanceOf(error, CodexSessionRuntimeRealtimeVoiceNegotiationError); + assert.include((error as Error).message, "mock negotiation rejected"); + } + }), + ); + + runtimeTest( + "times out a hung start request and clears the negotiation slot", + { realtimeStarts: [{ hangRequest: true }, { sdp: "retry-answer-sdp" }] }, + (runtime) => + Effect.gen(function* () { + const exit = yield* Effect.exit(runtime.startRealtimeVoice("hung-offer")); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + assert.instanceOf( + Cause.squash(exit.cause), + CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError, + ); + } + + const answer = yield* runtime.startRealtimeVoice("retry-offer"); + assert.equal(answer, "retry-answer-sdp"); + }), + ); + + runtimeTest( + "bounds realtime cleanup when the stop request hangs", + { realtimeStarts: [{ sdp: "answer-sdp" }], hangRealtimeStop: true }, + (runtime) => + Effect.gen(function* () { + yield* runtime.startRealtimeVoice("offer-sdp"); + yield* runtime.close; + }), + ); + + runtimeTest( + "closes a pending start even when the stop request hangs", + { realtimeStarts: [{ hangRequest: true, started: true }], hangRealtimeStop: true }, + (runtime) => + Effect.gen(function* () { + const started = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const start = yield* runtime.startRealtimeVoice("offer-sdp").pipe(Effect.forkScoped); + yield* Fiber.join(started); + + yield* runtime.close; + + const exit = yield* Fiber.await(start); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + assert.instanceOf(Cause.squash(exit.cause), CodexSessionRuntimeRealtimeVoiceStoppedError); + } + }), + ); +}); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 3f9d2f328906..ecbec4327a21 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -161,6 +161,8 @@ export interface CodexSessionRuntimeOptions { readonly serviceTier?: CodexServiceTier | undefined; readonly resumeCursor?: CodexResumeCursor; readonly appServerArgs?: ReadonlyArray; + readonly realtimeVoiceNegotiationTimeoutMs?: number; + readonly realtimeVoiceStopTimeoutMs?: number; } export interface CodexSessionRuntimeSendTurnInput { @@ -227,7 +229,12 @@ export type CodexSessionRuntimeError = | CodexSessionRuntimePendingApprovalNotFoundError | CodexSessionRuntimePendingUserInputNotFoundError | CodexSessionRuntimeInvalidUserInputAnswersError - | CodexSessionRuntimeThreadIdMissingError; + | CodexSessionRuntimeThreadIdMissingError + | CodexSessionRuntimeRealtimeVoiceAlreadyStartingError + | CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError + | CodexSessionRuntimeRealtimeVoiceNegotiationError + | CodexSessionRuntimeRealtimeVoiceStoppedError + | CodexSessionRuntimeRealtimeVoiceStopTimeoutError; export class CodexSessionRuntimePendingApprovalNotFoundError extends Schema.TaggedErrorClass()( "CodexSessionRuntimePendingApprovalNotFoundError", @@ -273,6 +280,56 @@ export class CodexSessionRuntimeThreadIdMissingError extends Schema.TaggedErrorC } } +export class CodexSessionRuntimeRealtimeVoiceAlreadyStartingError extends Schema.TaggedErrorClass()( + "CodexSessionRuntimeRealtimeVoiceAlreadyStartingError", + { providerThreadId: Schema.String }, +) { + override get message(): string { + return `Codex realtime voice is already starting for ${this.providerThreadId}`; + } +} + +export class CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError extends Schema.TaggedErrorClass()( + "CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError", + { providerThreadId: Schema.String, timeoutMs: Schema.Number }, +) { + override get message(): string { + return `Codex realtime voice did not return an SDP answer within ${this.timeoutMs}ms for ${this.providerThreadId}`; + } +} + +export class CodexSessionRuntimeRealtimeVoiceNegotiationError extends Schema.TaggedErrorClass()( + "CodexSessionRuntimeRealtimeVoiceNegotiationError", + { providerThreadId: Schema.String, detail: Schema.String }, +) { + override get message(): string { + return `Codex realtime voice negotiation failed for ${this.providerThreadId}: ${this.detail}`; + } +} + +export class CodexSessionRuntimeRealtimeVoiceStoppedError extends Schema.TaggedErrorClass()( + "CodexSessionRuntimeRealtimeVoiceStoppedError", + { providerThreadId: Schema.String }, +) { + override get message(): string { + return `Codex realtime voice negotiation was stopped for ${this.providerThreadId}`; + } +} + +export class CodexSessionRuntimeRealtimeVoiceStopTimeoutError extends Schema.TaggedErrorClass()( + "CodexSessionRuntimeRealtimeVoiceStopTimeoutError", + { providerThreadId: Schema.String, timeoutMs: Schema.Number }, +) { + override get message(): string { + return `Codex realtime voice did not stop within ${this.timeoutMs}ms for ${this.providerThreadId}`; + } +} + +interface PendingRealtimeVoice { + readonly providerThreadId: string; + readonly answer: Deferred.Deferred; +} + interface PendingApproval { readonly requestId: ApprovalRequestId; readonly jsonRpcId: string; @@ -1142,7 +1199,48 @@ export const makeCodexSessionRuntime = ( const collabChildLiveTurnsRef = yield* Ref.make(new Map()); const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter(); const closedRef = yield* Ref.make(false); - const pendingRealtimeSdpRef = yield* Ref.make | null>(null); + const pendingRealtimeVoiceRef = yield* Ref.make(null); + const realtimeVoiceNegotiationTimeoutMs = Math.max( + 1, + options.realtimeVoiceNegotiationTimeoutMs ?? 20_000, + ); + const realtimeVoiceStopTimeoutMs = Math.max(1, options.realtimeVoiceStopTimeoutMs ?? 3_000); + + const clearPendingRealtimeVoice = (pending: PendingRealtimeVoice) => + Ref.update(pendingRealtimeVoiceRef, (current) => + current?.answer === pending.answer ? null : current, + ); + + const failPendingRealtimeVoice = (providerThreadId: string, error: CodexSessionRuntimeError) => + Ref.get(pendingRealtimeVoiceRef).pipe( + Effect.flatMap((pending) => + pending?.providerThreadId === providerThreadId + ? Deferred.fail(pending.answer, error).pipe(Effect.asVoid) + : Effect.void, + ), + ); + + const cancelPendingRealtimeVoice = ( + providerThreadId: string, + error: CodexSessionRuntimeError, + ) => + Ref.modify(pendingRealtimeVoiceRef, (current) => + current?.providerThreadId === providerThreadId + ? ([current, null] as const) + : ([null, current] as const), + ).pipe( + Effect.flatMap((pending) => + pending === null ? Effect.void : Deferred.fail(pending.answer, error).pipe(Effect.asVoid), + ), + ); + + const requestRealtimeVoiceStop = (providerThreadId: string) => + client.raw + .request("thread/realtime/stop", { threadId: providerThreadId }) + .pipe(Effect.timeoutOption(`${realtimeVoiceStopTimeoutMs} millis`)); + + const requestRealtimeVoiceStopBestEffort = (providerThreadId: string) => + requestRealtimeVoiceStop(providerThreadId).pipe(Effect.ignore); // `~` is not shell-expanded when env vars are set via // `child_process.spawn`; `expandHomePath` lets a configured @@ -1731,17 +1829,51 @@ export const makeCodexSessionRuntime = ( if (providerThreadId && payload.threadId !== providerThreadId) { return Effect.void; } - return Ref.get(pendingRealtimeSdpRef).pipe( + return Ref.get(pendingRealtimeVoiceRef).pipe( Effect.flatMap((pending) => - pending === null + pending?.providerThreadId !== payload.threadId ? Effect.void - : Deferred.succeed(pending, payload.sdp).pipe(Effect.asVoid), + : Deferred.succeed(pending.answer, payload.sdp).pipe(Effect.asVoid), ), ); }), ), ); + yield* client.handleServerNotification("thread/realtime/error", (payload) => + currentSessionProviderThreadId.pipe( + Effect.flatMap((providerThreadId) => { + if (providerThreadId && payload.threadId !== providerThreadId) { + return Effect.void; + } + return failPendingRealtimeVoice( + payload.threadId, + new CodexSessionRuntimeRealtimeVoiceNegotiationError({ + providerThreadId: payload.threadId, + detail: payload.message, + }), + ); + }), + ), + ); + + yield* client.handleServerNotification("thread/realtime/closed", (payload) => + currentSessionProviderThreadId.pipe( + Effect.flatMap((providerThreadId) => { + if (providerThreadId && payload.threadId !== providerThreadId) { + return Effect.void; + } + return failPendingRealtimeVoice( + payload.threadId, + new CodexSessionRuntimeRealtimeVoiceNegotiationError({ + providerThreadId: payload.threadId, + detail: payload.reason ?? "Realtime transport closed during negotiation.", + }), + ); + }), + ), + ); + yield* client.handleServerRequest("item/commandExecution/requestApproval", (payload) => Effect.gen(function* () { const requestId = ApprovalRequestId.make(yield* randomUUIDv4("command-approval-request")); @@ -2105,9 +2237,11 @@ export const makeCodexSessionRuntime = ( yield* settlePendingUserInputs({}); const providerThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); if (providerThreadId) { - yield* client.raw - .request("thread/realtime/stop", { threadId: providerThreadId }) - .pipe(Effect.ignore); + yield* cancelPendingRealtimeVoice( + providerThreadId, + new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), + ); + yield* requestRealtimeVoiceStopBestEffort(providerThreadId); } yield* updateSession(sessionRef, { status: "closed", @@ -2219,38 +2353,55 @@ export const makeCodexSessionRuntime = ( startRealtimeVoice: (sdp) => Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; - const pending = yield* Deferred.make(); - const installed = yield* Ref.modify(pendingRealtimeSdpRef, (current) => + const answer = yield* Deferred.make(); + const pending = { providerThreadId, answer } satisfies PendingRealtimeVoice; + const installed = yield* Ref.modify(pendingRealtimeVoiceRef, (current) => current === null ? ([true, pending] as const) : ([false, current] as const), ); if (!installed) { - return yield* CodexErrors.CodexAppServerRequestError.invalidRequest( - "A Codex realtime voice session is already connecting.", - { method: "thread/realtime/start" }, - ); + return yield* new CodexSessionRuntimeRealtimeVoiceAlreadyStartingError({ + providerThreadId, + }); } - return yield* Effect.gen(function* () { - yield* client.raw.request( - "thread/realtime/start", - buildCodexRealtimeStartParams(providerThreadId, sdp), - ); - const answer = yield* Deferred.await(pending).pipe(Effect.timeoutOption("20 seconds")); - if (Option.isNone(answer)) { - yield* client.raw - .request("thread/realtime/stop", { threadId: providerThreadId }) - .pipe(Effect.ignore); - return yield* CodexErrors.CodexAppServerRequestError.invalidRequest( - "Codex realtime voice did not return an SDP answer.", - { method: "thread/realtime/start" }, - ); - } - return answer.value; - }).pipe(Effect.ensuring(Ref.set(pendingRealtimeSdpRef, null))); + const result = yield* Effect.raceFirst( + client.raw + .request( + "thread/realtime/start", + buildCodexRealtimeStartParams(providerThreadId, sdp), + ) + .pipe(Effect.andThen(Effect.never)), + Deferred.await(answer), + ).pipe( + Effect.timeoutOption(`${realtimeVoiceNegotiationTimeoutMs} millis`), + Effect.onExit((exit) => + Exit.isSuccess(exit) && Option.isSome(exit.value) + ? Effect.void + : requestRealtimeVoiceStopBestEffort(providerThreadId), + ), + Effect.ensuring(clearPendingRealtimeVoice(pending)), + ); + if (Option.isNone(result)) { + return yield* new CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError({ + providerThreadId, + timeoutMs: realtimeVoiceNegotiationTimeoutMs, + }); + } + return result.value; }), stopRealtimeVoice: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; - yield* client.raw.request("thread/realtime/stop", { threadId: providerThreadId }); + yield* cancelPendingRealtimeVoice( + providerThreadId, + new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), + ); + const result = yield* requestRealtimeVoiceStop(providerThreadId); + if (Option.isNone(result)) { + return yield* new CodexSessionRuntimeRealtimeVoiceStopTimeoutError({ + providerThreadId, + timeoutMs: realtimeVoiceStopTimeoutMs, + }); + } }), readThread: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; diff --git a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs index d3bdee367125..13489f82e7fe 100644 --- a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs +++ b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs @@ -17,6 +17,7 @@ const script = JSON.parse(NodeFS.readFileSync(process.env.T3_CODEX_COLLAB_SCRIPT const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`); let turnStartCount = 0; +let realtimeStartCount = 0; let activeTurn; const rl = NodeReadline.createInterface({ input: process.stdin }); @@ -130,6 +131,57 @@ rl.on("line", (line) => { write({ id, result: {} }); return; } + if (method === "thread/realtime/start") { + const realtimeStart = script.realtimeStarts?.[realtimeStartCount] ?? {}; + realtimeStartCount += 1; + NodeFS.appendFileSync( + `${process.env.T3_CODEX_COLLAB_SCRIPT}.realtime-starts`, + `${JSON.stringify(message.params)}\n`, + ); + if (realtimeStart.hangRequest !== true) { + write({ id, result: {} }); + } + const threadId = message.params?.threadId ?? script.rootThreadId; + if (realtimeStart.started === true) { + write({ + jsonrpc: "2.0", + method: "thread/realtime/started", + params: { threadId, version: "v3", realtimeSessionId: `mock-${realtimeStartCount}` }, + }); + } + if (typeof realtimeStart.error === "string") { + write({ + jsonrpc: "2.0", + method: "thread/realtime/error", + params: { threadId, message: realtimeStart.error }, + }); + } + if (typeof realtimeStart.closed === "string") { + write({ + jsonrpc: "2.0", + method: "thread/realtime/closed", + params: { threadId, reason: realtimeStart.closed }, + }); + } + if (typeof realtimeStart.sdp === "string") { + write({ + jsonrpc: "2.0", + method: "thread/realtime/sdp", + params: { threadId, sdp: realtimeStart.sdp }, + }); + } + return; + } + if (method === "thread/realtime/stop") { + NodeFS.appendFileSync( + `${process.env.T3_CODEX_COLLAB_SCRIPT}.realtime-stops`, + `${JSON.stringify(message.params)}\n`, + ); + if (script.hangRealtimeStop !== true) { + write({ id, result: {} }); + } + return; + } if (id !== undefined) { write({ id, result: {} }); } diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index c9cbf902bec7..47a3cc5a23b0 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1608,12 +1608,17 @@ const makeWsRpcLayer = ( observeRpcEffect( WS_METHODS.providerRealtimeVoiceStart, providerService.startRealtimeVoice(input).pipe( + Effect.tapError((cause) => + Effect.logError("Failed to start provider realtime voice.", { + threadId: input.threadId, + cause, + }), + ), Effect.mapError( - (cause) => + () => new ProviderRealtimeVoiceError({ threadId: input.threadId, operation: "start", - cause, }), ), ), @@ -1623,12 +1628,17 @@ const makeWsRpcLayer = ( observeRpcEffect( WS_METHODS.providerRealtimeVoiceStop, providerService.stopRealtimeVoice(input).pipe( + Effect.tapError((cause) => + Effect.logError("Failed to stop provider realtime voice.", { + threadId: input.threadId, + cause, + }), + ), Effect.mapError( - (cause) => + () => new ProviderRealtimeVoiceError({ threadId: input.threadId, operation: "stop", - cause, }), ), ), diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index fe774cfe3364..32698ca914b8 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -3565,7 +3565,9 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) > {showMobilePendingAnswerActions ? null : inlineTasksBadge} {showMobilePendingAnswerActions ? null : inlineStashBadge} - {selectedProvider === ProviderDriverKind.make("codex") && activeThreadId ? ( + {routeKind === "server" && + selectedProvider === ProviderDriverKind.make("codex") && + activeThreadId ? ( { expect(markup).toContain('aria-label="Mute microphone"'); expect(markup).toContain('aria-label="End Codex voice"'); expect(markup.match(/\[--control-icon-color:currentColor\]/g)).toHaveLength(2); + expect(markup).toContain("bg-success"); }); it("renders the muted state without hiding the end action", () => { diff --git a/apps/web/src/components/chat/ComposerVoiceControl.tsx b/apps/web/src/components/chat/ComposerVoiceControl.tsx index ac0605b48818..512c59dd8148 100644 --- a/apps/web/src/components/chat/ComposerVoiceControl.tsx +++ b/apps/web/src/components/chat/ComposerVoiceControl.tsx @@ -66,7 +66,7 @@ export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { aria-hidden="true" className={cn( "me-1 size-1.5 rounded-full", - voice.status === "connecting" ? "bg-muted-foreground" : "bg-emerald-500", + voice.status === "connecting" ? "bg-muted-foreground" : "bg-success", )} /> {label} diff --git a/apps/web/src/hooks/useCodexRealtimeVoice.test.ts b/apps/web/src/hooks/useCodexRealtimeVoice.test.ts new file mode 100644 index 000000000000..217ff68e49d2 --- /dev/null +++ b/apps/web/src/hooks/useCodexRealtimeVoice.test.ts @@ -0,0 +1,66 @@ +import { afterEach, describe, expect, it, vi } from "vite-plus/test"; + +import { waitForIceGathering } from "./useCodexRealtimeVoice"; + +class FakePeerConnection { + iceGatheringState: RTCIceGatheringState = "gathering"; + readonly listeners = new Set(); + + addEventListener(_type: string, listener: EventListenerOrEventListenerObject) { + this.listeners.add(listener); + } + + removeEventListener(_type: string, listener: EventListenerOrEventListenerObject) { + this.listeners.delete(listener); + } + + complete() { + this.iceGatheringState = "complete"; + const event = new Event("icegatheringstatechange"); + for (const listener of this.listeners) { + if (typeof listener === "function") listener(event); + else listener.handleEvent(event); + } + } +} + +const asPeerConnection = (peer: FakePeerConnection) => peer as unknown as RTCPeerConnection; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("waitForIceGathering", () => { + it("resolves only after ICE gathering completes", async () => { + const peer = new FakePeerConnection(); + const result = waitForIceGathering(asPeerConnection(peer), new AbortController().signal); + + peer.complete(); + + await expect(result).resolves.toBeUndefined(); + expect(peer.listeners).toHaveLength(0); + }); + + it("rejects when the session is cancelled", async () => { + const peer = new FakePeerConnection(); + const abort = new AbortController(); + const result = waitForIceGathering(asPeerConnection(peer), abort.signal); + + abort.abort(); + + await expect(result).rejects.toMatchObject({ name: "AbortError" }); + expect(peer.listeners).toHaveLength(0); + }); + + it("rejects instead of sending a partial offer after the ICE deadline", async () => { + vi.useFakeTimers(); + const peer = new FakePeerConnection(); + const result = waitForIceGathering(asPeerConnection(peer), new AbortController().signal); + const assertion = expect(result).rejects.toThrow("WebRTC ICE gathering timed out."); + + await vi.advanceTimersByTimeAsync(15_000); + + await assertion; + expect(peer.listeners).toHaveLength(0); + }); +}); diff --git a/apps/web/src/hooks/useCodexRealtimeVoice.ts b/apps/web/src/hooks/useCodexRealtimeVoice.ts index e6480f804a05..3983b4466037 100644 --- a/apps/web/src/hooks/useCodexRealtimeVoice.ts +++ b/apps/web/src/hooks/useCodexRealtimeVoice.ts @@ -1,4 +1,5 @@ import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; import { useCallback, useEffect, useRef, useState } from "react"; import { threadEnvironment } from "../state/threads"; @@ -21,9 +22,10 @@ interface LocalVoiceSession { readonly microphone: MediaStream; readonly events: RTCDataChannel; readonly audio: HTMLAudioElement; + readonly abort: AbortController; } -const ICE_GATHERING_TIMEOUT_MS = 5_000; +const ICE_GATHERING_TIMEOUT_MS = 15_000; const MINIMUM_CODEX_REALTIME_VOICE_VERSION = [0, 145, 0] as const; export function supportsCodexRealtimeVoiceVersion(version: string | null): boolean { @@ -40,22 +42,33 @@ export function supportsCodexRealtimeVoiceVersion(version: string | null): boole return true; } -function waitForIceGathering(peer: RTCPeerConnection): Promise { +export function waitForIceGathering(peer: RTCPeerConnection, signal: AbortSignal): Promise { if (peer.iceGatheringState === "complete") { return Promise.resolve(); } + if (signal.aborted) { + return Promise.reject(signal.reason ?? new DOMException("Aborted", "AbortError")); + } - return new Promise((resolve) => { - const finish = () => { - window.clearTimeout(timeoutId); + return new Promise((resolve, reject) => { + const finish = (cause?: unknown) => { + globalThis.clearTimeout(timeoutId); peer.removeEventListener("icegatheringstatechange", handleStateChange); - resolve(); + signal.removeEventListener("abort", handleAbort); + if (cause === undefined) resolve(); + else reject(cause); }; const handleStateChange = () => { if (peer.iceGatheringState === "complete") finish(); }; - const timeoutId = window.setTimeout(finish, ICE_GATHERING_TIMEOUT_MS); + const handleAbort = () => finish(signal.reason ?? new DOMException("Aborted", "AbortError")); + const timeoutId = globalThis.setTimeout( + () => finish(new Error("WebRTC ICE gathering timed out.")), + ICE_GATHERING_TIMEOUT_MS, + ); peer.addEventListener("icegatheringstatechange", handleStateChange); + signal.addEventListener("abort", handleAbort, { once: true }); + if (signal.aborted) handleAbort(); }); } @@ -69,6 +82,7 @@ function voiceErrorMessage(cause: unknown): string { function disposeLocalSession(session: LocalVoiceSession | null): void { if (!session) return; + session.abort.abort(); session.peer.onconnectionstatechange = null; session.peer.ontrack = null; session.events.close(); @@ -150,11 +164,12 @@ export function useCodexRealtimeVoice(input: { const peer = new RTCPeerConnection(); const events = peer.createDataChannel("oai-events"); const audio = new Audio(); + const abort = new AbortController(); audio.autoplay = true; audio.setAttribute("playsinline", ""); microphone.getAudioTracks().forEach((track) => peer.addTrack(track, microphone)); - const session = { peer, microphone, events, audio } satisfies LocalVoiceSession; + const session = { peer, microphone, events, audio, abort } satisfies LocalVoiceSession; localSessionRef.current = session; peer.ontrack = (event) => { audio.srcObject = event.streams[0] ?? new MediaStream([event.track]); @@ -181,7 +196,7 @@ export function useCodexRealtimeVoice(input: { const offer = await peer.createOffer(); await peer.setLocalDescription(offer); - await waitForIceGathering(peer); + await waitForIceGathering(peer, abort.signal); const localSdp = peer.localDescription?.sdp; if (!localSdp) throw new Error("WebRTC did not produce a local session description."); @@ -197,7 +212,9 @@ export function useCodexRealtimeVoice(input: { return; } if (result._tag !== "Success") { - throw new Error("Codex rejected the realtime voice session."); + throw new Error("Codex rejected the realtime voice session.", { + cause: squashAtomCommandFailure(result), + }); } remoteStartedRef.current = true; @@ -237,15 +254,19 @@ export function useCodexRealtimeVoice(input: { }, []); useEffect(() => { - if (!input.enabled && (localSessionRef.current || remoteStartedRef.current)) { + if (!input.enabled) { void stop(); } }, [input.enabled, stop]); - useEffect( - () => () => { + useEffect(() => { + setStatus("idle"); + setMuted(false); + setError(null); + + return () => { generationRef.current += 1; - const shouldStopRemote = remoteStartedRef.current; + const shouldStopRemote = remoteStartedRef.current || localSessionRef.current !== null; remoteStartedRef.current = false; clearLocalSession(); if (shouldStopRemote && input.threadId) { @@ -254,9 +275,8 @@ export function useCodexRealtimeVoice(input: { input: { threadId: input.threadId }, }); } - }, - [clearLocalSession, input.environmentId, input.threadId, stopRemoteVoice], - ); + }; + }, [clearLocalSession, input.environmentId, input.threadId, stopRemoteVoice]); return { supported, status, muted, error, start, stop, toggleMuted }; } diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 772091b8a4f6..7dcd0bb59df5 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -214,13 +214,13 @@ export function createThreadEnvironmentAtoms( label: "environment-data:commands:thread:start-realtime-voice", tag: WS_METHODS.providerRealtimeVoiceStart, scheduler, - concurrency, + concurrency: { mode: "parallel" }, }), stopRealtimeVoice: createEnvironmentRpcCommand(runtime, { label: "environment-data:commands:thread:stop-realtime-voice", tag: WS_METHODS.providerRealtimeVoiceStop, scheduler, - concurrency, + concurrency: { mode: "parallel" }, }), }; } diff --git a/packages/contracts/src/provider.test.ts b/packages/contracts/src/provider.test.ts index e03750e22de5..aac6349e8be1 100644 --- a/packages/contracts/src/provider.test.ts +++ b/packages/contracts/src/provider.test.ts @@ -27,6 +27,7 @@ const decodeProviderRealtimeVoiceStartInput = Schema.decodeUnknownSync( const decodeProviderRealtimeVoiceStartResult = Schema.decodeUnknownSync( ProviderRealtimeVoiceStartResult, ); +const encodeProviderRealtimeVoiceError = Schema.encodeUnknownSync(ProviderRealtimeVoiceError); function getOptionValue( options: ReadonlyArray<{ id: string; value: unknown }> | undefined, @@ -218,16 +219,19 @@ describe("provider realtime voice", () => { expect(() => decodeProviderRealtimeVoiceStartResult({ sdp: "" })).toThrow(); }); - it("does not expose upstream request details in its message", () => { - const cause = new Error("provider request secret"); + it("does not encode upstream request details on the public RPC error", () => { const error = new ProviderRealtimeVoiceError({ threadId: ThreadId.make("thread-1"), operation: "start", - cause, }); + const encoded = encodeProviderRealtimeVoiceError(error); expect(error.message).toBe("Failed to start realtime voice for thread thread-1."); - expect(error.message).not.toContain("provider request secret"); + expect(encoded).toEqual({ + _tag: "ProviderRealtimeVoiceError", + threadId: "thread-1", + operation: "start", + }); }); }); diff --git a/packages/contracts/src/provider.ts b/packages/contracts/src/provider.ts index e96d4d6c7a58..01962813cdbe 100644 --- a/packages/contracts/src/provider.ts +++ b/packages/contracts/src/provider.ts @@ -147,7 +147,6 @@ export class ProviderRealtimeVoiceError extends Schema.TaggedErrorClass Date: Thu, 27 Aug 2026 06:48:35 +0800 Subject: [PATCH 4/8] fix(codex): close realtime voice lifecycle gaps --- .../CodexRealtimeVoice.integration.test.ts | 46 +++- .../provider/Layers/CodexSessionRuntime.ts | 246 ++++++++++++++---- .../testFixtures/codexCollabMockPeer.mjs | 21 ++ apps/web/src/hooks/useCodexRealtimeVoice.ts | 48 +++- 4 files changed, 301 insertions(+), 60 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts b/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts index 8d59c045c5a4..40f6296cae38 100644 --- a/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts @@ -19,6 +19,7 @@ import { CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError, CodexSessionRuntimeRealtimeVoiceNegotiationError, CodexSessionRuntimeRealtimeVoiceStoppedError, + CodexSessionRuntimeRealtimeVoiceStopTimeoutError, makeCodexSessionRuntime, type CodexSessionRuntimeError, } from "./CodexSessionRuntime.ts"; @@ -68,8 +69,9 @@ describe("CodexSessionRuntime realtime voice", () => { "stops an in-flight negotiation and permits an immediate restart", { realtimeStarts: [{ started: true }, { sdp: "answer-sdp" }], + realtimeStopSdp: "stale-answer-sdp", }, - (runtime) => + (runtime, scriptPath) => Effect.gen(function* () { const started = yield* runtime.events.pipe( Stream.filter((event) => event.method === "thread/realtime/started"), @@ -89,6 +91,8 @@ describe("CodexSessionRuntime realtime voice", () => { CodexSessionRuntimeRealtimeVoiceStoppedError, ); } + const stops = NodeFS.readFileSync(`${scriptPath}.realtime-stops`, "utf8"); + assert.lengthOf(stops.trim().split("\n"), 1); const answer = yield* runtime.startRealtimeVoice("second-offer"); assert.equal(answer, "answer-sdp"); @@ -116,6 +120,46 @@ describe("CodexSessionRuntime realtime voice", () => { }), ); + runtimeTest( + "does not start a retry while the prior stop is unresolved", + { + realtimeStarts: [{ started: true }, { sdp: "must-not-start" }], + hangRealtimeStop: true, + realtimeStopStarted: true, + }, + (runtime, scriptPath) => + Effect.gen(function* () { + const started = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startRealtimeVoice("first-offer").pipe(Effect.forkScoped); + yield* Fiber.join(started); + + const stopObserved = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.stopRealtimeVoice.pipe(Effect.forkScoped); + yield* Fiber.join(stopObserved); + const retry = yield* Effect.exit(runtime.startRealtimeVoice("retry-offer")); + + assert.isTrue(Exit.isFailure(retry)); + if (Exit.isFailure(retry)) { + assert.instanceOf( + Cause.squash(retry.cause), + CodexSessionRuntimeRealtimeVoiceStopTimeoutError, + ); + } + const starts = NodeFS.readFileSync(`${scriptPath}.realtime-starts`, "utf8"); + assert.lengthOf(starts.trim().split("\n"), 1); + }), + ); + runtimeTest( "fails immediately when Codex rejects negotiation", { realtimeStarts: [{ error: "mock negotiation rejected" }] }, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index ecbec4327a21..8d60a9350b64 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -28,6 +28,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -326,10 +327,45 @@ export class CodexSessionRuntimeRealtimeVoiceStopTimeoutError extends Schema.Tag } interface PendingRealtimeVoice { + readonly _tag: "pending"; readonly providerThreadId: string; readonly answer: Deferred.Deferred; } +interface StoppingRealtimeVoice { + readonly _tag: "stopping"; + readonly providerThreadId: string; + readonly done: Deferred.Deferred; +} + +interface FailedRealtimeVoiceStop { + readonly _tag: "stopFailed"; + readonly providerThreadId: string; + readonly error: CodexSessionRuntimeError; +} + +type RealtimeVoiceState = PendingRealtimeVoice | StoppingRealtimeVoice | FailedRealtimeVoiceStop; + +type RealtimeVoiceClosedDecision = + | { readonly _tag: "none" } + | { readonly _tag: "pending"; readonly pending: PendingRealtimeVoice } + | { readonly _tag: "stopping"; readonly done: StoppingRealtimeVoice["done"] }; + +type RealtimeVoiceInstallDecision = + | { readonly _tag: "installed" } + | { readonly _tag: "occupied" } + | { readonly _tag: "failed"; readonly error: CodexSessionRuntimeError } + | { readonly _tag: "wait"; readonly done: StoppingRealtimeVoice["done"] }; + +type RealtimeVoiceStopDecision = + | { readonly _tag: "skip" } + | { readonly _tag: "follow"; readonly done: StoppingRealtimeVoice["done"] } + | { + readonly _tag: "lead"; + readonly pending: PendingRealtimeVoice | null; + readonly stopping: StoppingRealtimeVoice; + }; + interface PendingApproval { readonly requestId: ApprovalRequestId; readonly jsonRpcId: string; @@ -1199,48 +1235,165 @@ export const makeCodexSessionRuntime = ( const collabChildLiveTurnsRef = yield* Ref.make(new Map()); const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter(); const closedRef = yield* Ref.make(false); - const pendingRealtimeVoiceRef = yield* Ref.make(null); + const realtimeVoiceStateRef = yield* Ref.make(null); const realtimeVoiceNegotiationTimeoutMs = Math.max( 1, options.realtimeVoiceNegotiationTimeoutMs ?? 20_000, ); const realtimeVoiceStopTimeoutMs = Math.max(1, options.realtimeVoiceStopTimeoutMs ?? 3_000); - const clearPendingRealtimeVoice = (pending: PendingRealtimeVoice) => - Ref.update(pendingRealtimeVoiceRef, (current) => - current?.answer === pending.answer ? null : current, + const completePendingRealtimeVoice = (pending: PendingRealtimeVoice) => + Ref.modify(realtimeVoiceStateRef, (current) => + current?._tag === "pending" && current.answer === pending.answer + ? ([true, null] as const) + : ([false, current] as const), ); const failPendingRealtimeVoice = (providerThreadId: string, error: CodexSessionRuntimeError) => - Ref.get(pendingRealtimeVoiceRef).pipe( - Effect.flatMap((pending) => - pending?.providerThreadId === providerThreadId - ? Deferred.fail(pending.answer, error).pipe(Effect.asVoid) + Ref.get(realtimeVoiceStateRef).pipe( + Effect.flatMap((state) => + state?._tag === "pending" && state.providerThreadId === providerThreadId + ? Deferred.fail(state.answer, error).pipe(Effect.asVoid) : Effect.void, ), ); - const cancelPendingRealtimeVoice = ( - providerThreadId: string, - error: CodexSessionRuntimeError, - ) => - Ref.modify(pendingRealtimeVoiceRef, (current) => - current?.providerThreadId === providerThreadId - ? ([current, null] as const) - : ([null, current] as const), + const settleClosedRealtimeVoice = (providerThreadId: string, error: CodexSessionRuntimeError) => + Ref.modify( + realtimeVoiceStateRef, + (current) => { + if (current?.providerThreadId !== providerThreadId) { + return [{ _tag: "none" } as const, current]; + } + if (current._tag === "pending") { + return [{ _tag: "pending", pending: current } as const, null]; + } + if (current._tag === "stopping") { + return [{ _tag: "stopping", done: current.done } as const, null]; + } + return [{ _tag: "none" } as const, null]; + }, ).pipe( - Effect.flatMap((pending) => - pending === null ? Effect.void : Deferred.fail(pending.answer, error).pipe(Effect.asVoid), - ), + Effect.flatMap((decision) => { + if (decision._tag === "pending") { + return Deferred.fail(decision.pending.answer, error).pipe(Effect.asVoid); + } + if (decision._tag === "stopping") { + return Deferred.succeed(decision.done, undefined).pipe(Effect.asVoid); + } + return Effect.void; + }), ); + const installPendingRealtimeVoice = (pending: PendingRealtimeVoice) => + Effect.gen(function* () { + while (true) { + const decision = yield* Ref.modify< + RealtimeVoiceState | null, + RealtimeVoiceInstallDecision + >(realtimeVoiceStateRef, (current) => { + if (current === null) { + return [{ _tag: "installed" } as const, pending]; + } + if (current._tag === "pending") { + return [{ _tag: "occupied" } as const, current]; + } + if (current._tag === "stopFailed") { + return [{ _tag: "failed", error: current.error } as const, current]; + } + return [{ _tag: "wait", done: current.done } as const, current]; + }); + + switch (decision._tag) { + case "installed": + return true; + case "occupied": + return false; + case "failed": + return yield* decision.error; + case "wait": + yield* Deferred.await(decision.done); + } + } + }); + const requestRealtimeVoiceStop = (providerThreadId: string) => - client.raw - .request("thread/realtime/stop", { threadId: providerThreadId }) - .pipe(Effect.timeoutOption(`${realtimeVoiceStopTimeoutMs} millis`)); + client.raw.request("thread/realtime/stop", { threadId: providerThreadId }).pipe( + Effect.timeoutOption(`${realtimeVoiceStopTimeoutMs} millis`), + Effect.flatMap((result) => + Option.isSome(result) + ? Effect.void + : Effect.fail( + new CodexSessionRuntimeRealtimeVoiceStopTimeoutError({ + providerThreadId, + timeoutMs: realtimeVoiceStopTimeoutMs, + }), + ), + ), + ); + + const coordinateRealtimeVoiceStop = ( + providerThreadId: string, + error: CodexSessionRuntimeError, + pending?: PendingRealtimeVoice, + ) => + Effect.gen(function* () { + const done = yield* Deferred.make(); + const decision = yield* Ref.modify( + realtimeVoiceStateRef, + (current) => { + if (current?._tag === "stopping" && current.providerThreadId === providerThreadId) { + return [{ _tag: "follow", done: current.done } as const, current]; + } + if (pending && (current?._tag !== "pending" || current.answer !== pending.answer)) { + return [{ _tag: "skip" } as const, current]; + } + + const stopping = { + _tag: "stopping", + providerThreadId, + done, + } satisfies StoppingRealtimeVoice; + return [ + { + _tag: "lead", + pending: current?._tag === "pending" ? current : null, + stopping, + } as const, + stopping, + ]; + }, + ); + + if (decision._tag === "skip") return; + if (decision._tag === "follow") { + return yield* Deferred.await(decision.done); + } + if (decision.pending) { + yield* Deferred.fail(decision.pending.answer, error); + } + + const result = yield* Effect.result( + Effect.raceFirst(requestRealtimeVoiceStop(providerThreadId), Deferred.await(done)), + ); + if (Result.isFailure(result)) { + const failed = { + _tag: "stopFailed", + providerThreadId, + error: result.failure, + } satisfies FailedRealtimeVoiceStop; + yield* Ref.update(realtimeVoiceStateRef, (current) => + current?._tag === "stopping" && current.done === done ? failed : current, + ); + yield* Deferred.fail(done, result.failure); + return yield* result.failure; + } - const requestRealtimeVoiceStopBestEffort = (providerThreadId: string) => - requestRealtimeVoiceStop(providerThreadId).pipe(Effect.ignore); + yield* Ref.update(realtimeVoiceStateRef, (current) => + current?._tag === "stopping" && current.done === done ? null : current, + ); + yield* Deferred.succeed(done, undefined); + }); // `~` is not shell-expanded when env vars are set via // `child_process.spawn`; `expandHomePath` lets a configured @@ -1829,11 +1982,11 @@ export const makeCodexSessionRuntime = ( if (providerThreadId && payload.threadId !== providerThreadId) { return Effect.void; } - return Ref.get(pendingRealtimeVoiceRef).pipe( - Effect.flatMap((pending) => - pending?.providerThreadId !== payload.threadId + return Ref.get(realtimeVoiceStateRef).pipe( + Effect.flatMap((state) => + state?._tag !== "pending" || state.providerThreadId !== payload.threadId ? Effect.void - : Deferred.succeed(pending.answer, payload.sdp).pipe(Effect.asVoid), + : Deferred.succeed(state.answer, payload.sdp).pipe(Effect.asVoid), ), ); }), @@ -1863,7 +2016,7 @@ export const makeCodexSessionRuntime = ( if (providerThreadId && payload.threadId !== providerThreadId) { return Effect.void; } - return failPendingRealtimeVoice( + return settleClosedRealtimeVoice( payload.threadId, new CodexSessionRuntimeRealtimeVoiceNegotiationError({ providerThreadId: payload.threadId, @@ -2237,11 +2390,10 @@ export const makeCodexSessionRuntime = ( yield* settlePendingUserInputs({}); const providerThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); if (providerThreadId) { - yield* cancelPendingRealtimeVoice( + yield* coordinateRealtimeVoiceStop( providerThreadId, new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), - ); - yield* requestRealtimeVoiceStopBestEffort(providerThreadId); + ).pipe(Effect.ignore); } yield* updateSession(sessionRef, { status: "closed", @@ -2354,10 +2506,12 @@ export const makeCodexSessionRuntime = ( Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; const answer = yield* Deferred.make(); - const pending = { providerThreadId, answer } satisfies PendingRealtimeVoice; - const installed = yield* Ref.modify(pendingRealtimeVoiceRef, (current) => - current === null ? ([true, pending] as const) : ([false, current] as const), - ); + const pending = { + _tag: "pending", + providerThreadId, + answer, + } satisfies PendingRealtimeVoice; + const installed = yield* installPendingRealtimeVoice(pending); if (!installed) { return yield* new CodexSessionRuntimeRealtimeVoiceAlreadyStartingError({ providerThreadId, @@ -2377,9 +2531,12 @@ export const makeCodexSessionRuntime = ( Effect.onExit((exit) => Exit.isSuccess(exit) && Option.isSome(exit.value) ? Effect.void - : requestRealtimeVoiceStopBestEffort(providerThreadId), + : coordinateRealtimeVoiceStop( + providerThreadId, + new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), + pending, + ).pipe(Effect.ignore), ), - Effect.ensuring(clearPendingRealtimeVoice(pending)), ); if (Option.isNone(result)) { return yield* new CodexSessionRuntimeRealtimeVoiceAnswerTimeoutError({ @@ -2387,21 +2544,18 @@ export const makeCodexSessionRuntime = ( timeoutMs: realtimeVoiceNegotiationTimeoutMs, }); } + const completed = yield* completePendingRealtimeVoice(pending); + if (!completed) { + return yield* new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }); + } return result.value; }), stopRealtimeVoice: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; - yield* cancelPendingRealtimeVoice( + yield* coordinateRealtimeVoiceStop( providerThreadId, new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), ); - const result = yield* requestRealtimeVoiceStop(providerThreadId); - if (Option.isNone(result)) { - return yield* new CodexSessionRuntimeRealtimeVoiceStopTimeoutError({ - providerThreadId, - timeoutMs: realtimeVoiceStopTimeoutMs, - }); - } }), readThread: Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; diff --git a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs index 13489f82e7fe..a7a51f318019 100644 --- a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs +++ b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs @@ -177,6 +177,27 @@ rl.on("line", (line) => { `${process.env.T3_CODEX_COLLAB_SCRIPT}.realtime-stops`, `${JSON.stringify(message.params)}\n`, ); + if (typeof script.realtimeStopSdp === "string") { + write({ + jsonrpc: "2.0", + method: "thread/realtime/sdp", + params: { + threadId: message.params?.threadId ?? script.rootThreadId, + sdp: script.realtimeStopSdp, + }, + }); + } + if (script.realtimeStopStarted === true) { + write({ + jsonrpc: "2.0", + method: "thread/realtime/started", + params: { + threadId: message.params?.threadId ?? script.rootThreadId, + version: "v3", + realtimeSessionId: "mock-stop-observed", + }, + }); + } if (script.hangRealtimeStop !== true) { write({ id, result: {} }); } diff --git a/apps/web/src/hooks/useCodexRealtimeVoice.ts b/apps/web/src/hooks/useCodexRealtimeVoice.ts index 3983b4466037..82316f360fb2 100644 --- a/apps/web/src/hooks/useCodexRealtimeVoice.ts +++ b/apps/web/src/hooks/useCodexRealtimeVoice.ts @@ -23,6 +23,7 @@ interface LocalVoiceSession { readonly events: RTCDataChannel; readonly audio: HTMLAudioElement; readonly abort: AbortController; + readonly microphoneEnded: EventListener; } const ICE_GATHERING_TIMEOUT_MS = 15_000; @@ -86,7 +87,10 @@ function disposeLocalSession(session: LocalVoiceSession | null): void { session.peer.onconnectionstatechange = null; session.peer.ontrack = null; session.events.close(); - session.microphone.getTracks().forEach((track) => track.stop()); + session.microphone.getTracks().forEach((track) => { + track.removeEventListener("ended", session.microphoneEnded); + track.stop(); + }); session.peer.close(); session.audio.pause(); session.audio.srcObject = null; @@ -169,11 +173,38 @@ export function useCodexRealtimeVoice(input: { audio.setAttribute("playsinline", ""); microphone.getAudioTracks().forEach((track) => peer.addTrack(track, microphone)); - const session = { peer, microphone, events, audio, abort } satisfies LocalVoiceSession; + const failSession = (message: string) => { + if (generationRef.current !== generation || localSessionRef.current?.peer !== peer) { + return; + } + generationRef.current += 1; + remoteStartedRef.current = false; + clearLocalSession(); + setStatus("error"); + setError(message); + void stopRemoteVoice({ + environmentId: input.environmentId, + input: { threadId }, + }); + }; + const microphoneEnded = () => failSession("Microphone access was lost. Try again."); + const session = { + peer, + microphone, + events, + audio, + abort, + microphoneEnded, + } satisfies LocalVoiceSession; localSessionRef.current = session; + microphone.getAudioTracks().forEach((track) => { + track.addEventListener("ended", microphoneEnded); + }); peer.ontrack = (event) => { audio.srcObject = event.streams[0] ?? new MediaStream([event.track]); - void audio.play().catch(() => undefined); + void audio + .play() + .catch(() => failSession("Codex voice audio playback was blocked. Try again.")); }; peer.onconnectionstatechange = () => { if (generationRef.current !== generation) return; @@ -182,16 +213,7 @@ export function useCodexRealtimeVoice(input: { return; } if (peer.connectionState !== "failed") return; - - generationRef.current += 1; - remoteStartedRef.current = false; - clearLocalSession(); - setStatus("error"); - setError("The Codex voice connection was lost."); - void stopRemoteVoice({ - environmentId: input.environmentId, - input: { threadId }, - }); + failSession("The Codex voice connection was lost."); }; const offer = await peer.createOffer(); From c5ec5af226419595bde44e37d80daf34183579f8 Mon Sep 17 00:00:00 2001 From: Peng_Lx Date: Thu, 27 Aug 2026 07:04:15 +0800 Subject: [PATCH 5/8] fix(codex): finish realtime voice recovery paths --- .../CodexRealtimeVoice.integration.test.ts | 78 ++++++++- .../provider/Layers/CodexSessionRuntime.ts | 149 +++++++++++------- .../testFixtures/codexCollabMockPeer.mjs | 14 +- .../chat/ComposerVoiceControl.test.tsx | 18 +++ .../components/chat/ComposerVoiceControl.tsx | 34 +++- apps/web/src/hooks/useCodexRealtimeVoice.ts | 45 +++++- 6 files changed, 263 insertions(+), 75 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts b/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts index 40f6296cae38..6936e4082234 100644 --- a/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexRealtimeVoice.integration.test.ts @@ -123,9 +123,8 @@ describe("CodexSessionRuntime realtime voice", () => { runtimeTest( "does not start a retry while the prior stop is unresolved", { - realtimeStarts: [{ started: true }, { sdp: "must-not-start" }], - hangRealtimeStop: true, - realtimeStopStarted: true, + realtimeStarts: [{ started: true }, { sdp: "recovered-answer" }], + realtimeStops: [{ hangRequest: true, started: true }, {}], }, (runtime, scriptPath) => Effect.gen(function* () { @@ -157,6 +156,79 @@ describe("CodexSessionRuntime realtime voice", () => { } const starts = NodeFS.readFileSync(`${scriptPath}.realtime-starts`, "utf8"); assert.lengthOf(starts.trim().split("\n"), 1); + + const answer = yield* runtime.startRealtimeVoice("recovered-offer"); + assert.equal(answer, "recovered-answer"); + }), + ); + + runtimeTest( + "recovers when the caller interrupts an in-flight stop", + { + realtimeStarts: [{ sdp: "first-answer" }, { sdp: "recovered-answer" }], + realtimeStops: [{ hangRequest: true, started: true }, {}], + }, + (runtime, scriptPath) => + Effect.gen(function* () { + yield* runtime.startRealtimeVoice("first-offer"); + const stopObserved = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const stop = yield* runtime.stopRealtimeVoice.pipe(Effect.forkScoped); + yield* Fiber.join(stopObserved); + + yield* Fiber.interrupt(stop); + + const answer = yield* runtime.startRealtimeVoice("recovered-offer"); + assert.equal(answer, "recovered-answer"); + const stops = NodeFS.readFileSync(`${scriptPath}.realtime-stops`, "utf8"); + assert.lengthOf(stops.trim().split("\n"), 2); + }), + ); + + runtimeTest( + "allows a retry after a definitive stop error response", + { + realtimeStarts: [{ started: true }, { sdp: "retry-answer" }], + realtimeStops: [{ error: "no live realtime session" }], + }, + (runtime) => + Effect.gen(function* () { + const started = yield* runtime.events.pipe( + Stream.filter((event) => event.method === "thread/realtime/started"), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const firstStart = yield* runtime.startRealtimeVoice("first-offer").pipe(Effect.forkScoped); + yield* Fiber.join(started); + + const stop = yield* Effect.exit(runtime.stopRealtimeVoice); + assert.isTrue(Exit.isFailure(stop)); + yield* Fiber.await(firstStart); + + const answer = yield* runtime.startRealtimeVoice("retry-offer"); + assert.equal(answer, "retry-answer"); + }), + ); + + runtimeTest( + "does not start realtime voice after the runtime closes", + { realtimeStarts: [{ sdp: "must-not-start" }] }, + (runtime, scriptPath) => + Effect.gen(function* () { + yield* runtime.close; + + const exit = yield* Effect.exit(runtime.startRealtimeVoice("offer-sdp")); + + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + assert.instanceOf(Cause.squash(exit.cause), CodexSessionRuntimeRealtimeVoiceStoppedError); + } + assert.isFalse(NodeFS.existsSync(`${scriptPath}.realtime-starts`)); }), ); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 8d60a9350b64..910f7ec08e8f 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1337,63 +1337,83 @@ export const makeCodexSessionRuntime = ( error: CodexSessionRuntimeError, pending?: PendingRealtimeVoice, ) => - Effect.gen(function* () { - const done = yield* Deferred.make(); - const decision = yield* Ref.modify( - realtimeVoiceStateRef, - (current) => { - if (current?._tag === "stopping" && current.providerThreadId === providerThreadId) { - return [{ _tag: "follow", done: current.done } as const, current]; - } - if (pending && (current?._tag !== "pending" || current.answer !== pending.answer)) { - return [{ _tag: "skip" } as const, current]; - } + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const done = yield* Deferred.make(); + const decision = yield* Ref.modify( + realtimeVoiceStateRef, + (current) => { + if (current?._tag === "stopping" && current.providerThreadId === providerThreadId) { + return [{ _tag: "follow", done: current.done } as const, current]; + } + if (pending && (current?._tag !== "pending" || current.answer !== pending.answer)) { + return [{ _tag: "skip" } as const, current]; + } - const stopping = { - _tag: "stopping", - providerThreadId, - done, - } satisfies StoppingRealtimeVoice; - return [ - { - _tag: "lead", - pending: current?._tag === "pending" ? current : null, + const stopping = { + _tag: "stopping", + providerThreadId, + done, + } satisfies StoppingRealtimeVoice; + return [ + { + _tag: "lead", + pending: current?._tag === "pending" ? current : null, + stopping, + } as const, stopping, - } as const, - stopping, - ]; - }, - ); + ]; + }, + ); - if (decision._tag === "skip") return; - if (decision._tag === "follow") { - return yield* Deferred.await(decision.done); - } - if (decision.pending) { - yield* Deferred.fail(decision.pending.answer, error); - } + if (decision._tag === "skip") return; + if (decision._tag === "follow") { + return yield* restore(Deferred.await(decision.done)); + } + if (decision.pending) { + yield* Deferred.fail(decision.pending.answer, error); + } - const result = yield* Effect.result( - Effect.raceFirst(requestRealtimeVoiceStop(providerThreadId), Deferred.await(done)), - ); - if (Result.isFailure(result)) { - const failed = { - _tag: "stopFailed", - providerThreadId, - error: result.failure, - } satisfies FailedRealtimeVoiceStop; - yield* Ref.update(realtimeVoiceStateRef, (current) => - current?._tag === "stopping" && current.done === done ? failed : current, + const settleFailedStop = (failure: CodexSessionRuntimeError, blockRetry: boolean) => { + const failed = { + _tag: "stopFailed", + providerThreadId, + error: failure, + } satisfies FailedRealtimeVoiceStop; + return Ref.update(realtimeVoiceStateRef, (current) => + current?._tag === "stopping" && current.done === done + ? blockRetry + ? failed + : null + : current, + ).pipe(Effect.andThen(Deferred.fail(done, failure)), Effect.asVoid); + }; + const result = yield* Effect.result( + restore( + Effect.raceFirst(requestRealtimeVoiceStop(providerThreadId), Deferred.await(done)), + ).pipe( + Effect.onInterrupt(() => + settleFailedStop( + new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), + true, + ), + ), + ), ); - yield* Deferred.fail(done, result.failure); - return yield* result.failure; - } + if (Result.isFailure(result)) { + yield* settleFailedStop( + result.failure, + result.failure._tag === "CodexSessionRuntimeRealtimeVoiceStopTimeoutError", + ); + return yield* result.failure; + } - yield* Ref.update(realtimeVoiceStateRef, (current) => - current?._tag === "stopping" && current.done === done ? null : current, - ); - yield* Deferred.succeed(done, undefined); - }); + yield* Ref.update(realtimeVoiceStateRef, (current) => + current?._tag === "stopping" && current.done === done ? null : current, + ); + yield* Deferred.succeed(done, undefined); + }), + ); // `~` is not shell-expanded when env vars are set via // `child_process.spawn`; `expandHomePath` lets a configured @@ -2505,6 +2525,16 @@ export const makeCodexSessionRuntime = ( startRealtimeVoice: (sdp) => Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; + const stoppedError = new CodexSessionRuntimeRealtimeVoiceStoppedError({ + providerThreadId, + }); + if (yield* Ref.get(closedRef)) { + return yield* stoppedError; + } + const previousState = yield* Ref.get(realtimeVoiceStateRef); + if (previousState?._tag === "stopFailed") { + yield* coordinateRealtimeVoiceStop(providerThreadId, stoppedError); + } const answer = yield* Deferred.make(); const pending = { _tag: "pending", @@ -2517,6 +2547,12 @@ export const makeCodexSessionRuntime = ( providerThreadId, }); } + if (yield* Ref.get(closedRef)) { + yield* coordinateRealtimeVoiceStop(providerThreadId, stoppedError, pending).pipe( + Effect.ignore, + ); + return yield* stoppedError; + } const result = yield* Effect.raceFirst( client.raw @@ -2531,11 +2567,9 @@ export const makeCodexSessionRuntime = ( Effect.onExit((exit) => Exit.isSuccess(exit) && Option.isSome(exit.value) ? Effect.void - : coordinateRealtimeVoiceStop( - providerThreadId, - new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }), - pending, - ).pipe(Effect.ignore), + : coordinateRealtimeVoiceStop(providerThreadId, stoppedError, pending).pipe( + Effect.ignore, + ), ), ); if (Option.isNone(result)) { @@ -2546,7 +2580,8 @@ export const makeCodexSessionRuntime = ( } const completed = yield* completePendingRealtimeVoice(pending); if (!completed) { - return yield* new CodexSessionRuntimeRealtimeVoiceStoppedError({ providerThreadId }); + yield* coordinateRealtimeVoiceStop(providerThreadId, stoppedError).pipe(Effect.ignore); + return yield* stoppedError; } return result.value; }), diff --git a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs index a7a51f318019..0d0924fa2cd7 100644 --- a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs +++ b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs @@ -18,6 +18,7 @@ const script = JSON.parse(NodeFS.readFileSync(process.env.T3_CODEX_COLLAB_SCRIPT const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`); let turnStartCount = 0; let realtimeStartCount = 0; +let realtimeStopCount = 0; let activeTurn; const rl = NodeReadline.createInterface({ input: process.stdin }); @@ -173,21 +174,24 @@ rl.on("line", (line) => { return; } if (method === "thread/realtime/stop") { + const realtimeStop = script.realtimeStops?.[realtimeStopCount] ?? {}; + realtimeStopCount += 1; NodeFS.appendFileSync( `${process.env.T3_CODEX_COLLAB_SCRIPT}.realtime-stops`, `${JSON.stringify(message.params)}\n`, ); - if (typeof script.realtimeStopSdp === "string") { + const stopSdp = realtimeStop.sdp ?? script.realtimeStopSdp; + if (typeof stopSdp === "string") { write({ jsonrpc: "2.0", method: "thread/realtime/sdp", params: { threadId: message.params?.threadId ?? script.rootThreadId, - sdp: script.realtimeStopSdp, + sdp: stopSdp, }, }); } - if (script.realtimeStopStarted === true) { + if (realtimeStop.started === true || script.realtimeStopStarted === true) { write({ jsonrpc: "2.0", method: "thread/realtime/started", @@ -198,7 +202,9 @@ rl.on("line", (line) => { }, }); } - if (script.hangRealtimeStop !== true) { + if (typeof realtimeStop.error === "string") { + write({ id, error: { code: -32000, message: realtimeStop.error } }); + } else if (realtimeStop.hangRequest !== true && script.hangRealtimeStop !== true) { write({ id, result: {} }); } return; diff --git a/apps/web/src/components/chat/ComposerVoiceControl.test.tsx b/apps/web/src/components/chat/ComposerVoiceControl.test.tsx index 34025cabe2a2..9fc2fe0b6fe4 100644 --- a/apps/web/src/components/chat/ComposerVoiceControl.test.tsx +++ b/apps/web/src/components/chat/ComposerVoiceControl.test.tsx @@ -16,6 +16,7 @@ const makeVoice = ( error: null, start: async () => {}, stop: async () => {}, + resumeAudio: async () => {}, toggleMuted: () => {}, ...overrides, }); @@ -59,4 +60,21 @@ describe("ComposerVoiceControl", () => { expect(markup).toContain('aria-label="Unmute microphone"'); expect(markup).toContain('aria-label="End Codex voice"'); }); + + it("keeps a blocked audio session live and offers a playback retry", () => { + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain("Audio paused"); + expect(markup).toContain('aria-label="Resume Codex audio"'); + expect(markup).toContain('aria-label="End Codex voice"'); + expect(markup).toContain("bg-warning"); + }); }); diff --git a/apps/web/src/components/chat/ComposerVoiceControl.tsx b/apps/web/src/components/chat/ComposerVoiceControl.tsx index 512c59dd8148..1fbdcf11a8e9 100644 --- a/apps/web/src/components/chat/ComposerVoiceControl.tsx +++ b/apps/web/src/components/chat/ComposerVoiceControl.tsx @@ -1,5 +1,5 @@ import { memo, type PointerEventHandler, type ReactElement } from "react"; -import { MicIcon, MicOffIcon, PhoneOffIcon } from "lucide-react"; +import { MicIcon, MicOffIcon, PhoneOffIcon, Volume2Icon } from "lucide-react"; import type { CodexRealtimeVoiceController } from "~/hooks/useCodexRealtimeVoice"; import { cn } from "~/lib/utils"; @@ -25,7 +25,8 @@ export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { readonly disabledReason?: string; }) { const { voice } = props; - const active = voice.status === "connecting" || voice.status === "live"; + const active = + voice.status === "connecting" || voice.status === "live" || voice.status === "playback-blocked"; if (!active) { const tooltip = !voice.supported @@ -56,7 +57,13 @@ export const ComposerVoiceControl = memo(function ComposerVoiceControl(props: { } const label = - voice.status === "connecting" ? "Connecting…" : voice.muted ? "Muted" : "Voice live"; + voice.status === "connecting" + ? "Connecting…" + : voice.status === "playback-blocked" + ? "Audio paused" + : voice.muted + ? "Muted" + : "Voice live"; return (