feat(codex): add provider-native voice mode

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,
   }),
 );

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.

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,

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);
     },

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);
       },

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);
     },

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<void> =>
       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;

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,

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<string, unknown>,
+  run: (
+    runtime: Effect.Success<ReturnType<typeof makeCodexSessionRuntime>>,
+    scriptPath: string,
+  ) => Effect.Effect<void, CodexSessionRuntimeError, Scope.Scope>,
+) {
+  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);
+        }
+      }),
+  );
+});

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");