roboomp
25c4ec07ed
Centralized URL-aware proxy resolution so Codex WebSocket transports share standard proxy environment fallbacks and NO_PROXY handling. Fixes #6770
423 lines
14 KiB
TypeScript
423 lines
14 KiB
TypeScript
import { type AuthStorage, isAuthRetryableError, type OAuthAccess, withOAuthAccess } from "@oh-my-pi/pi-ai";
|
|
import { getProxyForUrl, wrapFetchForProxy } from "@oh-my-pi/pi-ai/utils/proxy";
|
|
import {
|
|
CODEX_BASE_URL,
|
|
CODEX_CLIENT_VERSION,
|
|
getCodexAccountId,
|
|
OPENAI_HEADERS,
|
|
} from "@oh-my-pi/pi-catalog/wire/codex";
|
|
import { LiveWebRtcPeer } from "@oh-my-pi/pi-natives";
|
|
import { generateCodexAttestation } from "./attestation";
|
|
import {
|
|
buildLiveSessionPayload,
|
|
type LiveClientMessage,
|
|
type LiveServerEvent,
|
|
parseLiveServerEvent,
|
|
} from "./protocol";
|
|
|
|
const SIGNALING_URL = `${CODEX_BASE_URL}/codex/realtime/calls?intent=quicksilver&architecture=avas`;
|
|
const MAX_ERROR_BODY_LENGTH = 2_048;
|
|
const SIDEBAND_CONNECT_ATTEMPTS = 5;
|
|
const SIDEBAND_CONNECT_TIMEOUT_MS = 15_000;
|
|
const LIVE_PROVIDER = "openai-codex";
|
|
const LIVE_ORIGINATOR = "Codex Desktop";
|
|
const LIVE_CALL_ID_PATTERN = /^rtc_[\w-]+$/;
|
|
|
|
type Lifecycle = "idle" | "connecting" | "connected" | "closing" | "closed";
|
|
|
|
interface LiveSignalingResult {
|
|
answer: string;
|
|
callId: string;
|
|
access: OAuthAccess;
|
|
attestation: string | undefined;
|
|
}
|
|
|
|
class LiveSignalingError extends Error {
|
|
status: number;
|
|
errorMessage: string;
|
|
|
|
constructor(status: number, message: string) {
|
|
super(message);
|
|
this.name = "LiveSignalingError";
|
|
this.status = status;
|
|
this.errorMessage = message;
|
|
}
|
|
}
|
|
|
|
/** Callbacks emitted by the live WebRTC transport. */
|
|
export interface LiveTransportCallbacks {
|
|
onEvent(event: LiveServerEvent): void;
|
|
onOutputLevel(level: number): void;
|
|
}
|
|
|
|
/** Configuration required to establish a Codex live call. */
|
|
export interface LiveTransportOptions {
|
|
authStorage: AuthStorage;
|
|
sessionId: string;
|
|
instructions: string;
|
|
voice: string;
|
|
callbacks: LiveTransportCallbacks;
|
|
signal?: AbortSignal;
|
|
}
|
|
|
|
/** Extracts the server-assigned `rtc_*` call ID from a signaling Location header. */
|
|
export function parseLiveCallId(location: string | null): string | undefined {
|
|
if (!location) return undefined;
|
|
return location
|
|
.split("?", 1)[0]
|
|
?.split("/")
|
|
.find(segment => LIVE_CALL_ID_PATTERN.test(segment));
|
|
}
|
|
|
|
/** Builds the Frameless Bidi sideband WebSocket URL for an accepted Codex call. */
|
|
export function buildLiveSidebandUrl(callId: string): string {
|
|
const url = new URL(`https://api.openai.com/v1/live/${encodeURIComponent(callId)}`);
|
|
url.protocol = "wss:";
|
|
return url.toString();
|
|
}
|
|
|
|
function liveSessionHeaders(
|
|
access: OAuthAccess,
|
|
sessionId: string,
|
|
realtimeSessionId: string,
|
|
attestation: string | undefined,
|
|
): Record<string, string> {
|
|
const headers: Record<string, string> = {
|
|
Authorization: `Bearer ${access.accessToken}`,
|
|
"OpenAI-Alpha": "quicksilver=v2",
|
|
"User-Agent": `Codex Desktop/${CODEX_CLIENT_VERSION}`,
|
|
"x-session-id": realtimeSessionId,
|
|
[OPENAI_HEADERS.ORIGINATOR]: LIVE_ORIGINATOR,
|
|
[OPENAI_HEADERS.VERSION]: CODEX_CLIENT_VERSION,
|
|
[OPENAI_HEADERS.SCOPED_SESSION_ID]: sessionId,
|
|
[OPENAI_HEADERS.THREAD_ID]: sessionId,
|
|
};
|
|
const accountId = access.accountId ?? getCodexAccountId(access.accessToken);
|
|
if (accountId) headers[OPENAI_HEADERS.ACCOUNT_ID] = accountId;
|
|
if (attestation) headers["x-oai-attestation"] = attestation;
|
|
return headers;
|
|
}
|
|
|
|
function boundedErrorBody(body: string, statusText: string): string {
|
|
const normalized = body.trim().replaceAll(/\s+/g, " ");
|
|
if (!normalized) return statusText || "empty response body";
|
|
if (normalized.length <= MAX_ERROR_BODY_LENGTH) return normalized;
|
|
return `${normalized.slice(0, MAX_ERROR_BODY_LENGTH)}…`;
|
|
}
|
|
|
|
function isAuthError(error: unknown): boolean {
|
|
return isAuthRetryableError(error);
|
|
}
|
|
|
|
function abortReason(signal: AbortSignal | undefined): Error {
|
|
if (signal?.reason instanceof Error) return signal.reason;
|
|
return new DOMException("Live connection aborted", "AbortError");
|
|
}
|
|
|
|
/** Native WebRTC transport for a Codex Frameless Bidi live session. */
|
|
export class CodexLiveTransport {
|
|
readonly #options: LiveTransportOptions;
|
|
#peer: LiveWebRtcPeer | undefined;
|
|
readonly #realtimeSessionId = crypto.randomUUID();
|
|
#sideband: Bun.WebSocket | undefined;
|
|
#state: Lifecycle = "idle";
|
|
#connectPromise: Promise<void> | undefined;
|
|
#closePromise: Promise<void> | undefined;
|
|
#sendTail: Promise<void> = Promise.resolve();
|
|
#muted = false;
|
|
#unexpectedFailureReported = false;
|
|
readonly #abortListener: () => void;
|
|
|
|
constructor(options: LiveTransportOptions) {
|
|
this.#options = options;
|
|
this.#abortListener = () => {
|
|
void this.close();
|
|
};
|
|
if (!options.signal?.aborted) options.signal?.addEventListener("abort", this.#abortListener, { once: true });
|
|
}
|
|
|
|
/** Establish the native peer, perform Codex signaling, and wait for the data channel. */
|
|
connect(): Promise<void> {
|
|
if (this.#state === "connected") return Promise.resolve();
|
|
if (this.#connectPromise) return this.#connectPromise;
|
|
if (this.#state === "closing" || this.#state === "closed")
|
|
return Promise.reject(new Error("Live transport is closed"));
|
|
if (this.#options.signal?.aborted) return Promise.reject(abortReason(this.#options.signal));
|
|
this.#state = "connecting";
|
|
const operation = this.#connect().catch(async error => {
|
|
await this.close();
|
|
throw error;
|
|
});
|
|
this.#connectPromise = operation;
|
|
return operation;
|
|
}
|
|
|
|
async #connect(): Promise<void> {
|
|
const peer = new LiveWebRtcPeer(
|
|
(error, payload) => {
|
|
if (error) {
|
|
this.#handlePeerFailure(error.message);
|
|
} else {
|
|
this.#handleServerEvent(payload);
|
|
}
|
|
},
|
|
(error, level) => {
|
|
if (error) {
|
|
this.#handlePeerFailure(error.message);
|
|
} else {
|
|
this.#handleOutputLevel(level);
|
|
}
|
|
},
|
|
(error, message) => this.#handlePeerFailure(error?.message ?? message),
|
|
);
|
|
this.#peer = peer;
|
|
const offer = await peer.createOffer();
|
|
if (this.#state !== "connecting") throw abortReason(this.#options.signal);
|
|
const signaling = await this.#signal(offer);
|
|
await peer.acceptAnswer(signaling.answer);
|
|
peer.setMuted(this.#muted);
|
|
await peer.waitForOpen();
|
|
if (this.#state !== "connecting") throw abortReason(this.#options.signal);
|
|
await this.#connectSideband(signaling.callId, signaling.access, signaling.attestation);
|
|
if (this.#state !== "connecting") throw abortReason(this.#options.signal);
|
|
this.#state = "connected";
|
|
}
|
|
|
|
async #signal(offer: string): Promise<LiveSignalingResult> {
|
|
const attestation = await generateCodexAttestation();
|
|
return await withOAuthAccess(
|
|
this.#options.authStorage,
|
|
LIVE_PROVIDER,
|
|
access => this.#signalWithAccess(offer, access, attestation),
|
|
{
|
|
sessionId: this.#options.sessionId,
|
|
signal: this.#options.signal,
|
|
isAuthError,
|
|
missingAccessMessage: "No Codex OAuth credential is available for a live call.",
|
|
},
|
|
);
|
|
}
|
|
|
|
async #signalWithAccess(
|
|
offer: string,
|
|
access: OAuthAccess,
|
|
attestation: string | undefined,
|
|
): Promise<LiveSignalingResult> {
|
|
const headers = new Headers({
|
|
...liveSessionHeaders(access, this.#options.sessionId, this.#realtimeSessionId, attestation),
|
|
Accept: "*/*",
|
|
"Content-Type": "application/json",
|
|
});
|
|
const fetchImpl = wrapFetchForProxy(fetch, LIVE_PROVIDER);
|
|
const response = await fetchImpl(SIGNALING_URL, {
|
|
method: "POST",
|
|
headers,
|
|
body: JSON.stringify({
|
|
sdp: offer,
|
|
session: buildLiveSessionPayload(this.#options.instructions, this.#options.voice),
|
|
}),
|
|
signal: this.#options.signal,
|
|
});
|
|
const responseBody = await response.text();
|
|
if (!response.ok) {
|
|
const detail = boundedErrorBody(responseBody, response.statusText);
|
|
throw new LiveSignalingError(response.status, `Codex live signaling failed (${response.status}): ${detail}`);
|
|
}
|
|
const answer = responseBody;
|
|
if (!answer.trim())
|
|
throw new LiveSignalingError(response.status, "Codex live signaling returned an empty SDP answer");
|
|
const callId = parseLiveCallId(response.headers.get("location"));
|
|
if (!callId) {
|
|
throw new LiveSignalingError(response.status, "Codex live signaling returned no valid call ID");
|
|
}
|
|
return { answer, callId, access, attestation };
|
|
}
|
|
|
|
async #connectSideband(callId: string, access: OAuthAccess, attestation: string | undefined): Promise<void> {
|
|
let failure = new Error("Codex live sideband connection failed");
|
|
for (let attempt = 0; attempt < SIDEBAND_CONNECT_ATTEMPTS; attempt++) {
|
|
try {
|
|
await this.#openSideband(callId, access, attestation);
|
|
return;
|
|
} catch (cause) {
|
|
failure = cause instanceof Error ? cause : new Error(String(cause));
|
|
if (this.#options.signal?.aborted) throw abortReason(this.#options.signal);
|
|
if (attempt + 1 < SIDEBAND_CONNECT_ATTEMPTS) await Bun.sleep(200 * 2 ** attempt);
|
|
}
|
|
}
|
|
throw failure;
|
|
}
|
|
|
|
async #openSideband(callId: string, access: OAuthAccess, attestation: string | undefined): Promise<void> {
|
|
const url = buildLiveSidebandUrl(callId);
|
|
const options = {
|
|
headers: liveSessionHeaders(access, this.#options.sessionId, this.#realtimeSessionId, attestation),
|
|
proxy: getProxyForUrl(LIVE_PROVIDER, new URL(url)),
|
|
} satisfies Bun.WebSocketOptions;
|
|
const socket: Bun.WebSocket = Reflect.construct(WebSocket, [url, options]);
|
|
socket.binaryType = "nodebuffer";
|
|
const { promise, resolve, reject } = Promise.withResolvers<void>();
|
|
let opened = false;
|
|
let settled = false;
|
|
let timeout: NodeJS.Timeout | undefined;
|
|
const cleanup = (): void => {
|
|
if (timeout) {
|
|
clearTimeout(timeout);
|
|
timeout = undefined;
|
|
}
|
|
this.#options.signal?.removeEventListener("abort", onAbort);
|
|
};
|
|
const rejectConnect = (error: Error): void => {
|
|
if (settled) return;
|
|
settled = true;
|
|
cleanup();
|
|
reject(error);
|
|
};
|
|
const onAbort = (): void => {
|
|
socket.close(1000, "aborted");
|
|
rejectConnect(abortReason(this.#options.signal));
|
|
};
|
|
socket.onopen = () => {
|
|
if (settled) {
|
|
socket.close(1000, "stale");
|
|
return;
|
|
}
|
|
opened = true;
|
|
settled = true;
|
|
cleanup();
|
|
this.#sideband = socket;
|
|
resolve();
|
|
};
|
|
socket.onmessage = event => {
|
|
if (typeof event.data !== "string") {
|
|
this.#reportFailure("Codex live sideband returned an unexpected binary frame.");
|
|
return;
|
|
}
|
|
this.#handleSidebandEvent(event.data);
|
|
};
|
|
socket.onerror = event => {
|
|
const detail = event instanceof ErrorEvent && event.message ? `: ${event.message}` : "";
|
|
if (!opened) {
|
|
rejectConnect(new Error(`Codex live sideband connection failed${detail}`));
|
|
socket.close(1011, "connection failed");
|
|
return;
|
|
}
|
|
this.#reportFailure(`Codex live sideband failed${detail}`);
|
|
};
|
|
socket.onclose = event => {
|
|
if (!opened) {
|
|
rejectConnect(new Error(`Codex live sideband closed before connecting (${event.code})`));
|
|
return;
|
|
}
|
|
if (this.#sideband !== socket) return;
|
|
this.#sideband = undefined;
|
|
if (this.#state === "connecting" || this.#state === "connected") {
|
|
const detail = event.reason ? `: ${event.reason}` : "";
|
|
this.#reportFailure(`Codex live sideband closed (${event.code})${detail}`);
|
|
}
|
|
};
|
|
if (this.#options.signal?.aborted) {
|
|
onAbort();
|
|
} else {
|
|
this.#options.signal?.addEventListener("abort", onAbort, { once: true });
|
|
timeout = setTimeout(() => {
|
|
socket.close(1000, "connect timeout");
|
|
rejectConnect(new Error("Codex live sideband connection timed out"));
|
|
}, SIDEBAND_CONNECT_TIMEOUT_MS);
|
|
timeout.unref?.();
|
|
}
|
|
await promise;
|
|
}
|
|
|
|
#handleSidebandEvent(payload: string): void {
|
|
if (this.#state === "closing" || this.#state === "closed") return;
|
|
const event = parseLiveServerEvent(payload);
|
|
if (!event) return;
|
|
try {
|
|
this.#options.callbacks.onEvent(event);
|
|
} catch {}
|
|
}
|
|
|
|
#handleServerEvent(payload: string): void {
|
|
if (this.#state === "closing" || this.#state === "closed") return;
|
|
const event = parseLiveServerEvent(payload);
|
|
if (!event || (this.#sideband?.readyState === WebSocket.OPEN && event.type !== "error")) return;
|
|
try {
|
|
this.#options.callbacks.onEvent(event);
|
|
} catch {}
|
|
}
|
|
|
|
#handleOutputLevel(level: number): void {
|
|
if (this.#state !== "connected" || !Number.isFinite(level)) return;
|
|
try {
|
|
this.#options.callbacks.onOutputLevel(Math.min(1, Math.max(0, level)));
|
|
} catch {}
|
|
}
|
|
|
|
#handlePeerFailure(message: string): void {
|
|
this.#reportFailure(message);
|
|
}
|
|
|
|
#reportFailure(message: string): void {
|
|
if ((this.#state !== "connecting" && this.#state !== "connected") || this.#unexpectedFailureReported) {
|
|
return;
|
|
}
|
|
this.#unexpectedFailureReported = true;
|
|
try {
|
|
this.#options.callbacks.onEvent({ type: "error", message });
|
|
} catch {}
|
|
}
|
|
|
|
/** Serialize one Frameless Bidi control message onto the call's sideband WebSocket. */
|
|
send(message: LiveClientMessage): Promise<void> {
|
|
const operation = this.#sendTail.then(() => {
|
|
if (this.#state !== "connected") throw new Error("Live transport is not connected");
|
|
const sideband = this.#sideband;
|
|
if (!sideband || sideband.readyState !== WebSocket.OPEN) {
|
|
throw new Error("Codex live sideband is not connected");
|
|
}
|
|
sideband.send(JSON.stringify(message));
|
|
});
|
|
this.#sendTail = operation.catch(() => {});
|
|
return operation;
|
|
}
|
|
|
|
/** Queue 16 kHz mono Float32 PCM for native Opus transmission. */
|
|
pushAudio(samples: Float32Array): void {
|
|
if (this.#state !== "connected" || this.#muted || samples.length === 0) return;
|
|
this.#peer?.pushAudio(samples);
|
|
}
|
|
|
|
/** Enable or disable the native audio source and discard partial input when muted. */
|
|
async setMuted(muted: boolean): Promise<void> {
|
|
this.#muted = muted;
|
|
if (this.#state === "connected") this.#peer?.setMuted(muted);
|
|
}
|
|
|
|
/** Stop sideband signaling and the native WebRTC media peer. Safe to call repeatedly. */
|
|
close(): Promise<void> {
|
|
if (this.#closePromise) return this.#closePromise;
|
|
this.#state = "closing";
|
|
const operation = this.#close();
|
|
this.#closePromise = operation;
|
|
return operation;
|
|
}
|
|
|
|
async #close(): Promise<void> {
|
|
this.#options.signal?.removeEventListener("abort", this.#abortListener);
|
|
const sideband = this.#sideband;
|
|
const peer = this.#peer;
|
|
this.#sideband = undefined;
|
|
this.#peer = undefined;
|
|
if (sideband && (sideband.readyState === WebSocket.OPEN || sideband.readyState === WebSocket.CONNECTING)) {
|
|
sideband.close(1000, "done");
|
|
}
|
|
if (peer) {
|
|
try {
|
|
await peer.close();
|
|
} catch {}
|
|
}
|
|
this.#state = "closed";
|
|
}
|
|
}
|