From dd786f08ca08d5a97e204c7d78351ad2aec9da8c Mon Sep 17 00:00:00 2001 From: Romil Bijarnia Date: Sat, 14 Nov 2026 00:36:24 +1100 Subject: [PATCH] feat(codex): add subscription-backed live voice conversations --- apps/server/src/liveVoice/LiveVoice.test.ts | 218 ++++++++++++ apps/server/src/liveVoice/LiveVoice.ts | 326 ++++++++++++++++++ apps/server/src/liveVoice/http.ts | 52 +++ apps/server/src/liveVoice/protocol.test.ts | 37 ++ apps/server/src/liveVoice/protocol.ts | 61 ++++ .../src/provider/CodexLiveVoice.test.ts | 42 +++ apps/server/src/provider/CodexLiveVoice.ts | 199 +++++++++++ .../src/provider/Drivers/CodexDriver.ts | 5 + .../provider/Drivers/CodexManagedProvider.ts | 15 + apps/server/src/server.ts | 4 + apps/web/src/components/chat/ChatComposer.tsx | 20 +- .../src/components/chat/LiveVoiceButton.tsx | 217 ++++++++++++ apps/web/src/speech/browserLiveVoice.test.ts | 59 ++++ apps/web/src/speech/browserLiveVoice.ts | 161 +++++++++ docs/user/voice-input.md | 23 ++ .../client-runtime/src/voice-input/index.ts | 2 + .../src/voice-input/liveVoice.ts | 77 +++++ packages/contracts/src/environmentHttp.ts | 43 +++ packages/contracts/src/index.ts | 2 + packages/contracts/src/liveVoice.ts | 60 ++++ .../fixtures/codex-app-server-mock-peer.ts | 13 +- packages/provider-core/src/server/driver.ts | 21 ++ 22 files changed, 1649 insertions(+), 8 deletions(-) create mode 100644 apps/server/src/liveVoice/LiveVoice.test.ts create mode 100644 apps/server/src/liveVoice/LiveVoice.ts create mode 100644 apps/server/src/liveVoice/http.ts create mode 100644 apps/server/src/liveVoice/protocol.test.ts create mode 100644 apps/server/src/liveVoice/protocol.ts create mode 100644 apps/server/src/provider/CodexLiveVoice.test.ts create mode 100644 apps/server/src/provider/CodexLiveVoice.ts create mode 100644 apps/web/src/components/chat/LiveVoiceButton.tsx create mode 100644 apps/web/src/speech/browserLiveVoice.test.ts create mode 100644 apps/web/src/speech/browserLiveVoice.ts create mode 100644 packages/client-runtime/src/voice-input/liveVoice.ts create mode 100644 packages/contracts/src/liveVoice.ts diff --git a/apps/server/src/liveVoice/LiveVoice.test.ts b/apps/server/src/liveVoice/LiveVoice.test.ts new file mode 100644 index 0000000000000..138565cf2784d --- /dev/null +++ b/apps/server/src/liveVoice/LiveVoice.test.ts @@ -0,0 +1,218 @@ +import { assert, it } from "@effect/vitest"; +import { + AuthSessionId, + ThreadId, + ProviderInstanceId, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; +import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as Providers from "../provider/ProviderInstanceRegistry.ts"; +import * as Projections from "../orchestration-v2/ProjectionStore.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as LiveVoice from "./LiveVoice.ts"; + +const threadId = ThreadId.make("voice-thread"); +const owner = AuthSessionId.make("owner"); +const input = (id: string) => ({ threadId, sessionId: id, voice: "sol", sdp: id }); +const fixture = Effect.fn("test.liveVoice.fixture")(function* ( + delayed = false, + events: ReadonlyArray = [], + delayedValidation = false, +) { + const validating = yield* Deferred.make(); + const validated = yield* Deferred.make(); + const settled = yield* Deferred.make(); + let reads = 0; + const opened = yield* Queue.unbounded(); + const ready = yield* Deferred.make(); + if (!delayed) yield* Deferred.succeed(ready, undefined); + const closed = yield* Queue.unbounded(); + const commands = yield* Ref.make>([]); + const instance = { + enabled: true, + instanceId: ProviderInstanceId.make("codex"), + snapshot: { + getSnapshot: Effect.succeed({ + auth: { status: "authenticated", accountLabel: "same-account" }, + }), + }, + liveVoice: { + options: Effect.succeed({ voices: ["sol"], defaultVoice: "sol" }), + open: (request: { sdp: string }) => + Effect.gen(function* () { + yield* Effect.addFinalizer(() => Queue.offer(closed, request.sdp)); + yield* Queue.offer(opened, request.sdp); + yield* Deferred.await(ready); + return { + sdp: "answer", + events: events.length + ? Stream.fromIterable(events).pipe( + Stream.ensuring(Deferred.succeed(settled, undefined)), + ) + : Stream.never, + send: () => Effect.void, + }; + }), + }, + } as unknown as ProviderInstance; + const selected = yield* Ref.make(instance); + const projection = { + thread: { + id: threadId, + archivedAt: null, + modelSelection: { instanceId: instance.instanceId, model: "gpt-6-luna" }, + }, + messages: [], + } as unknown as OrchestrationV2ThreadProjection; + const dependencies = Layer.mergeAll( + Layer.mock(Providers.ProviderInstanceRegistry, { getInstance: () => Ref.get(selected) }), + Layer.mock(Projections.ProjectionStoreV2, { + getThreadSnapshotWindow: () => + Effect.gen(function* () { + if (delayedValidation && ++reads > 1) { + yield* Deferred.succeed(validating, undefined); + yield* Deferred.await(validated); + } + return { schemaVersion: 2 as const, snapshotSequence: 0, projection }; + }), + }), + Layer.mock(Orchestrator.OrchestratorV2, { + dispatch: (command) => + Ref.update(commands, (values) => [...values, command]).pipe( + Effect.as({ sequence: 1, storedEvents: [] }), + ), + streamStoredEventsFrom: () => Stream.never, + }), + ); + return { + layer: LiveVoice.layer.pipe(Layer.provide(dependencies)), + ready, + validating, + validated, + settled, + opened, + closed, + commands, + selected, + instance, + }; +}); + +it.effect("stops only the owning client's session and closes it when its account changes", () => + Effect.gen(function* () { + const test = yield* fixture(); + yield* Effect.gen(function* () { + const service = yield* LiveVoice.LiveVoice; + yield* service.start(input("first"), owner); + const denied = yield* service.stop("first", AuthSessionId.make("other")).pipe(Effect.flip); + assert.equal(denied.reason, "expired"); + yield* Ref.set(test.selected, { ...test.instance }); + const state = yield* service.poll("first", owner, 0); + assert.equal(state.phase, "error"); + assert.equal(yield* Queue.take(test.closed), "first"); + }).pipe(Effect.provide(test.layer)); + }), +); + +it.effect("a cancelled startup cannot release a newer session's reservation", () => + Effect.gen(function* () { + const test = yield* fixture(true); + yield* Effect.gen(function* () { + const service = yield* LiveVoice.LiveVoice; + const first = yield* service.start(input("first"), owner).pipe(Effect.forkChild); + assert.equal(yield* Queue.take(test.opened), "first"); + yield* service.stop("first", owner); + const second = yield* service.start(input("second"), owner).pipe(Effect.forkChild); + assert.equal(yield* Queue.take(test.opened), "second"); + yield* Deferred.succeed(test.ready, undefined); + assert.equal((yield* Fiber.join(first).pipe(Effect.flip)).reason, "expired"); + yield* Fiber.join(second); + const third = yield* service.start(input("third"), owner).pipe(Effect.flip); + assert.equal(third.reason, "busy"); + yield* service.stop("second", owner); + }).pipe(Effect.provide(test.layer)); + }), +); + +it.effect("expires a disconnected client's session without a stop request", () => + Effect.gen(function* () { + const test = yield* fixture(); + yield* Effect.gen(function* () { + const service = yield* LiveVoice.LiveVoice; + yield* service.start(input("first"), owner); + yield* TestClock.adjust("50 seconds"); + assert.equal(yield* Queue.take(test.closed), "first"); + }).pipe(Effect.provide(test.layer)); + }), +); + +it.effect("dispatches a repeated voice delegation only once through the normal orchestrator", () => + Effect.gen(function* () { + const delegation = JSON.stringify({ + type: "delegation.created", + item: { + id: "request-1", + type: "delegation", + target: "client", + content: [{ type: "input_text", text: "Check the build" }], + }, + }); + const test = yield* fixture(false, [delegation, delegation]); + yield* Effect.gen(function* () { + const service = yield* LiveVoice.LiveVoice; + yield* service.start(input("first"), owner); + yield* Queue.take(test.closed); + assert.deepEqual(yield* Ref.get(test.commands), [ + { + type: "message.dispatch", + commandId: "voice:first:request-1", + messageId: "voice:first:request-1", + threadId, + createdBy: "user", + creationSource: "web", + text: "Check the build", + attachments: [], + deliveryIntent: "auto", + dispatchMode: { type: "start_immediately" }, + }, + ]); + }).pipe(Effect.provide(test.layer)); + }), +); + +it.effect("stops a delegation waiting for thread validation before it can dispatch work", () => + Effect.gen(function* () { + const test = yield* fixture( + false, + [ + JSON.stringify({ + type: "delegation.created", + item: { + id: "cancelled-request", + type: "delegation", + target: "client", + content: [{ type: "input_text", text: "Check the build" }], + }, + }), + ], + true, + ); + yield* Effect.gen(function* () { + const service = yield* LiveVoice.LiveVoice; + yield* service.start(input("first"), owner); + yield* Deferred.await(test.validating); + yield* service.stop("first", owner); + yield* Deferred.succeed(test.validated, undefined); + yield* Deferred.await(test.settled); + assert.deepEqual(yield* Ref.get(test.commands), []); + }).pipe(Effect.provide(test.layer)); + }), +); diff --git a/apps/server/src/liveVoice/LiveVoice.ts b/apps/server/src/liveVoice/LiveVoice.ts new file mode 100644 index 0000000000000..4bb7979801c1f --- /dev/null +++ b/apps/server/src/liveVoice/LiveVoice.ts @@ -0,0 +1,326 @@ +import { + CommandId, + MessageId, + LiveVoiceError, + type AuthSessionId, + type ThreadId, + type LiveVoiceOptions, + type LiveVoiceStart, + type LiveVoiceState, +} from "@t3tools/contracts"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; +import type { ProviderLiveVoice } from "@t3tools/provider-core/server/driver"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +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 Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as ProviderInstances from "../provider/ProviderInstanceRegistry.ts"; +import * as Projections from "../orchestration-v2/ProjectionStore.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import { VoiceEvent, voiceContextChunks, voiceDelegation } from "./protocol.ts"; + +const isLiveVoiceError = Schema.is(LiveVoiceError); +const decodeVoiceEvent = Schema.decodeEffect(Schema.fromJsonString(VoiceEvent)); + +type Connection = Effect.Success>; +type Entry = { + id: string; + owner: AuthSessionId; + threadId: ThreadId; + scope: Scope.Closeable; + instance: ProviderInstance; + authStamp: string; + selection: string; + connection?: Connection; + state: LiveVoiceState; + changed: Deferred.Deferred; + expires: number; + closed: boolean; + delegated: Set; +}; + +export class LiveVoice extends Context.Service< + LiveVoice, + { + readonly options: (threadId: ThreadId) => Effect.Effect; + readonly start: ( + input: LiveVoiceStart, + owner: AuthSessionId, + ) => Effect.Effect<{ readonly sdp: string }, LiveVoiceError>; + readonly poll: ( + id: string, + owner: AuthSessionId, + revision: number, + ) => Effect.Effect; + readonly stop: (id: string, owner: AuthSessionId) => Effect.Effect; + } +>()("t3/liveVoice/LiveVoice") {} + +const make = Effect.gen(function* () { + const providers = yield* ProviderInstances.ProviderInstanceRegistry; + const projections = yield* Projections.ProjectionStoreV2; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const serviceScope = yield* Scope.Scope; + const entries = new Map(); + const byThread = new Map(); + const unavailable = () => new LiveVoiceError({ reason: "unavailable" }); + const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); + const load = Effect.fn("LiveVoice.load")( + function* (threadId: ThreadId) { + const snapshot = yield* projections.getThreadSnapshotWindow(threadId, { + rowLimit: 40, + userTurnLimit: 10, + }); + const { thread } = snapshot.projection; + const instance = yield* providers.getInstance(thread.modelSelection.instanceId); + if (!instance?.enabled || !instance.liveVoice || thread.archivedAt !== null) + return yield* unavailable(); + const account = yield* instance.snapshot.getSnapshot; + return { + instance, + voice: instance.liveVoice, + projection: snapshot.projection, + authStamp: yield* encodeJson(account.auth), + selection: yield* encodeJson(thread.modelSelection), + }; + }, + Effect.mapError((error) => (isLiveVoiceError(error) ? error : unavailable())), + ); + const publish = Effect.fn("LiveVoice.publish")(function* ( + entry: Entry, + update: Partial, + ) { + entry.state = { ...entry.state, ...update, revision: entry.state.revision + 1 }; + const previous = entry.changed; + entry.changed = yield* Deferred.make(); + yield* Deferred.succeed(previous, undefined); + }); + const finish = Effect.fn("LiveVoice.finish")(function* ( + entry: Entry, + error: string | null = null, + ) { + if (entry.closed) return; + entry.closed = true; + if (byThread.get(entry.threadId) === entry.id) byThread.delete(entry.threadId); + yield* publish(entry, { phase: error ? "error" : "ended", error }); + yield* Scope.close(entry.scope, Exit.void); + entry.expires = (yield* Clock.currentTimeMillis) + 45_000; + }); + const current = Effect.fn("LiveVoice.current")(function* (entry: Entry) { + const selected = yield* load(entry.threadId); + if ( + selected.instance !== entry.instance || + selected.authStamp !== entry.authStamp || + selected.selection !== entry.selection + ) { + return yield* new LiveVoiceError({ reason: "thread-changed" }); + } + return selected; + }); + const speak = Effect.fn("LiveVoice.speak")(function* ( + entry: Entry, + delegation: string, + text: string, + ) { + for (const chunk of voiceContextChunks(text.slice(0, 8000))) { + yield* entry.connection!.send({ + type: "delegation.context.append", + delegation_item_id: delegation, + channel: "speakable", + content: [{ type: "input_text", text: chunk }], + }); + } + }); + // Control frames share the session lifecycle; avoid a trace span for each streamed event. + const handle = Effect.fnUntraced(function* (entry: Entry, raw: string) { + const event = yield* decodeVoiceEvent(raw); + if (entry.closed) return; + if (event.type === "error") { + yield* finish(entry, "The voice service ended the connection.").pipe( + Effect.forkIn(serviceScope), + ); + return; + } + if ( + event.type === "turn.done" && + event.turn && + (event.turn.role === "user" || event.turn.role === "assistant") && + event.turn.transcript + ) { + const transcript: LiveVoiceState["transcripts"][number] = { + id: event.turn.id, + role: event.turn.role, + text: event.turn.transcript.slice(0, 16_384), + }; + yield* publish(entry, { + transcripts: [ + ...entry.state.transcripts.filter((item) => item.id !== transcript.id), + transcript, + ].slice(-100), + }); + } + const task = voiceDelegation(event); + if (!task || entry.delegated.has(task.id)) return; + entry.delegated.add(task.id); + yield* current(entry); + if (entry.closed) return; + const dispatched = yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`voice:${entry.id}:${task.id}`), + messageId: MessageId.make(`voice:${entry.id}:${task.id}`), + threadId: entry.threadId, + createdBy: "user", + creationSource: "web", + text: task.text, + attachments: [], + deliveryIntent: "auto", + dispatchMode: { type: "start_immediately" }, + }); + const runIds = new Set( + dispatched.storedEvents.flatMap(({ event }) => + event.type === "run.created" + ? [event.payload.id] + : event.type === "message.updated" && event.payload.role === "user" && event.payload.runId + ? [event.payload.runId] + : [], + ), + ); + yield* orchestrator + .streamStoredEventsFrom({ threadId: entry.threadId, afterSequence: dispatched.sequence }) + .pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event) => + event.type === "run.updated" && + runIds.has(event.payload.id) && + ["completed", "failed", "interrupted", "cancelled", "rolled_back"].includes( + event.payload.status, + ), + ), + Stream.take(1), + Stream.runForEach( + Effect.fn("LiveVoice.deliverResult")(function* (event) { + if (event.type !== "run.updated") return; + const snapshot = yield* projections.getThreadSnapshotWindow(entry.threadId, { + rowLimit: 40, + requiredRunId: event.payload.id, + }); + const reply = snapshot.projection.messages + .filter( + (message) => + message.runId === event.payload.id && + message.role === "assistant" && + !message.streaming, + ) + .at(-1)?.text; + const result = + event.payload.status === "completed" + ? reply || "The coding task completed. See the T3 conversation for details." + : `The coding task ${event.payload.status}. Check the T3 conversation for details.`; + yield* speak(entry, task.id, result); + }), + ), + Effect.catch(() => + speak(entry, task.id, "The task result could not be read. Check the conversation in T3."), + ), + Effect.forkIn(entry.scope), + ); + }); + const owned = Effect.fn("LiveVoice.owned")(function* (id: string, owner: AuthSessionId) { + const entry = entries.get(id); + if (!entry || entry.owner !== owner) return yield* new LiveVoiceError({ reason: "expired" }); + return entry; + }); + yield* Effect.addFinalizer(() => + Effect.forEach(entries.values(), (entry) => finish(entry), { discard: true }), + ); + yield* Effect.gen(function* () { + while (true) { + yield* Effect.sleep("10 seconds"); + const now = yield* Clock.currentTimeMillis; + for (const entry of entries.values()) { + if (entry.expires > now) continue; + yield* finish(entry); + entries.delete(entry.id); + } + } + }).pipe(Effect.forkScoped); + return LiveVoice.of({ + options: Effect.fn("LiveVoice.options")(function* (threadId) { + return yield* (yield* load(threadId)).voice.options; + }), + start: Effect.fn("LiveVoice.start")(function* (input, owner) { + if (entries.has(input.sessionId) || byThread.has(input.threadId)) + return yield* new LiveVoiceError({ reason: "busy" }); + byThread.set(input.threadId, input.sessionId); + const scope = yield* Scope.make(); + let entry: Entry | undefined; + return yield* Effect.gen(function* () { + const selected = yield* load(input.threadId); + entry = { + id: input.sessionId, + owner, + threadId: input.threadId, + scope, + instance: selected.instance, + authStamp: selected.authStamp, + selection: selected.selection, + state: { revision: 0, phase: "connected", transcripts: [], error: null }, + changed: yield* Deferred.make(), + expires: (yield* Clock.currentTimeMillis) + 45_000, + closed: false, + delegated: new Set(), + }; + entries.set(entry.id, entry); + const context = selected.projection.messages + .filter((message) => message.role !== "system" && !message.streaming) + .slice(-20) + .map((message) => ({ + role: message.role as "user" | "assistant", + text: message.text.slice(-1500), + })); + const connection = yield* selected.voice + .open({ sdp: input.sdp, voice: input.voice, context }) + .pipe(Effect.provideService(Scope.Scope, scope)); + if (entry.closed) return yield* new LiveVoiceError({ reason: "expired" }); + entry.connection = connection; + const active = entry; + yield* connection.events.pipe( + Stream.runForEach((event) => handle(active, event)), + Effect.matchEffect({ + onSuccess: () => finish(active), + onFailure: () => + finish(active, "Voice disconnected. Your T3 tasks remain in the conversation."), + }), + Effect.forkIn(serviceScope), + ); + return { sdp: connection.sdp }; + }).pipe( + Effect.onError(() => { + if (byThread.get(input.threadId) === input.sessionId) byThread.delete(input.threadId); + entries.delete(input.sessionId); + return Scope.close(scope, Exit.void); + }), + ); + }), + poll: Effect.fn("LiveVoice.poll")(function* (id, owner, revision) { + const entry = yield* owned(id, owner); + if (!entry.closed) { + yield* current(entry).pipe(Effect.catch((error) => finish(entry, error.message))); + entry.expires = (yield* Clock.currentTimeMillis) + 45_000; + } + if (!entry.closed && entry.state.revision === revision) + yield* Deferred.await(entry.changed).pipe(Effect.timeoutOption("15 seconds")); + return entry.state; + }), + stop: Effect.fn("LiveVoice.stop")(function* (id, owner) { + yield* finish(yield* owned(id, owner)); + }), + }); +}); +export const layer = Layer.effect(LiveVoice, make); diff --git a/apps/server/src/liveVoice/http.ts b/apps/server/src/liveVoice/http.ts new file mode 100644 index 0000000000000..1fad201a636d2 --- /dev/null +++ b/apps/server/src/liveVoice/http.ts @@ -0,0 +1,52 @@ +import { + AuthOrchestrationOperateScope, + EnvironmentAuthenticatedPrincipal, + EnvironmentHttpApi, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; +import { annotateEnvironmentRequest, requireEnvironmentScope } from "../auth/http.ts"; +import * as LiveVoice from "./LiveVoice.ts"; + +export const layer = HttpApiBuilder.group( + EnvironmentHttpApi, + "liveVoice", + Effect.fnUntraced(function* (handlers) { + const service = yield* LiveVoice.LiveVoice; + const authorize = Effect.fnUntraced(function* (endpoint: string) { + yield* annotateEnvironmentRequest(endpoint); + yield* requireEnvironmentScope(AuthOrchestrationOperateScope); + return (yield* EnvironmentAuthenticatedPrincipal).sessionId; + }); + return handlers + .handle( + "options", + Effect.fn("environment.liveVoice.options")(function* (args) { + yield* authorize(args.endpoint.name); + return yield* service.options(args.params.threadId); + }), + ) + .handle( + "start", + Effect.fn("environment.liveVoice.start")(function* (args) { + return yield* service.start(args.payload, yield* authorize(args.endpoint.name)); + }), + ) + .handle( + "poll", + Effect.fn("environment.liveVoice.poll")(function* (args) { + return yield* service.poll( + args.params.sessionId, + yield* authorize(args.endpoint.name), + args.query.revision, + ); + }), + ) + .handle( + "stop", + Effect.fn("environment.liveVoice.stop")(function* (args) { + yield* service.stop(args.params.sessionId, yield* authorize(args.endpoint.name)); + }), + ); + }), +); diff --git a/apps/server/src/liveVoice/protocol.test.ts b/apps/server/src/liveVoice/protocol.test.ts new file mode 100644 index 0000000000000..027c0cbde3246 --- /dev/null +++ b/apps/server/src/liveVoice/protocol.test.ts @@ -0,0 +1,37 @@ +import { describe, it, expect } from "@effect/vitest"; +import { voiceContextChunks, voiceDelegation } from "./protocol.ts"; + +describe("subscription voice protocol", () => { + it("accepts only bounded client delegations, not transcripts or hosted tasks", () => { + const item = { + id: "task-1", + type: "delegation", + target: "client", + content: [ + { type: "input_text", text: "Check the build" }, + { type: "image", text: "ignored" }, + ], + }; + expect(voiceDelegation({ type: "delegation.created", item })).toEqual({ + id: "task-1", + text: "Check the build", + }); + expect(voiceDelegation({ type: "input_transcript.added", item })).toBeNull(); + expect( + voiceDelegation({ type: "delegation.created", item: { ...item, target: "responses" } }), + ).toBeNull(); + expect( + voiceDelegation({ + type: "delegation.created", + item: { ...item, content: [{ type: "input_text", text: "x".repeat(16385) }] }, + }), + ).toBeNull(); + }); + it("keeps Unicode results intact within the wire's byte limit", () => { + const text = "Result: 中文 𝄞 café. ".repeat(100); + const chunks = voiceContextChunks(text); + expect(chunks.join("")).toBe(text); + expect(chunks.every((chunk) => new TextEncoder().encode(chunk).length <= 480)).toBe(true); + expect(voiceContextChunks("")).toEqual([]); + }); +}); diff --git a/apps/server/src/liveVoice/protocol.ts b/apps/server/src/liveVoice/protocol.ts new file mode 100644 index 0000000000000..7a10b8c4624c9 --- /dev/null +++ b/apps/server/src/liveVoice/protocol.ts @@ -0,0 +1,61 @@ +import * as Schema from "effect/Schema"; + +export const VoiceEvent = Schema.Struct({ + type: Schema.String, + turn: Schema.optionalKey( + Schema.Struct({ + id: Schema.String, + role: Schema.String, + transcript: Schema.optionalKey(Schema.String), + }), + ), + item: Schema.optionalKey( + Schema.Struct({ + id: Schema.optionalKey(Schema.String), + type: Schema.optionalKey(Schema.String), + target: Schema.optionalKey(Schema.String), + content: Schema.optionalKey( + Schema.Array( + Schema.Struct({ type: Schema.String, text: Schema.optionalKey(Schema.String) }), + ), + ), + }), + ), +}); +export type VoiceEvent = typeof VoiceEvent.Type; + +export function voiceDelegation(event: VoiceEvent) { + const item = event.item; + if ( + event.type !== "delegation.created" || + item?.type !== "delegation" || + item.target !== "client" || + !item.id + ) + return null; + const text = item.content + ?.filter((part) => part.type === "input_text") + .map((part) => part.text ?? "") + .join("\n") + .trim(); + return text && text.length <= 16_384 ? { id: item.id, text } : null; +} + +export function voiceContextChunks(text: string): string[] { + const encoder = new TextEncoder(); + const chunks: string[] = []; + let chunk = ""; + let bytes = 0; + for (const character of text) { + const length = encoder.encode(character).length; + if (bytes + length > 480) { + chunks.push(chunk); + chunk = ""; + bytes = 0; + } + chunk += character; + bytes += length; + } + if (chunk) chunks.push(chunk); + return chunks; +} diff --git a/apps/server/src/provider/CodexLiveVoice.test.ts b/apps/server/src/provider/CodexLiveVoice.test.ts new file mode 100644 index 0000000000000..269e356ac0162 --- /dev/null +++ b/apps/server/src/provider/CodexLiveVoice.test.ts @@ -0,0 +1,42 @@ +import { assert, it } from "@effect/vitest"; +import { CodexSettings } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { FetchHttpClient } from "effect/http"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as NodeURL from "node:url"; +import { makeCodexLiveVoice } from "./CodexLiveVoice.ts"; + +const config = Schema.decodeSync(CodexSettings)({ binaryPath: process.execPath }); +const peer = NodeURL.fileURLToPath( + new URL( + "../../../../packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts", + import.meta.url, + ), +); +it.layer(Layer.merge(NodeServices.layer, FetchHttpClient.layer))( + "Codex subscription authentication", + (it) => { + it.effect("rejects an API-key provider before attempting a voice request", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const voice = yield* makeCodexLiveVoice(Effect.succeed({ config, environment: {} })).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, { + ...spawner, + spawn: () => + spawner.spawn( + ChildProcess.make(process.execPath, [peer], { + env: { CODEX_APP_SERVER_TEST_ACCOUNT: "apiKey" }, + extendEnv: true, + }), + ), + }), + ); + const error = yield* voice.options.pipe(Effect.flip); + assert.equal(error.reason, "authentication"); + }), + ); + }, +); diff --git a/apps/server/src/provider/CodexLiveVoice.ts b/apps/server/src/provider/CodexLiveVoice.ts new file mode 100644 index 0000000000000..b4b91b95fce98 --- /dev/null +++ b/apps/server/src/provider/CodexLiveVoice.ts @@ -0,0 +1,199 @@ +import { LiveVoiceError, type CodexSettings } from "@t3tools/contracts"; +import type { ProviderLiveVoice } from "@t3tools/provider-core/server/driver"; +import * as Effect from "effect/Effect"; +import * as Redacted from "effect/Redacted"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { HttpClient, HttpClientRequest, FetchHttpClient } from "effect/http"; +import * as Socket from "effect/socket/Socket"; +import { ChildProcessSpawner } from "effect/process"; +import * as NodeSocket from "@effect/platform-node/NodeSocket"; +import { withCodexAppServerClient } from "./CodexProvider.ts"; + +// A cancelled handshake must not leave an unhandled WebSocket error. +const webSocketConstructor = Layer.effect( + Socket.WebSocketConstructor, + Effect.gen(function* () { + const make = yield* Socket.WebSocketConstructor; + return (url, options) => { + const socket = make(url, options); + socket.addEventListener("error", () => undefined); + return socket; + }; + }), +).pipe(Layer.provide(NodeSocket.layerWebSocketConstructor)); + +const VoiceList = Schema.Struct({ + voices: Schema.Struct({ + v1: Schema.Array(Schema.String), + v3: Schema.optionalKey(Schema.Array(Schema.String)), + defaultV1: Schema.String, + }), +}); +const Claims = Schema.Struct({ + "https://api.openai.com/auth": Schema.optionalKey( + Schema.Struct({ + chatgpt_account_id: Schema.optionalKey(Schema.String), + }), + ), +}); + +const decodeVoiceList = Schema.decodeUnknownEffect(VoiceList); +const decodeClaims = Schema.decodeEffect(Schema.fromJsonString(Claims)); +const isLiveVoiceError = Schema.is(LiveVoiceError); +const encodeSession = Schema.encodeEffect(Schema.fromJsonString(Schema.Json)); +const encodeEvent = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); + +export const makeCodexLiveVoice = Effect.fn("CodexLiveVoice.make")(function* ( + resolve: Effect.Effect< + { + readonly config: CodexSettings; + readonly environment: NodeJS.ProcessEnv; + readonly subscriptionToken?: Redacted.Redacted; + }, + E, + R + >, +) { + const http = yield* HttpClient.HttpClient; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const context = yield* Effect.context(); + const configuration = resolve.pipe( + Effect.provide(context), + Effect.mapError(() => new LiveVoiceError({ reason: "authentication" })), + ); + const read = Effect.fn("CodexLiveVoice.read")( + function* () { + const { config, environment, subscriptionToken } = yield* configuration; + const { client } = yield* withCodexAppServerClient({ + binaryPath: config.binaryPath, + homePath: config.homePath, + launchArgs: config.launchArgs, + environment, + cwd: config.homePath || ".", + }); + let token: Redacted.Redacted; + if (subscriptionToken) { + token = subscriptionToken; + } else { + const account = yield* client.request("account/read", { refreshToken: false }); + if (account.account?.type !== "chatgpt") + return yield* new LiveVoiceError({ reason: "authentication" }); + const auth = yield* client.request("getAuthStatus", { + includeToken: true, + refreshToken: false, + }); + if (auth.authMethod !== "chatgpt" || !auth.authToken) + return yield* new LiveVoiceError({ reason: "unsupported" }); + token = Redacted.make(auth.authToken); + } + const raw = yield* client.raw.request("thread/realtime/listVoices", {}); + const list = yield* decodeVoiceList(raw); + const voices = list.voices.v3?.length ? list.voices.v3 : list.voices.v1; + const claims = subscriptionToken + ? undefined + : yield* decodeClaims( + Buffer.from(Redacted.value(token).split(".")[1] ?? "", "base64url").toString(), + ); + const accountId = claims?.["https://api.openai.com/auth"]?.chatgpt_account_id; + return { + token, + accountId, + voices, + defaultVoice: list.voices.defaultV1, + sharing: subscriptionToken !== undefined, + }; + }, + Effect.scoped, + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.mapError((error) => + isLiveVoiceError(error) ? error : new LiveVoiceError({ reason: "unsupported" }), + ), + ); + return { + options: read().pipe(Effect.map(({ voices, defaultVoice }) => ({ voices, defaultVoice }))), + open: Effect.fn("CodexLiveVoice.open")( + function* (input: Parameters[0]) { + const auth = yield* read(); + if (!auth.voices.includes(input.voice)) + return yield* new LiveVoiceError({ reason: "unsupported" }); + const headers = { + authorization: `Bearer ${Redacted.value(auth.token)}`, + "OpenAI-Alpha": "quicksilver=v2", + originator: "T3 Code", + ...(auth.accountId ? { "ChatGPT-Account-Id": auth.accountId } : {}), + }; + const session = { + model: "gpt-live-1-codex", + instructions: "", + audio: { output: { voice: input.voice } }, + delegation: { type: "client" }, + initial_items: input.context.map(({ role, text }) => ({ + type: "message", + role, + content: [{ type: role === "assistant" ? "output_text" : "input_text", text }], + })), + }; + // Native realtime RPCs start Codex turns outside T3 accounting; T3 must own delegation. + // Preserve the selected connection's subscription credential route; never use API keys. + let request = HttpClientRequest.post( + auth.sharing + ? "https://api.openai.com/v1/live" + : "https://chatgpt.com/backend-api/codex/realtime/calls?intent=quicksilver&architecture=avas", + ).pipe(HttpClientRequest.setHeaders(headers)); + if (auth.sharing) { + const form = new FormData(); + form.set("sdp", input.sdp); + form.set("session", yield* encodeSession(session)); + request = request.pipe(HttpClientRequest.bodyFormData(form)); + } else { + request = request.pipe(HttpClientRequest.bodyJsonUnsafe({ sdp: input.sdp, session })); + } + const response = yield* http + .execute(request) + .pipe( + Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }), + Effect.provideService(HttpClient.TracerPropagationEnabled, false), + ); + if (response.status < 200 || response.status >= 300) + return yield* new LiveVoiceError({ reason: "connection", status: response.status }); + const sdp = yield* response.text; + const location = response.headers.location; + const callId = location?.split("/").filter(Boolean).at(-1); + if (!callId || !/^[A-Za-z0-9_-]+$/.test(callId)) + return yield* new LiveVoiceError({ reason: "connection" }); + const socket = yield* Socket.makeWebSocket(`wss://api.openai.com/v1/live/${callId}`, { + headers, + openTimeout: "15 seconds", + }).pipe(Effect.provide(webSocketConstructor)); + const writer = yield* socket.writer; + const send = Effect.fn("CodexLiveVoice.send")((event: unknown) => + encodeEvent(event).pipe( + Effect.flatMap((encoded) => writer.write(encoded)), + Effect.mapError(() => new LiveVoiceError({ reason: "connection" })), + ), + ); + let connected = false; + yield* Effect.addFinalizer(() => + connected + ? send({ type: "session.close" }).pipe(Effect.timeout("1 second"), Effect.ignore) + : Effect.void, + ); + const { pull } = yield* socket.reader; + connected = true; + const events = Stream.fromPull(Effect.succeed(pull)).pipe( + Stream.map((frame) => + typeof frame === "string" ? frame : new TextDecoder().decode(frame), + ), + Stream.mapError(() => new LiveVoiceError({ reason: "connection" })), + ); + return { sdp, events, send }; + }, + Effect.timeout("30 seconds"), + Effect.mapError((error) => + isLiveVoiceError(error) ? error : new LiveVoiceError({ reason: "connection" }), + ), + ), + } satisfies ProviderLiveVoice; +}); diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index f8a753169b54a..53dc7aaaa8d20 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -1,4 +1,5 @@ import { resolveProviderProbeTimeoutMs } from "@t3tools/provider-core/server/snapshotProbe"; +import { makeCodexLiveVoice } from "../CodexLiveVoice.ts"; import { CodexSettings, ProviderDriverKind } from "@t3tools/contracts"; import * as Digest from "effect/crypto/Digest"; import * as SecureRandom from "effect/crypto/SecureRandom"; @@ -316,6 +317,9 @@ export const CodexDriver: ProviderDriver< }), ), ); + const liveVoice = yield* makeCodexLiveVoice( + Effect.succeed({ config: effectiveConfig, environment: processEnv }), + ); const textGeneration = yield* makeCodexTextGeneration( effectiveConfig, processEnv, @@ -425,6 +429,7 @@ export const CodexDriver: ProviderDriver< consumeResetCredit, orchestrationAdapter, textGeneration, + liveVoice, } satisfies ProviderInstance; }), }; diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index f182245f8840d..5616cbc3e3917 100644 --- a/apps/server/src/provider/Drivers/CodexManagedProvider.ts +++ b/apps/server/src/provider/Drivers/CodexManagedProvider.ts @@ -1,3 +1,5 @@ +import * as Redacted from "effect/Redacted"; +import { makeCodexLiveVoice } from "../CodexLiveVoice.ts"; import { ProviderDriverKind, ProviderSetupError, @@ -275,6 +277,18 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu }), ), ); + const liveVoice = yield* makeCodexLiveVoice( + resolveRuntime.pipe( + Effect.flatMap((effective) => + runtime.auth.access.pipe( + Effect.map((credentials) => ({ + ...effective, + subscriptionToken: Redacted.make(credentials.accessToken), + })), + ), + ), + ), + ); const textGeneration: ProviderInstance["textGeneration"] = { generateCommitMessage: (value) => protect("generateCommitMessage", nativeGeneration.generateCommitMessage(value)), @@ -302,6 +316,7 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu snapshot, orchestrationAdapter, textGeneration, + liveVoice, auth: runtime.auth.controller, snapshotForCwd: (cwd: string) => enabled diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index dae95787afe82..be9e235ef6a90 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -1,3 +1,5 @@ +import * as LiveVoice from "./liveVoice/LiveVoice.ts"; +import * as LiveVoiceHttp from "./liveVoice/http.ts"; import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay"; import * as Random from "effect/Random"; @@ -806,6 +808,7 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(PullRequestHttp.layer), Layer.provide(ProjectHttp.layer), Layer.provide(speechHttpApiLayer), + Layer.provide(LiveVoiceHttp.layer), Layer.provide(ServerHttp.layerServerEnvironmentHttpApi), Layer.provide(WebhookRoute.layer.pipe(Layer.provide(RelayDeliveryProof.layer))), Layer.provide(AuthHttp.layerAuthenticatedAuth), @@ -858,6 +861,7 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(SpeechModelStore.layer), ), ), + Layer.provide(LiveVoice.layer.pipe(Layer.provide(ProjectionStoreV2.layer))), Layer.provide(CloudSpeech.layer.pipe(Layer.provide(SecretResolver.layer))), Layer.provide(layerServerSettings), Layer.provide(PreviewAutomationBroker.layer), diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index eb01b8242e2df..970b1e6a9a3ba 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -1,3 +1,4 @@ +import { LiveVoiceButton } from "./LiveVoiceButton"; import { VoiceInputSetup } from "./VoiceInputSetup"; import { useComposerTypingGuard } from "./useComposerTypingGuard"; import { insertAttachmentReferencesForDraft } from "./composerAttachmentReferences"; @@ -1805,6 +1806,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) isComposerApprovalState || activePendingIsResponding || activePendingProgress?.activeQuestion?.allowCustomAnswer === false; + const [liveVoiceActive, setLiveVoiceActive] = useState(false); const speechInput = useEnvironmentSpeechInput({ environmentId, projectId: activeProjectId, @@ -1854,7 +1856,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const dictationShortcutLabel = useDictationShortcut({ keybindings, speech: speechInput, - disabled: dictationDisabled, + disabled: dictationDisabled || liveVoiceActive, terminalOpen, modelPickerOpen: isComposerModelPickerOpen, ...(backgroundThread ? { targetRef: composerFormRef } : {}), @@ -3722,6 +3724,18 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) keepFullHistory={props.keepFullHistory} onToggleKeepFullHistory={props.onToggleKeepFullHistory} > + {selectedProvider === "codex" && + routeKind === "server" && + activeThreadId && + !multipleModelSelections ? ( + + ) : null} {speechInput.available ? ( <> void speechInput.start()} onStop={() => void speechInput.stop()} onCancel={() => speechInput.cancel()} @@ -3739,7 +3753,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) void speechInput.start()} /> diff --git a/apps/web/src/components/chat/LiveVoiceButton.tsx b/apps/web/src/components/chat/LiveVoiceButton.tsx new file mode 100644 index 0000000000000..08d50c181c002 --- /dev/null +++ b/apps/web/src/components/chat/LiveVoiceButton.tsx @@ -0,0 +1,217 @@ +import * as Option from "effect/Option"; +import type { EnvironmentId, ThreadId, LiveVoiceOptions, LiveVoiceState } from "@t3tools/contracts"; +import { + liveVoiceOptions, + startLiveVoice, + pollLiveVoice, + stopLiveVoice, +} from "@t3tools/client-runtime/voice-input"; +import { HeadphonesIcon, MicIcon, MicOffIcon, PhoneOffIcon } from "lucide-react"; +import { useEffect, useRef, useState } from "react"; +import { usePreparedConnection } from "../../state/session"; +import { chooseRecordingMicrophone } from "../../speech/browserMicrophones"; +import { runEnvironmentSpeech } from "../../speech/runEnvironmentSpeech"; +import { BrowserLiveVoice, type LiveVoicePhase } from "../../speech/browserLiveVoice"; +import { Button } from "../ui/button"; +import { + Dialog, + DialogPopup, + DialogHeader, + DialogTitle, + DialogDescription, + DialogPanel, + DialogFooter, +} from "../ui/dialog"; +import { Select, SelectTrigger, SelectValue, SelectPopup, SelectItem } from "../ui/select"; + +export function LiveVoiceButton(props: { + environmentId: EnvironmentId; + threadId: ThreadId; + disabled: boolean; + onActiveChange(active: boolean): void; +}) { + const { disabled, onActiveChange } = props; + const prepared = Option.getOrNull(usePreparedConnection(props.environmentId)); + const [open, setOpen] = useState(false); + const [options, setOptions] = useState(null); + const [voice, setVoice] = useState(""); + const [phase, setPhase] = useState("idle"); + const [error, setError] = useState(null); + const [muted, setMuted] = useState(false); + const [transcripts, setTranscripts] = useState([]); + const client = useRef(null); + const loading = useRef(0); + const active = phase === "connecting" || phase === "connected"; + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- Release microphone and peer resources when the selected environment or conversation unmounts. + useEffect(() => { + if (disabled || !prepared) client.current?.stop(); + return () => { + loading.current++; + client.current?.stop(false); + onActiveChange(false); + }; + }, [disabled, onActiveChange, prepared]); + + async function changeOpen(next: boolean) { + setOpen(next); + const request = ++loading.current; + if (!next) { + client.current?.stop(); + props.onActiveChange(false); + return; + } + setError(null); + setOptions(null); + if (!prepared) { + setError("Reconnect this environment before starting voice."); + return; + } + try { + const result = await runEnvironmentSpeech(liveVoiceOptions(prepared, props.threadId)); + if (loading.current !== request) return; + setOptions(result); + setVoice((previous) => (result.voices.includes(previous) ? previous : result.defaultVoice)); + } catch (cause) { + if (loading.current === request) + setError( + cause instanceof Error + ? cause.message + : "Subscription voice is unavailable for this account.", + ); + } + } + function start() { + if (!prepared || !voice) return; + setMuted(false); + setError(null); + client.current?.stop(false); + const connection = new BrowserLiveVoice({ + open: (sessionId, selected, sdp, signal) => + runEnvironmentSpeech( + startLiveVoice(prepared, { threadId: props.threadId, sessionId, voice: selected, sdp }), + { signal }, + ), + poll: (id, revision, signal) => + runEnvironmentSpeech(pollLiveVoice(prepared, id, revision), { signal }), + stop: (id) => runEnvironmentSpeech(stopLiveVoice(prepared, id)), + getMicrophoneId: chooseRecordingMicrophone, + onTranscript: setTranscripts, + onPhase: (next, message) => { + setPhase(next); + setError(message); + props.onActiveChange(next === "connected" || next === "connecting"); + }, + }); + client.current = connection; + void connection.start(voice); + } + return ( + <> + + void changeOpen(next)}> + + + ChatGPT live voice + + Use the ChatGPT account connected to this conversation. Voice requests appear as + normal T3 tasks. + + + +
+
+ Voice + +
+

+ {phase === "connecting" + ? "Connecting…" + : phase === "connected" + ? muted + ? "Microphone muted" + : "Listening" + : "Microphone is off"} +

+ {error ? ( +

+ {error} +

+ ) : null} +
+ {transcripts.map((item) => ( +

+ {item.role === "user" ? "You" : "Voice"}: + {item.text} +

+ ))} +
+

+ This transcript is shown for the current voice session. Ending voice leaves any + coding tasks running in the conversation. +

+
+ + + {active ? ( + <> + + + + ) : ( + + )} + + + + + ); +} diff --git a/apps/web/src/speech/browserLiveVoice.test.ts b/apps/web/src/speech/browserLiveVoice.test.ts new file mode 100644 index 0000000000000..d88dd22783755 --- /dev/null +++ b/apps/web/src/speech/browserLiveVoice.test.ts @@ -0,0 +1,59 @@ +import { expect, it, vi } from "vite-plus/test"; +import { BrowserLiveVoice } from "./browserLiveVoice"; + +it("releases a microphone granted after the user cancels without opening a remote call", async () => { + let grant!: (stream: MediaStream) => void; + const permission = new Promise((resolve) => { + grant = resolve; + }); + const stop = vi.fn(); + const open = vi.fn(); + vi.stubGlobal("navigator", { mediaDevices: { getUserMedia: () => permission } }); + vi.stubGlobal("RTCPeerConnection", vi.fn()); + try { + const voice = new BrowserLiveVoice({ + open, + poll: vi.fn(), + stop: vi.fn(), + onPhase: vi.fn(), + onTranscript: vi.fn(), + }); + const starting = voice.start("sol"); + voice.stop(); + grant({ getTracks: () => [{ stop }] } as unknown as MediaStream); + await starting; + expect(stop).toHaveBeenCalledOnce(); + expect(open).not.toHaveBeenCalled(); + } finally { + vi.unstubAllGlobals(); + } +}); + +it("does not request microphone access when cancelled while resolving the preferred device", async () => { + let choose!: (id: string) => void; + const preference = new Promise((resolve) => { + choose = resolve; + }); + const capture = vi.fn(); + const open = vi.fn(); + vi.stubGlobal("navigator", { mediaDevices: { getUserMedia: capture } }); + vi.stubGlobal("RTCPeerConnection", vi.fn()); + try { + const voice = new BrowserLiveVoice({ + open, + poll: vi.fn(), + stop: vi.fn(), + getMicrophoneId: () => preference, + onPhase: vi.fn(), + onTranscript: vi.fn(), + }); + const starting = voice.start("aube"); + voice.stop(); + choose("preferred-device"); + await starting; + expect(capture).not.toHaveBeenCalled(); + expect(open).not.toHaveBeenCalled(); + } finally { + vi.unstubAllGlobals(); + } +}); diff --git a/apps/web/src/speech/browserLiveVoice.ts b/apps/web/src/speech/browserLiveVoice.ts new file mode 100644 index 0000000000000..dd33c0534ed08 --- /dev/null +++ b/apps/web/src/speech/browserLiveVoice.ts @@ -0,0 +1,161 @@ +import * as SecureRandom from "effect/crypto/SecureRandom"; +import type { LiveVoiceState } from "@t3tools/contracts"; + +export type LiveVoicePhase = "idle" | "connecting" | "connected" | "error"; +export type LiveVoiceBrowserDependencies = { + open: ( + sessionId: string, + voice: string, + sdp: string, + signal: AbortSignal, + ) => Promise<{ sdp: string }>; + poll: (sessionId: string, revision: number, signal: AbortSignal) => Promise; + stop: (sessionId: string) => Promise; + getMicrophoneId?: () => string | Promise; + onPhase: (phase: LiveVoicePhase, error: string | null) => void; + onTranscript: (transcripts: LiveVoiceState["transcripts"]) => void; +}; + +export class BrowserLiveVoice { + private abort: AbortController | null = null; + private stream: MediaStream | null = null; + private peer: RTCPeerConnection | null = null; + private audio: HTMLAudioElement | null = null; + private sessionId: string | null = null; + private generation = 0; + + constructor(private readonly dependencies: LiveVoiceBrowserDependencies) {} + + stop(notify = true) { + this.generation++; + this.abort?.abort(); + this.abort = null; + this.stream?.getTracks().forEach((track) => track.stop()); + this.stream = null; + if (this.peer) { + this.peer.onconnectionstatechange = null; + this.peer.ontrack = null; + this.peer.close(); + this.peer = null; + } + if (this.audio) { + this.audio.pause(); + this.audio.srcObject = null; + this.audio = null; + } + const id = this.sessionId; + this.sessionId = null; + if (id) void this.dependencies.stop(id).catch(() => undefined); + if (notify) this.dependencies.onPhase("idle", null); + } + + mute(muted: boolean) { + this.stream?.getAudioTracks().forEach((track) => { + track.enabled = !muted; + }); + } + + private fail(error: unknown) { + this.stop(false); + this.dependencies.onPhase( + "error", + error instanceof Error ? error.message : "Voice could not connect. Try again.", + ); + } + + async start(voice: string) { + this.stop(false); + const generation = this.generation; + const abort = new AbortController(); + this.abort = abort; + this.dependencies.onTranscript([]); + this.dependencies.onPhase("connecting", null); + try { + if (!navigator.mediaDevices?.getUserMedia || typeof RTCPeerConnection === "undefined") + throw new Error( + "Live voice needs a browser with microphone access over HTTPS or localhost.", + ); + const microphone = this.dependencies.getMicrophoneId + ? await this.dependencies.getMicrophoneId() + : ""; + if (generation !== this.generation) return; + const stream = await navigator.mediaDevices.getUserMedia({ + audio: { + echoCancellation: true, + noiseSuppression: true, + ...(microphone ? { deviceId: { exact: microphone } } : {}), + }, + }); + if (generation !== this.generation) { + stream.getTracks().forEach((track) => track.stop()); + return; + } + this.stream = stream; + const peer = new RTCPeerConnection(); + this.peer = peer; + const audio = new Audio(); + audio.autoplay = true; + this.audio = audio; + peer.ontrack = (event) => { + if (generation !== this.generation) return; + audio.srcObject = new MediaStream([event.track]); + void audio.play().catch(() => { + if (generation === this.generation) + this.fail( + new Error("The browser blocked voice playback. End voice and try Start voice again."), + ); + }); + }; + peer.onconnectionstatechange = () => { + if (generation !== this.generation) return; + if (peer.connectionState === "connected") this.dependencies.onPhase("connected", null); + if (peer.connectionState === "failed" || peer.connectionState === "closed") + this.fail(new Error("The voice connection ended. Your coding tasks are still in T3.")); + }; + stream.getAudioTracks().forEach((track) => peer.addTrack(track, stream)); + peer.createDataChannel("oai-events"); + await peer.setLocalDescription(await peer.createOffer()); + await new Promise((resolve, reject) => { + const finish = () => { + clearTimeout(timeout); + peer.removeEventListener("icegatheringstatechange", changed); + abort.signal.removeEventListener("abort", cancelled); + resolve(); + }; + const changed = () => { + if (peer.iceGatheringState === "complete") finish(); + }; + const cancelled = () => { + finish(); + reject(new DOMException("Voice start cancelled", "AbortError")); + }; + const timeout = setTimeout(finish, 5000); + peer.addEventListener("icegatheringstatechange", changed); + abort.signal.addEventListener("abort", cancelled, { once: true }); + changed(); + }); + if (generation !== this.generation) return; + const sdp = peer.localDescription?.sdp; + if (!sdp) throw new Error("The browser could not prepare a voice connection."); + const id = SecureRandom.uuidV4Unsafe(); + this.sessionId = id; + const answer = await this.dependencies.open(id, voice, sdp, abort.signal); + if (generation !== this.generation) return; + await peer.setRemoteDescription({ type: "answer", sdp: answer.sdp }); + let revision = -1; + while (generation === this.generation) { + const state = await this.dependencies.poll(id, revision, abort.signal); + if (generation !== this.generation) return; + revision = state.revision; + this.dependencies.onTranscript(state.transcripts); + if (state.phase !== "connected") { + this.stop(false); + this.dependencies.onPhase(state.error ? "error" : "idle", state.error); + return; + } + } + } catch (error) { + if (generation === this.generation) this.fail(error); + } + } +} diff --git a/docs/user/voice-input.md b/docs/user/voice-input.md index 592e110c896cc..7780a5cb0d09f 100644 --- a/docs/user/voice-input.md +++ b/docs/user/voice-input.md @@ -10,6 +10,29 @@ On iPhone and Android, voice input transcribes on the device by default. To opt to a named environment on that phone, open **Settings → Voice → Transcribe on**. See [Voice input on mobile](./composer.md#voice-input-on-mobile). +## Live conversations with ChatGPT + +On web and desktop, open a saved Codex conversation and choose the headphones +button beside the composer. Pick a voice and choose **Start voice**. This uses +the ChatGPT account already connected to that conversation, not a separate voice +account, an API key, or a downloaded transcription model. + +Your microphone starts only after **Start voice** and browser permission. Audio +goes to OpenAI. Use **Mute** to pause your microphone, or **End voice** to close +the call. Closing the voice panel or changing conversations also ends the call. +Remote browsers need HTTPS; localhost also works. + +Requests handed to the coding agent appear in the same conversation and use its +normal permissions. Use T3's existing approval controls when needed. Ending voice +does not cancel coding tasks that are already running. The voice transcript is +shown for the current call; coding requests and results remain in chat history. + +Live voice is experimental and depends on the account connection and Codex +version. A successful coding sign-in does not guarantee voice access. A rejected +connection reports an error instead of switching accounts, models, or billing +routes. The native iPhone and Android apps do not yet have these live-call controls; +their dictation remains unchanged. + ## Choose where audio goes On web and desktop, you choose where transcription runs. diff --git a/packages/client-runtime/src/voice-input/index.ts b/packages/client-runtime/src/voice-input/index.ts index 5283d64eb0dc3..3db5b2769d452 100644 --- a/packages/client-runtime/src/voice-input/index.ts +++ b/packages/client-runtime/src/voice-input/index.ts @@ -39,3 +39,5 @@ export { type VoiceTranscriptionErrorCode, type VoiceTranscriptionOptions, } from "./transcription.ts"; + +export * from "./liveVoice.ts"; diff --git a/packages/client-runtime/src/voice-input/liveVoice.ts b/packages/client-runtime/src/voice-input/liveVoice.ts new file mode 100644 index 0000000000000..44112fe3c606f --- /dev/null +++ b/packages/client-runtime/src/voice-input/liveVoice.ts @@ -0,0 +1,77 @@ +import type { LiveVoiceStart, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; +import * as ManagedRelay from "../relay/managedRelay.ts"; +import type { PreparedConnection } from "../connection/model.ts"; +import { executeAuthenticatedEnvironmentHttpRequest } from "../state/environmentHttpAuth.ts"; + +export const liveVoiceOptions = Effect.fn("liveVoice.options")(function* ( + prepared: PreparedConnection, + threadId: ThreadId, +) { + return yield* executeAuthenticatedEnvironmentHttpRequest({ + prepared, + group: "liveVoice", + method: "GET", + timeoutMs: 30_000, + signer: yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner), + remoteAuthorization: yield* Effect.serviceOption( + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization, + ), + url: (urls) => urls.options({ params: { threadId } }), + request: ({ client, headers }) => client.options({ params: { threadId }, headers }), + }); +}); +export const startLiveVoice = Effect.fn("liveVoice.start")(function* ( + prepared: PreparedConnection, + payload: LiveVoiceStart, +) { + return yield* executeAuthenticatedEnvironmentHttpRequest({ + prepared, + group: "liveVoice", + method: "POST", + timeoutMs: 40_000, + signer: yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner), + remoteAuthorization: yield* Effect.serviceOption( + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization, + ), + url: (urls) => urls.start(), + request: ({ client, headers }) => client.start({ payload, headers }), + }); +}); +export const pollLiveVoice = Effect.fn("liveVoice.poll")(function* ( + prepared: PreparedConnection, + sessionId: string, + revision: number, +) { + return yield* executeAuthenticatedEnvironmentHttpRequest({ + prepared, + group: "liveVoice", + method: "GET", + timeoutMs: 25_000, + signer: yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner), + remoteAuthorization: yield* Effect.serviceOption( + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization, + ), + url: (urls) => urls.poll({ params: { sessionId }, query: { revision } }), + request: ({ client, headers }) => + client.poll({ params: { sessionId }, query: { revision }, headers }), + }); +}); +export const stopLiveVoice = Effect.fn("liveVoice.stop")(function* ( + prepared: PreparedConnection, + sessionId: string, +) { + return yield* executeAuthenticatedEnvironmentHttpRequest({ + prepared, + group: "liveVoice", + method: "POST", + timeoutMs: 10_000, + signer: yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner), + remoteAuthorization: yield* Effect.serviceOption( + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization, + ), + url: (urls) => urls.stop({ params: { sessionId } }), + request: ({ client, headers }) => client.stop({ params: { sessionId }, headers }), + }); +}); diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index a9c635486db7d..fbdb41ebfcd80 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -1,3 +1,10 @@ +import { + LiveVoiceId, + LiveVoiceOptions, + LiveVoiceStart, + LiveVoiceState, + LiveVoiceError, +} from "./liveVoice.ts"; import * as Context from "effect/Context"; import type * as DateTime from "effect/DateTime"; import * as Schema from "effect/Schema"; @@ -796,6 +803,41 @@ export class EnvironmentVoiceBodyLimit extends HttpApiMiddleware.Service()("LiveVoiceError", { + reason: Schema.Literals([ + "unavailable", + "authentication", + "unsupported", + "connection", + "busy", + "expired", + "thread-changed", + ]), + status: Schema.optionalKey(Schema.Number), +}) { + override get message() { + switch (this.reason) { + case "authentication": + return "Live voice requires the ChatGPT account connected to this provider. Reconnect it in provider settings."; + case "unsupported": + return "This account connection or Codex version does not support subscription voice."; + case "connection": + return `The subscription voice service could not connect${this.status ? ` (HTTP ${this.status})` : ""}. No API-key fallback was used.`; + case "busy": + return "This conversation already has an active voice session."; + case "expired": + return "The voice session has ended. Start a new session."; + case "thread-changed": + return "The conversation or account changed. Start voice again to use the current connection."; + default: + return "Live voice is available in saved Codex conversations with a connected ChatGPT account."; + } + } +} diff --git a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts index 59e5eb28faf94..bbc925fbb367c 100644 --- a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts +++ b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts @@ -72,11 +72,14 @@ const handleMethod = (message: Record) => { } case "account/read": { respond(message.id as number | string, { - account: { - type: "chatgpt", - email: "mock@example.com", - planType: "plus", - }, + account: + process.env.CODEX_APP_SERVER_TEST_ACCOUNT === "apiKey" + ? { type: "apiKey" } + : { + type: "chatgpt", + email: "mock@example.com", + planType: "plus", + }, requiresOpenaiAuth: false, }); return; diff --git a/packages/provider-core/src/server/driver.ts b/packages/provider-core/src/server/driver.ts index ac5ea46f39f16..c4bd6ca9c5f2f 100644 --- a/packages/provider-core/src/server/driver.ts +++ b/packages/provider-core/src/server/driver.ts @@ -1,6 +1,8 @@ // Context tags are runtime singletons; drivers need multiple instances. import type { ProviderConsumeResetCreditOutcome, + LiveVoiceError, + LiveVoiceOptions, AcpRegistryListSessionsResult, AcpRegistryListProvidersResult, AcpRegistryOperationError, @@ -14,6 +16,7 @@ import type { import type * as Effect from "effect/Effect"; import type * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; +import type * as Stream from "effect/Stream"; import type * as ProviderAdapter from "./ProviderAdapter.ts"; import type { ProviderAuthController } from "./auth.ts"; @@ -37,6 +40,23 @@ export type ProviderWorkspaceSnapshot = ServerProvider & readonly workspaceModelsPending?: boolean; }; +export interface ProviderLiveVoice { + readonly options: Effect.Effect; + readonly open: (input: { + readonly sdp: string; + readonly voice: string; + readonly context: ReadonlyArray<{ readonly role: "user" | "assistant"; readonly text: string }>; + }) => Effect.Effect< + { + readonly sdp: string; + readonly events: Stream.Stream; + readonly send: (event: unknown) => Effect.Effect; + }, + LiveVoiceError, + Scope.Scope + >; +} + export interface ProviderInstance { readonly instanceId: ProviderInstanceId; readonly driverKind: ProviderDriverKind; @@ -57,6 +77,7 @@ export interface ProviderInstance { ) => Effect.Effect; readonly orchestrationAdapter: ProviderAdapter.ProviderAdapterV2["Service"]; readonly textGeneration: ProviderTextGeneration; + readonly liveVoice?: ProviderLiveVoice; readonly auth?: ProviderAuthController; readonly acpSessionManagement?: { readonly listSessions: (input: {