diff --git a/open-sse/executors/chatgpt-web.ts b/open-sse/executors/chatgpt-web.ts index 9a4eb3cf36..d7592e7048 100644 --- a/open-sse/executors/chatgpt-web.ts +++ b/open-sse/executors/chatgpt-web.ts @@ -20,7 +20,6 @@ import { createHash, randomUUID, randomBytes } from "node:crypto"; import { tlsFetchChatGpt, TlsClientUnavailableError, - type TlsFetchOptions, type TlsFetchResult, } from "../services/chatgptTlsClient.ts"; @@ -112,12 +111,12 @@ const TOKEN_TTL_MS = 5 * 60 * 1000; // 5min — accessTokens are short-lived const tokenCache = new Map(); function cookieKey(cookie: string): string { - let hash = 0x811c9dc5; - for (let i = 0; i < cookie.length; i++) { - hash ^= cookie.charCodeAt(i); - hash = (hash * 0x01000193) >>> 0; - } - return hash.toString(16).padStart(8, "0"); + // SHA-256 prefix (64 bits). Used as the Map key for tokenCache and + // warmupCache; the previous 32-bit FNV-1a was small enough that a + // birthday-paradox collision could surface one user's cached accessToken + // to another's request. 64 bits is overkill for the 200-entry cache but + // costs essentially nothing. + return createHash("sha256").update(cookie).digest("hex").slice(0, 16); } function tokenLookup(cookie: string): TokenEntry | null { @@ -153,15 +152,69 @@ interface SessionResponse { user?: { id?: string }; } -function extractRefreshedCookie(setCookieHeader: string | null): string | null { +/** + * Merge any rotated session-token chunks from a Set-Cookie response into the + * original cookie blob, preserving every other cookie the caller pasted + * (cf_clearance, __cf_bm, _cfuvid, _puid, ...). Returns null if no rotation + * occurred or the rotated chunks match what's already there. + * + * Returning only the matched session-token chunks here was a bug: when the + * caller pastes a full DevTools Cookie line (the recommended form), the + * Cloudflare cookies are required for subsequent requests, and dropping + * them re-triggers `cf-mitigated: challenge`. + */ +function mergeRefreshedCookie( + originalCookie: string, + setCookieHeader: string | null +): string | null { if (!setCookieHeader) return null; - // Set-Cookie can rotate either an unchunked token or any of the chunks. - // Capture all relevant chunks and re-emit them as a single Cookie header value. const matches = Array.from( setCookieHeader.matchAll(/(__Secure-next-auth\.session-token(?:\.\d+)?)=([^;,\s]+)/g) ); if (matches.length === 0) return null; - return matches.map((m) => `${m[1]}=${m[2]}`).join("; "); + + const refreshed = new Map(); + for (const m of matches) refreshed.set(m[1], m[2]); + + let blob = originalCookie.trim(); + if (/^cookie\s*:\s*/i.test(blob)) blob = blob.replace(/^cookie\s*:\s*/i, ""); + + // Bare value (no `=`): the original was just the session-token contents. + // Replace with the new chunked form. + if (!/=/.test(blob)) { + return Array.from(refreshed, ([k, v]) => `${k}=${v}`).join("; "); + } + + const pairs = blob.split(/;\s*/).filter(Boolean); + const replaced = new Set(); + const result: string[] = []; + let mutated = false; + for (const pair of pairs) { + const eqIdx = pair.indexOf("="); + if (eqIdx < 0) { + result.push(pair); + continue; + } + const name = pair.slice(0, eqIdx).trim(); + const value = pair.slice(eqIdx + 1); + if (refreshed.has(name)) { + const newValue = refreshed.get(name)!; + if (newValue !== value) mutated = true; + result.push(`${name}=${newValue}`); + replaced.add(name); + } else { + result.push(`${name}=${value}`); + } + } + // Append any rotated chunks that weren't in the original (e.g., the user + // pasted only `.0` but rotation returned `.0` and `.1`). + for (const [name, value] of refreshed) { + if (!replaced.has(name)) { + result.push(`${name}=${value}`); + mutated = true; + } + } + return mutated ? result.join("; ") : null; } /** @@ -212,7 +265,7 @@ async function exchangeSession( throw new Error(`Session exchange failed (HTTP ${response.status})`); } - const refreshed = extractRefreshedCookie(response.headers.get("set-cookie")); + const refreshed = mergeRefreshedCookie(cookie, response.headers.get("set-cookie")); let data: SessionResponse = {}; try { data = JSON.parse(response.text || "{}"); @@ -646,7 +699,6 @@ interface ChatGptMessage { function buildConversationBody( parsed: ParsedMessages, modelSlug: string, - conversationId: string | null, parentMessageId: string ): Record { // Critical: do NOT send prior turns as separate `assistant` and `user` @@ -662,7 +714,7 @@ function buildConversationBody( if (parsed.systemMsg.trim()) { systemParts.push(parsed.systemMsg.trim()); } - if (parsed.history.length > 0 && !conversationId) { + if (parsed.history.length > 0) { const formatted = parsed.history .map((h) => `${h.role === "assistant" ? "Assistant" : "User"}: ${h.content}`) .join("\n\n"); @@ -672,7 +724,7 @@ function buildConversationBody( } const messages: ChatGptMessage[] = []; - if (systemParts.length > 0 && !conversationId) { + if (systemParts.length > 0) { messages.push({ id: randomUUID(), author: { role: "system" }, @@ -690,7 +742,9 @@ function buildConversationBody( action: "next", messages, model: modelSlug, - conversation_id: conversationId, + // Conversation continuity intentionally disabled — Temporary Chat mode + // expires conversation_ids quickly upstream and 404s on reuse. + conversation_id: null, parent_message_id: parentMessageId, timezone_offset_min: -new Date().getTimezoneOffset(), history_and_training_disabled: true, @@ -776,8 +830,8 @@ async function* readChatGptSseEvents( // ─── Content extraction ───────────────────────────────────────────────────── // ChatGPT SSE chunks contain CUMULATIVE content (full text so far in `parts[0]`), -// not deltas. Diff against `seenLen` to emit incremental tokens — same pattern -// perplexity-web.ts uses for markdown blocks (lines 386-397). +// not deltas. Diff against the emitted length to produce incremental tokens — +// same pattern perplexity-web.ts uses for markdown blocks (lines 386-397). interface ContentChunk { delta?: string; @@ -792,14 +846,21 @@ async function* extractContent( eventStream: ReadableStream, signal?: AbortSignal | null ): AsyncGenerator { - let fullAnswer = ""; - let seenLen = 0; + // ChatGPT may echo prior assistant turns at the start of the stream with + // status: "finished_successfully" and full content, before sending the new + // generation. If we emit those bytes downstream, streaming consumers see + // the previous answer prepended to the new one (visible in Open WebUI as + // run-on output across turns). Strategy: only emit deltas after we've seen + // status === "in_progress" for the current message id (i.e., it's being + // generated live in this stream). Echoes always arrive already finished + // and never transition through in_progress, so they get suppressed. An + // end-of-stream fallback handles the rare case where a real turn arrives + // as a single already-finished event (instant/cached responses). let conversationId: string | null = null; - let messageId: string | null = null; - // ChatGPT may echo prior messages in the stream for context before sending - // the new assistant turn. Track which message id we're currently consuming - // so we reset the accumulator when a new turn starts, and so "finished_*" - // on an old echoed message doesn't end the stream prematurely. + let currentId: string | null = null; + let currentParts = ""; + let emittedLen = 0; + let isLive = false; for await (const event of readChatGptSseEvents(eventStream, signal)) { if (event.error) { @@ -815,45 +876,58 @@ async function* extractContent( const m = event.message; if (!m) continue; - - const role = m.author?.role ?? ""; - if (role !== "assistant") continue; + if (m.author?.role !== "assistant") continue; const id = m.id ?? null; - if (id && id !== messageId) { - // A new assistant message has started — reset accumulator. The previous - // message was either an echo of prior context or an aside; we only emit - // content for the latest message. - messageId = id; - fullAnswer = ""; - seenLen = 0; + const status = m.status ?? ""; + + if (id && id !== currentId) { + currentId = id; + currentParts = ""; + emittedLen = 0; + isLive = false; + } + + if (status === "in_progress") { + isLive = true; } const parts = m.content?.parts ?? []; if (parts.length === 0) continue; const cumulative = parts.map((p) => (typeof p === "string" ? p : "")).join(""); + if (cumulative.length > currentParts.length) { + currentParts = cumulative; + } - if (cumulative.length > seenLen) { - const delta = cumulative.slice(seenLen); - fullAnswer = cumulative; - seenLen = cumulative.length; + if (isLive && currentParts.length > emittedLen) { + const delta = currentParts.slice(emittedLen); + emittedLen = currentParts.length; yield { delta, - answer: fullAnswer, + answer: currentParts, conversationId: conversationId ?? undefined, - messageId: messageId ?? undefined, + messageId: currentId ?? undefined, }; } - // Don't break on finished_successfully here — the server may still send - // the new turn after echoing prior messages. Let the stream end naturally - // (when the upstream closes or sends [DONE]). + } + + // End-of-stream fallback: if we never observed status === "in_progress" + // for the current id (single-event reply, cached/instant response), emit + // the accumulated content now so the consumer doesn't get an empty stream. + if (!isLive && currentParts.length > emittedLen) { + yield { + delta: currentParts.slice(emittedLen), + answer: currentParts, + conversationId: conversationId ?? undefined, + messageId: currentId ?? undefined, + }; } yield { delta: "", - answer: fullAnswer, + answer: currentParts, conversationId: conversationId ?? undefined, - messageId: messageId ?? undefined, + messageId: currentId ?? undefined, done: true, }; } @@ -869,8 +943,6 @@ function buildStreamingResponse( model: string, cid: string, created: number, - history: Array<{ role: string; content: string }>, - currentMsg: string, signal?: AbortSignal | null ): ReadableStream { const encoder = new TextEncoder(); @@ -893,14 +965,7 @@ function buildStreamingResponse( ) ); - let fullAnswer = ""; - let respConversationId: string | null = null; - let respMessageId: string | null = null; - for await (const chunk of extractContent(eventStream, signal)) { - if (chunk.conversationId) respConversationId = chunk.conversationId; - if (chunk.messageId) respMessageId = chunk.messageId; - if (chunk.error) { controller.enqueue( encoder.encode( @@ -925,7 +990,6 @@ function buildStreamingResponse( } if (chunk.done) { - fullAnswer = chunk.answer || fullAnswer; break; } @@ -953,7 +1017,6 @@ function buildStreamingResponse( ); } } - if (chunk.answer) fullAnswer = chunk.answer; } controller.enqueue( @@ -969,12 +1032,6 @@ function buildStreamingResponse( ) ); controller.enqueue(encoder.encode("data: [DONE]\n\n")); - // Conversation continuity intentionally not persisted (Temporary Chat - // mode — see comment near the top of the file). - void respConversationId; - void respMessageId; - void history; - void currentMsg; } catch (err) { controller.enqueue( encoder.encode( @@ -1010,17 +1067,12 @@ async function buildNonStreamingResponse( model: string, cid: string, created: number, - history: Array<{ role: string; content: string }>, currentMsg: string, signal?: AbortSignal | null ): Promise { let fullAnswer = ""; - let respConversationId: string | null = null; - let respMessageId: string | null = null; for await (const chunk of extractContent(eventStream, signal)) { - if (chunk.conversationId) respConversationId = chunk.conversationId; - if (chunk.messageId) respMessageId = chunk.messageId; if (chunk.error) { return new Response( JSON.stringify({ @@ -1036,11 +1088,6 @@ async function buildNonStreamingResponse( if (chunk.answer) fullAnswer = chunk.answer; } - // Conversation continuity intentionally not persisted (Temporary Chat mode). - void respConversationId; - void respMessageId; - void history; - fullAnswer = cleanChatGptText(fullAnswer); const promptTokens = Math.ceil(currentMsg.length / 4); const completionTokens = Math.ceil(fullAnswer.length / 4); @@ -1284,11 +1331,10 @@ export class ChatGptWebExecutor extends BaseExecutor { // chatgpt.com expires those conversation_ids quickly — re-using them // returns 404. Each request starts a fresh conversation; clients (Open // WebUI, OpenAI-API-style) send the full history each turn anyway. - const conversationId: string | null = null; const parentMessageId = randomUUID(); const modelSlug = MODEL_MAP[model] ?? model; - const cgptBody = buildConversationBody(parsed, modelSlug, conversationId, parentMessageId); + const cgptBody = buildConversationBody(parsed, modelSlug, parentMessageId); const headers: Record = { ...browserHeaders(), @@ -1305,10 +1351,7 @@ export class ChatGptWebExecutor extends BaseExecutor { if (proofToken) headers["openai-sentinel-proof-token"] = proofToken; if (turnstileToken) headers["openai-sentinel-turnstile-token"] = turnstileToken; - log?.info?.( - "CGPT-WEB", - `Conversation request → ${modelSlug} (continue=${!!conversationId}, pow=${!!proofToken})` - ); + log?.info?.("CGPT-WEB", `Conversation request → ${modelSlug} (pow=${!!proofToken})`); let response: TlsFetchResult; try { @@ -1388,15 +1431,7 @@ export class ChatGptWebExecutor extends BaseExecutor { let finalResponse: Response; if (stream) { - const sseStream = buildStreamingResponse( - bodyStream, - model, - cid, - created, - parsed.history, - parsed.currentMsg, - signal - ); + const sseStream = buildStreamingResponse(bodyStream, model, cid, created, signal); finalResponse = new Response(sseStream, { status: 200, headers: { @@ -1411,7 +1446,6 @@ export class ChatGptWebExecutor extends BaseExecutor { model, cid, created, - parsed.history, parsed.currentMsg, signal ); diff --git a/open-sse/services/chatgptTlsClient.ts b/open-sse/services/chatgptTlsClient.ts index c42a0dfbf8..9d66d44323 100644 --- a/open-sse/services/chatgptTlsClient.ts +++ b/open-sse/services/chatgptTlsClient.ts @@ -62,14 +62,11 @@ async function getClient(): Promise<{ }; await client.start(); - console.log("[CGPT-TLS] Native runtime ready (Firefox 148 fingerprint)."); installExitHook(); return client; } catch (err) { clientPromise = null; const msg = err instanceof Error ? err.message : String(err); - - console.log(`[CGPT-TLS] FAILED to start: ${msg}`); throw new TlsClientUnavailableError( `TLS impersonation client failed to start: ${msg}. ` + `Verify tls-client-node is installed and its native binary downloaded.` @@ -234,11 +231,10 @@ async function tlsFetchStreaming( // Wait briefly for the file to appear so we can detect early errors. const ready = await waitForFile(path, 5_000); if (!ready) { - // File never appeared — request must have errored out before any body - // bytes. Wait for it to settle and surface as a non-streaming response. const r = await requestPromise.catch( (e) => ({ status: 502, headers: {}, body: String(e) }) as TlsResponseLike ); + await cleanupTempPath(path); return { status: r.status, headers: toHeaders(r.headers), @@ -247,16 +243,18 @@ async function tlsFetchStreaming( }; } - // Peek the first bytes to distinguish a JSON error envelope from an SSE - // body. Errors typically come back as `{"detail":"..."}`; SSE bodies start - // with `data:` or empty lines. If it looks like an error, wait for the - // full body and return non-streaming so the executor can read response.text. + // Peek the first bytes to decide whether this looks like SSE. Anything + // that doesn't positively look like SSE (JSON `{...}`, HTML `<...>`, plain + // text rate-limit messages, Cloudflare challenge pages, etc.) gets surfaced + // as a non-streaming response so the executor sees the real upstream status + // and body — otherwise non-2xx error pages get silently treated as 200 OK + // and the SSE parser produces an empty completion. const peek = await readFirstBytes(path, 256); - const trimmedPeek = peek.replace(/^[\s\r\n]+/, ""); - if (trimmedPeek.startsWith("{")) { + if (!looksLikeSse(peek)) { const r = await requestPromise.catch( (e) => ({ status: 502, headers: {}, body: String(e) }) as TlsResponseLike ); + await cleanupTempPath(path); return { status: r.status, headers: toHeaders(r.headers), @@ -265,9 +263,9 @@ async function tlsFetchStreaming( }; } - // Tail the file as a real-time stream. We assume HTTP 200 here — if the - // upstream errored, we'd have caught it via the JSON-peek above. The - // request promise is still tracked so cleanup can run after it settles. + // Looks like SSE — start tailing. SSE bodies in practice are always 2xx; + // tls-client-node doesn't expose response status separately from full-body + // completion, so we report 200 and let the SSE parser consume the stream. const stream = tailFile(path, eofSymbol, requestPromise, signal); const headers = new Headers({ "Content-Type": "text/event-stream", @@ -276,6 +274,26 @@ async function tlsFetchStreaming( return { status: 200, headers, text: null, body: stream }; } +/** + * Returns true if the peeked response body looks like an SSE stream — i.e., + * begins (after any leading whitespace) with one of the SSE field markers + * (`data:`, `event:`, `id:`, `retry:`) or a comment line (`:`). + * + * Exported for tests. + */ +export function looksLikeSse(text: string): boolean { + const trimmed = text.replace(/^[\s\r\n]+/, ""); + if (!trimmed) return false; + if (trimmed.startsWith(":")) return true; + return /^(data|event|id|retry):/i.test(trimmed); +} + +async function cleanupTempPath(path: string): Promise { + await unlink(path).catch(() => {}); + const dir = path.substring(0, path.lastIndexOf("/")); + await rmdir(dir).catch(() => {}); +} + async function readFirstBytes(path: string, n: number): Promise { const fd = await open(path, "r"); try { @@ -313,11 +331,21 @@ function tailFile( let offset = 0; let finished = false; let aborted = false; + let upstreamError: Error | null = null; - // Mark when the request completes so we know to drain the rest. - done.finally(() => { - finished = true; - }); + // Track request settlement, capturing both fulfillment and rejection. + // Without the rejection branch, a mid-stream tls-client-node error + // becomes an unhandledRejection — the stream cleans up silently and + // the consumer sees what looks like a successful truncated response. + done.then( + () => { + finished = true; + }, + (err) => { + upstreamError = err instanceof Error ? err : new Error(String(err)); + finished = true; + } + ); // If the caller aborts, stop tailing immediately. const onAbort = () => { @@ -328,6 +356,7 @@ function tailFile( else signal.addEventListener("abort", onAbort, { once: true }); } + let errored = false; try { while (!aborted) { const { bytesRead } = await fd.read(buf, 0, buf.length, offset); @@ -342,7 +371,13 @@ function tailFile( } controller.enqueue(new Uint8Array(chunk)); } else if (finished) { - // No more data and request completed — drain done. + // No more data and request completed. If the request rejected, + // surface the error so the consumer doesn't think the stream + // ended cleanly. + if (upstreamError) { + controller.error(upstreamError); + errored = true; + } break; } else { await sleep(25); @@ -350,13 +385,14 @@ function tailFile( } } catch (err) { controller.error(err); + errored = true; } finally { if (signal) signal.removeEventListener("abort", onAbort); await fd.close().catch(() => {}); await unlink(path).catch(() => {}); const dir = path.substring(0, path.lastIndexOf("/")); await rmdir(dir).catch(() => {}); - controller.close(); + if (!errored) controller.close(); } }, }); diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index 11382c42f8..0baa14b2c6 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -148,6 +148,10 @@ export async function executeChatWithBreaker({ expiresIn: newCreds.expiresIn, expiresAt: newCreds.expiresAt, providerSpecificData: newCreds.providerSpecificData, + // Cookie/session providers (chatgpt-web) rotate the stored + // apiKey blob mid-request — forward it so the DB credential + // doesn't go stale after Set-Cookie rotation. + apiKey: newCreds.apiKey, testStatus: "active", }); }, diff --git a/src/sse/services/tokenRefresh.ts b/src/sse/services/tokenRefresh.ts index adee5a3e84..6df10e2f72 100755 --- a/src/sse/services/tokenRefresh.ts +++ b/src/sse/services/tokenRefresh.ts @@ -106,6 +106,15 @@ export async function updateProviderCredentials(connectionId: string, newCredent if (newCredentials.providerSpecificData) { updates.providerSpecificData = newCredentials.providerSpecificData; } + // Cookie/session providers (chatgpt-web, ...) refresh by rotating the + // stored apiKey blob — propagate that here too so DB credentials don't + // go stale after Set-Cookie rotation. + if (newCredentials.apiKey) { + updates.apiKey = newCredentials.apiKey; + } + if (newCredentials.testStatus) { + updates.testStatus = newCredentials.testStatus; + } const result = await updateProviderConnection(connectionId, updates); log.info("TOKEN_REFRESH", "Credentials updated in localDb", { diff --git a/tests/unit/chatgpt-web.test.ts b/tests/unit/chatgpt-web.test.ts index 1f1ec730b3..4a16192145 100644 --- a/tests/unit/chatgpt-web.test.ts +++ b/tests/unit/chatgpt-web.test.ts @@ -4,7 +4,7 @@ import assert from "node:assert/strict"; const { ChatGptWebExecutor, __resetChatGptWebCachesForTesting } = await import("../../open-sse/executors/chatgpt-web.ts"); const { getExecutor, hasSpecializedExecutor } = await import("../../open-sse/executors/index.ts"); -const { __setTlsFetchOverrideForTesting } = +const { __setTlsFetchOverrideForTesting, looksLikeSse, TlsClientUnavailableError } = await import("../../open-sse/services/chatgptTlsClient.ts"); // ─── Helpers ──────────────────────────────────────────────────────────────── @@ -892,3 +892,347 @@ test("Provider registry: chatgpt-web is registered with gpt-5.3-instant model", assert.equal(entry.authHeader, "cookie"); assert.ok(entry.models.find((m) => m.id === "gpt-5.3-instant")); }); + +// ─── Cookie rotation preserves Cloudflare cookies ─────────────────────────── + +test("Cookie rotation: full DevTools blob keeps cf_clearance/__cf_bm/_cfuvid", async () => { + // When the user pastes the recommended full DevTools Cookie line and + // NextAuth rotates the session-token chunks, only those chunks should + // change — the Cloudflare cookies must be preserved or every subsequent + // request gets cf-mitigated: challenge. + reset(); + const m = installMockFetch({ + session: { + status: 200, + body: { + accessToken: "jwt-abc", + expires: new Date(Date.now() + 3600_000).toISOString(), + user: { id: "user-1" }, + }, + setCookie: + "__Secure-next-auth.session-token.0=NEW0; Path=/; HttpOnly, " + + "__Secure-next-auth.session-token.1=NEW1; Path=/; HttpOnly", + }, + }); + try { + let refreshed = null; + const executor = new ChatGptWebExecutor(); + await executor.execute({ + model: "gpt-5.3-instant", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: false, + credentials: { + apiKey: + "__Secure-next-auth.session-token.0=OLD0; " + + "__Secure-next-auth.session-token.1=OLD1; " + + "cf_clearance=CFCLEAR; __cf_bm=CFBM; _cfuvid=CFUV; _puid=PUID", + }, + signal: AbortSignal.timeout(10_000), + log: null, + onCredentialsRefreshed: (creds) => { + refreshed = creds; + }, + }); + + assert.ok(refreshed, "callback should fire on rotation"); + assert.match(refreshed.apiKey, /session-token\.0=NEW0/, "session-token.0 rotated"); + assert.match(refreshed.apiKey, /session-token\.1=NEW1/, "session-token.1 rotated"); + assert.match(refreshed.apiKey, /cf_clearance=CFCLEAR/, "cf_clearance preserved"); + assert.match(refreshed.apiKey, /__cf_bm=CFBM/, "__cf_bm preserved"); + assert.match(refreshed.apiKey, /_cfuvid=CFUV/, "_cfuvid preserved"); + assert.match(refreshed.apiKey, /_puid=PUID/, "_puid preserved"); + // Old session-token values must NOT survive in the merged blob. + assert.doesNotMatch(refreshed.apiKey, /OLD0/); + assert.doesNotMatch(refreshed.apiKey, /OLD1/); + } finally { + m.restore(); + } +}); + +test("Cookie rotation: returns null when Set-Cookie has no session-token", async () => { + // When NextAuth doesn't rotate (Set-Cookie sets only unrelated cookies, or + // returns the same session-token value), the callback shouldn't fire. + reset(); + const m = installMockFetch({ + session: { + status: 200, + body: { + accessToken: "jwt-abc", + expires: new Date(Date.now() + 3600_000).toISOString(), + user: { id: "user-1" }, + }, + setCookie: "some-other-cookie=value; Path=/", + }, + }); + try { + let refreshed = null; + const executor = new ChatGptWebExecutor(); + await executor.execute({ + model: "gpt-5.3-instant", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: false, + credentials: { apiKey: "cookie-v1" }, + signal: AbortSignal.timeout(10_000), + log: null, + onCredentialsRefreshed: (creds) => { + refreshed = creds; + }, + }); + assert.equal(refreshed, null, "no rotation should not fire callback"); + } finally { + m.restore(); + } +}); + +// ─── Echo suppression in extractContent ───────────────────────────────────── + +test("Stream parser: echoed prior assistant turn is suppressed (streaming)", async () => { + // chatgpt.com sometimes echoes prior assistant turns at the start of the + // stream with status: finished_successfully BEFORE the new generation + // starts. The parser must not emit echoed bytes — otherwise the SSE + // consumer sees old content prepended to the new answer. + reset(); + const m = installMockFetch({ + conv: { + status: 200, + events: [ + // Echo of a prior assistant turn — full content, finished, never + // transitions through in_progress. + { + conversation_id: "c1", + message: { + id: "echo-1", + author: { role: "assistant" }, + content: { content_type: "text", parts: ["OLD ECHO ANSWER"] }, + status: "finished_successfully", + }, + }, + // The real new turn — streams as in_progress, then finishes. + { + conversation_id: "c1", + message: { + id: "new-1", + author: { role: "assistant" }, + content: { content_type: "text", parts: ["Hello"] }, + status: "in_progress", + }, + }, + { + conversation_id: "c1", + message: { + id: "new-1", + author: { role: "assistant" }, + content: { content_type: "text", parts: ["Hello world"] }, + status: "finished_successfully", + }, + }, + ], + }, + }); + try { + const executor = new ChatGptWebExecutor(); + const result = await executor.execute({ + model: "gpt-5.3-instant", + body: { messages: [{ role: "user", content: "hi" }], stream: true }, + stream: true, + credentials: { apiKey: "test" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + + const text = await result.response.text(); + const contentDeltas = text + .split("\n") + .filter((l) => l.startsWith("data: ") && l !== "data: [DONE]") + .map((l) => { + try { + return JSON.parse(l.slice(6)); + } catch { + return null; + } + }) + .filter((j) => j?.choices?.[0]?.delta?.content) + .map((j) => j.choices[0].delta.content); + + const joined = contentDeltas.join(""); + assert.equal(joined, "Hello world", "only the new turn is emitted"); + assert.doesNotMatch(joined, /OLD ECHO/, "echoed content must not appear in stream"); + } finally { + m.restore(); + } +}); + +test("Stream parser: echoed prior assistant turn is suppressed (non-streaming)", async () => { + reset(); + const m = installMockFetch({ + conv: { + status: 200, + events: [ + { + conversation_id: "c1", + message: { + id: "echo-1", + author: { role: "assistant" }, + content: { content_type: "text", parts: ["OLD ECHO ANSWER"] }, + status: "finished_successfully", + }, + }, + { + conversation_id: "c1", + message: { + id: "new-1", + author: { role: "assistant" }, + content: { content_type: "text", parts: ["Hello world"] }, + status: "in_progress", + }, + }, + ], + }, + }); + try { + const executor = new ChatGptWebExecutor(); + const result = await executor.execute({ + model: "gpt-5.3-instant", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: false, + credentials: { apiKey: "test" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + const json = await result.response.json(); + assert.equal(json.choices[0].message.content, "Hello world"); + } finally { + m.restore(); + } +}); + +test("Stream parser: instant single-event reply still surfaces via fallback", async () => { + // Edge case: a real reply that arrives in a single event with status + // already finished_successfully (cached/instant). End-of-stream fallback + // should emit it; otherwise streaming consumers would get nothing. + reset(); + const m = installMockFetch({ + conv: { + status: 200, + events: [ + { + conversation_id: "c1", + message: { + id: "instant-1", + author: { role: "assistant" }, + content: { content_type: "text", parts: ["instant reply"] }, + status: "finished_successfully", + }, + }, + ], + }, + }); + try { + const executor = new ChatGptWebExecutor(); + const result = await executor.execute({ + model: "gpt-5.3-instant", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: false, + credentials: { apiKey: "test" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + const json = await result.response.json(); + assert.equal(json.choices[0].message.content, "instant reply"); + } finally { + m.restore(); + } +}); + +// ─── TLS client unavailable ───────────────────────────────────────────────── + +test("Error: TlsClientUnavailableError returns 502 with TLS_UNAVAILABLE code", async () => { + reset(); + // Make the override throw TlsClientUnavailableError on the conversation + // call (after a successful session/sentinel/dpl pass). The executor catches + // the error and surfaces TLS_UNAVAILABLE so operators can identify missing + // native binary issues quickly. + let convAttempted = false; + __setTlsFetchOverrideForTesting(async (url) => { + if (url === "https://chatgpt.com/" || url === "https://chatgpt.com") { + return { + status: 200, + headers: makeHeaders({ "Content-Type": "text/html" }), + text: '', + body: null, + }; + } + if (url.includes("/api/auth/session")) { + return { + status: 200, + headers: makeHeaders({ "Content-Type": "application/json" }), + text: JSON.stringify({ + accessToken: "jwt", + expires: new Date(Date.now() + 3600_000).toISOString(), + user: { id: "u" }, + }), + body: null, + }; + } + if (url.includes("/sentinel/chat-requirements")) { + return { + status: 200, + headers: makeHeaders({ "Content-Type": "application/json" }), + text: JSON.stringify({ token: "t", proofofwork: { required: false } }), + body: null, + }; + } + if (url.endsWith("/backend-api/f/conversation")) { + convAttempted = true; + throw new TlsClientUnavailableError("native binary not loaded"); + } + return { + status: 200, + headers: makeHeaders(), + text: "", + body: null, + }; + }); + try { + const executor = new ChatGptWebExecutor(); + const result = await executor.execute({ + model: "gpt-5.3-instant", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: false, + credentials: { apiKey: "test" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + assert.ok(convAttempted); + assert.equal(result.response.status, 502); + const json = await result.response.json(); + assert.equal(json.error.code, "TLS_UNAVAILABLE"); + } finally { + __setTlsFetchOverrideForTesting(null); + } +}); + +// ─── looksLikeSse heuristic ───────────────────────────────────────────────── + +test("looksLikeSse: detects SSE bodies", () => { + assert.equal(looksLikeSse('data: {"v":"hi"}\n\n'), true); + assert.equal(looksLikeSse("\r\n\r\ndata: foo"), true, "leading blank lines OK"); + assert.equal(looksLikeSse("event: end\ndata: []"), true); + assert.equal(looksLikeSse("id: 42\ndata: x"), true); + assert.equal(looksLikeSse(": comment\ndata: x"), true, "SSE comment lines start with :"); + assert.equal(looksLikeSse("retry: 3000\n"), true); +}); + +test("looksLikeSse: rejects non-SSE bodies that previously passed as 200", () => { + // The original peek heuristic only looked for `{` to detect JSON errors, + // letting Cloudflare HTML challenge pages and plain-text 4xx bodies + // masquerade as 200 SSE responses. looksLikeSse must reject these. + assert.equal(looksLikeSse('{"detail":"rate limited"}'), false, "JSON error"); + assert.equal(looksLikeSse("\n"), false, "HTML doctype"); + assert.equal(looksLikeSse(""), false, "HTML page"); + assert.equal(looksLikeSse("Just a moment..."), false, "Cloudflare plain-text challenge"); + assert.equal(looksLikeSse("Attention Required! | Cloudflare"), false); + assert.equal(looksLikeSse(""), false, "empty body"); + assert.equal(looksLikeSse(" \n\n"), false, "whitespace only"); + assert.equal(looksLikeSse("error: rate limit"), false, "non-SSE field name"); +});