diff --git a/README.md b/README.md index 040fee89..4a6c54b3 100644 --- a/README.md +++ b/README.md @@ -1299,7 +1299,7 @@ Thanks to all contributors who helped make 9Router better! Built on the shoulders of giants: -- **CLIProxyAPI(https://github.com/router-for-me/CLIProxyAPI)** — original Go implementation that inspired this JavaScript port. +- **[CLIProxyAPI](https://github.com/router-for-me/CLIProxyAPI)** — original Go implementation that inspired this JavaScript port. - **[RTK](https://github.com/rtk-ai/rtk)** ![Stars](https://img.shields.io/github/stars/rtk-ai/rtk?style=flat&color=yellow) — Rust token-saver. 9Router ports its compression pipeline to JS → **−20-40% input tokens** on every request. - **[Caveman](https://github.com/JuliusBrussee/caveman)** ![Stars](https://img.shields.io/github/stars/JuliusBrussee/caveman?style=flat&color=yellow) by **[@JuliusBrussee](https://github.com/JuliusBrussee)** — viral *"why use many token when few token do trick"*. 9Router adapts its prompt → **−65% output tokens**. diff --git a/open-sse/config/runtimeConfig.js b/open-sse/config/runtimeConfig.js index a82cff3a..c2a4888e 100644 --- a/open-sse/config/runtimeConfig.js +++ b/open-sse/config/runtimeConfig.js @@ -32,7 +32,10 @@ export const MEMORY_CONFIG = { }; // Stream stall timeout: abort if no chunk received within this duration -export const STREAM_STALL_TIMEOUT_MS = 3 * 60 * 1000; +export const STREAM_STALL_TIMEOUT_MS = 60 * 1000; + +// Fetch connect timeout: abort if upstream doesn't return response headers within this duration +export const FETCH_CONNECT_TIMEOUT_MS = 30 * 1000; // Default token limits export const DEFAULT_MAX_TOKENS = 64000; diff --git a/open-sse/executors/base.js b/open-sse/executors/base.js index 8cb46721..adb942b1 100644 --- a/open-sse/executors/base.js +++ b/open-sse/executors/base.js @@ -1,5 +1,6 @@ -import { HTTP_STATUS, RETRY_CONFIG, DEFAULT_RETRY_CONFIG, resolveRetryEntry } from "../config/runtimeConfig.js"; +import { HTTP_STATUS, RETRY_CONFIG, DEFAULT_RETRY_CONFIG, resolveRetryEntry, FETCH_CONNECT_TIMEOUT_MS } from "../config/runtimeConfig.js"; import { proxyAwareFetch } from "../utils/proxyFetch.js"; +import { dbg } from "../utils/debugLog.js"; /** * BaseExecutor - Base class for provider executors @@ -121,13 +122,25 @@ export class BaseExecutor { if (!retryAttemptsByUrl[urlIndex]) retryAttemptsByUrl[urlIndex] = 0; + // Abort if upstream doesn't return response headers within FETCH_CONNECT_TIMEOUT_MS + const connectCtrl = new AbortController(); + const connectTimer = setTimeout(() => connectCtrl.abort(new Error("fetch connect timeout")), FETCH_CONNECT_TIMEOUT_MS); + const mergedSignal = signal ? AbortSignal.any([signal, connectCtrl.signal]) : connectCtrl.signal; + try { + const bodyStr = JSON.stringify(transformedBody); + const fetchT0 = Date.now(); + dbg("FETCH", `${this.provider.toUpperCase()} → ${url} | body=${bodyStr.length}B | connectTimeout=${FETCH_CONNECT_TIMEOUT_MS}ms`); const response = await proxyAwareFetch(url, { method: "POST", headers, - body: JSON.stringify(transformedBody), - signal + body: bodyStr, + signal: mergedSignal }, proxyOptions); + clearTimeout(connectTimer); + const ct = response.headers?.get?.("content-type") || ""; + const cl = response.headers?.get?.("content-length") || "?"; + dbg("FETCH", `${this.provider.toUpperCase()} ← ${response.status} | ttft=${Date.now() - fetchT0}ms | ct=${ct} | cl=${cl}`); if (await tryRetry(urlIndex, response.status, `status ${response.status}`)) { urlIndex--; continue; } @@ -139,8 +152,12 @@ export class BaseExecutor { return { response, url, headers, transformedBody }; } catch (error) { + clearTimeout(connectTimer); lastError = error; - if (error.name === "AbortError") throw error; + const isConnectTimeout = connectCtrl.signal.aborted && error.name === "AbortError"; + dbg("FETCH", `${this.provider.toUpperCase()} ✖ ${error.name}: ${error.message}${isConnectTimeout ? " (connect timeout)" : ""}`); + // Connect timeout is internal — convert to retryable network error, don't propagate AbortError + if (error.name === "AbortError" && !isConnectTimeout) throw error; // Map network/fetch exceptions to 502 retry config if (await tryRetry(urlIndex, HTTP_STATUS.BAD_GATEWAY, `network "${error.message}"`)) { urlIndex--; continue; } diff --git a/open-sse/executors/codex.js b/open-sse/executors/codex.js index 80cbd6d3..eaac4bf6 100644 --- a/open-sse/executors/codex.js +++ b/open-sse/executors/codex.js @@ -6,11 +6,100 @@ import { normalizeResponsesInput } from "../translator/helpers/responsesApiHelpe import { fetchImageAsBase64 } from "../translator/helpers/imageHelper.js"; import { getModelUpstreamId } from "../config/providerModels.js"; import { getConsistentMachineId } from "../../src/shared/utils/machineId.js"; +import { DEFAULT_RETRY_CONFIG, resolveRetryEntry } from "../config/runtimeConfig.js"; +import { dbg } from "../utils/debugLog.js"; + +// SSE error patterns inside 200-OK body that should trigger retry as if 503 +const CODEX_SSE_OVERLOADED_PATTERNS = ["server_is_overloaded", "service_unavailable_error"]; +const CODEX_SSE_PEEK_BYTES = 4096; // In-memory map: hash(machineId + first assistant content) → { sessionId, lastUsed } const SESSION_TTL_MS = 60 * 60 * 1000; // 1 hour const assistantSessionMap = new Map(); +// Server-generated item id prefixes that Codex /responses cannot resolve when store=false +const SERVER_ID_PATTERN = /^(rs|fc|resp|msg)_/; + +// Hosted tool types that Codex/OpenAI Responses executes server-side +const CODEX_HOSTED_TOOL_TYPES = new Set([ + "image_generation", "web_search", "web_search_preview", "file_search", + "computer", "computer_use_preview", "code_interpreter", "mcp", "local_shell" +]); + +// Allowlist of fields accepted by Codex Responses API — anything else is stripped +const RESPONSES_API_ALLOWLIST = new Set([ + "model", "input", "instructions", "tools", "tool_choice", "stream", "store", + "reasoning", "service_tier", "include", "prompt_cache_key", "client_metadata" +]); + +// Convert role=system → role=developer in body.input (keeps content in cacheable prefix) +function convertSystemToDeveloperRole(body) { + if (!Array.isArray(body.input)) return; + for (const item of body.input) { + if (!item || typeof item !== "object" || Array.isArray(item)) continue; + const isSystemMsg = item.role === "system" && (!item.type || item.type === "message"); + if (isSystemMsg) item.role = "developer"; + } +} + +// Strip server-generated item IDs (rs_/fc_/resp_/msg_) from input — avoids 404 with store=false +function stripStoredItemReferences(body) { + if (!Array.isArray(body.input)) return; + body.input = body.input.filter((item) => { + if (typeof item === "string" && SERVER_ID_PATTERN.test(item)) return false; + if (item && typeof item === "object" && !Array.isArray(item)) { + if (item.type === "item_reference") return false; + if (typeof item.id === "string" && SERVER_ID_PATTERN.test(item.id)) delete item.id; + } + return true; + }); +} + +// Flatten Chat-Completions tool shape into Responses flat format + filter unsupported tools +function normalizeCodexTools(body) { + if (!Array.isArray(body.tools)) return; + const validNames = new Set(); + body.tools = body.tools.filter((tool) => { + if (!tool || typeof tool !== "object" || Array.isArray(tool)) return false; + const type = typeof tool.type === "string" ? tool.type : ""; + if (type === "namespace") { + if (Array.isArray(tool.tools)) { + for (const st of tool.tools) { + const n = typeof st?.name === "string" ? st.name.trim().slice(0, 128) : ""; + if (n) validNames.add(n); + } + } + return true; + } + if (type !== "function") { + if (!type || tool.function || typeof tool.name === "string") return false; + return CODEX_HOSTED_TOOL_TYPES.has(type); + } + const fn = tool.function && typeof tool.function === "object" && !Array.isArray(tool.function) ? tool.function : null; + const rawName = typeof tool.name === "string" ? tool.name : (typeof fn?.name === "string" ? fn.name : ""); + const name = rawName.trim(); + if (!name) return false; + const description = typeof tool.description === "string" ? tool.description : (typeof fn?.description === "string" ? fn.description : ""); + const parameters = (tool.parameters && typeof tool.parameters === "object" && !Array.isArray(tool.parameters)) + ? tool.parameters + : (fn?.parameters && typeof fn.parameters === "object" && !Array.isArray(fn.parameters) ? fn.parameters : { type: "object", properties: {} }); + for (const k of Object.keys(tool)) delete tool[k]; + tool.type = "function"; + tool.name = name.slice(0, 128); + if (description) tool.description = description; + tool.parameters = parameters; + validNames.add(name); + return true; + }); + // Drop tool_choice if it references an unknown function name + if (body.tool_choice && typeof body.tool_choice === "object" && !Array.isArray(body.tool_choice)) { + if (body.tool_choice.type === "function") { + const n = typeof body.tool_choice.name === "string" ? body.tool_choice.name.trim() : ""; + if (!n || !validNames.has(n)) delete body.tool_choice; + } + } +} + // Cache machine ID at module level (resolved once) let cachedMachineId = null; getConsistentMachineId().then(id => { cachedMachineId = id; }); @@ -33,32 +122,54 @@ function extractItemText(item) { return ""; } -// Resolve session_id from first assistant message + machineId to avoid cross-user collision -function resolveConversationSessionId(input, machineId) { - const machineSessionId = machineId ? `sess_${hashContent(machineId)}` : generateSessionId(); - if (!Array.isArray(input) || input.length === 0) return machineSessionId; +// Normalize a session id candidate (trim, length cap) +function normalizeSessionId(value) { + if (typeof value !== "string") return null; + const v = value.trim(); + if (!v || v.length > 256) return null; + return v; +} - // Find first assistant message that has actual text content - let text = ""; - for (const item of input) { - if (item.role === "assistant") { - text = extractItemText(item); - if (text) break; +// Resolve prompt-cache session id with priority: body → assistant-text-hash → workspaceId → machineId +function resolveCacheSessionId(body, credentials, machineId) { + // 1. Client-provided session/conversation id (highest priority — stable per conversation) + const fromBody = + normalizeSessionId(body?.prompt_cache_key) || + normalizeSessionId(body?.session_id) || + normalizeSessionId(body?.conversation_id); + if (fromBody) return fromBody; + + // 2. Hash accumulated assistant text (≥50 chars) — sticky session across turns + if (Array.isArray(body?.input) && body.input.length > 0) { + let text = ""; + const MIN_LEN = 50; + const CAP_LEN = 200; + for (const item of body.input) { + if (item?.role !== "assistant") continue; + const t = extractItemText(item); + if (!t) continue; + text += t; + if (text.length >= CAP_LEN) break; + } + if (text.length >= MIN_LEN) { + const hash = hashContent((machineId || "") + text.slice(0, CAP_LEN)); + const entry = assistantSessionMap.get(hash); + if (entry) { + entry.lastUsed = Date.now(); + return entry.sessionId; + } + const sessionId = generateSessionId(); + assistantSessionMap.set(hash, { sessionId, lastUsed: Date.now() }); + return sessionId; } } - if (!text) return machineSessionId; - const hash = hashContent((machineId || "") + text); - const entry = assistantSessionMap.get(hash); - if (entry) { - entry.lastUsed = Date.now(); - return entry.sessionId; - } + // 3. Account-wide fallback (workspaceId from connection) + const workspaceId = normalizeSessionId(credentials?.providerSpecificData?.workspaceId); + if (workspaceId) return workspaceId; - - const sessionId = generateSessionId(); - assistantSessionMap.set(hash, { sessionId, lastUsed: Date.now() }); - return sessionId; + // 4. Last resort — stable per-machine id + return machineId ? `sess_${hashContent(machineId)}` : generateSessionId(); } // Cleanup expired entries periodically @@ -80,12 +191,19 @@ export class CodexExecutor extends BaseExecutor { } /** - * Override headers to add session_id per conversation - * transformRequest runs BEFORE buildHeaders, sets this._currentSessionId + * Override headers to add codex-specific identity headers. + * transformRequest runs BEFORE buildHeaders, sets this._currentSessionId. */ buildHeaders(credentials, stream = true) { const headers = super.buildHeaders(credentials, stream); headers["session_id"] = this._currentSessionId || credentials?.connectionId || "default"; + // Identify client type to Codex backend (matches official codex CLI) + if (!headers["originator"]) headers["originator"] = "codex_cli_rs"; + // Workspace binding header — improves account scope + cache affinity + const workspaceId = credentials?.providerSpecificData?.workspaceId; + if (typeof workspaceId === "string" && workspaceId && !headers["chatgpt-account-id"]) { + headers["chatgpt-account-id"] = workspaceId; + } return headers; } @@ -117,9 +235,100 @@ export class CodexExecutor extends BaseExecutor { } async execute(args) { - // Fetch remote images before the synchronous transform/execute pipeline - await this.prefetchImages(args.body); - return super.execute(args); + const imgCount = Array.isArray(args.body?.input) ? args.body.input.reduce((n, it) => n + (Array.isArray(it.content) ? it.content.filter(c => c.type === "image_url").length : 0), 0) : 0; + const inputLen = Array.isArray(args.body?.input) ? args.body.input.length : 0; + dbg("CODEX", `execute start | inputItems=${inputLen} | images=${imgCount} | sessionId=${this._currentSessionId || "pending"}`); + if (imgCount > 0) { + const t0 = Date.now(); + await this.prefetchImages(args.body); + dbg("CODEX", `prefetchImages done | ${Date.now() - t0}ms`); + } else { + await this.prefetchImages(args.body); + } + + // Retry loop for SSE-level overloaded errors (200 OK body contains event: error) + // Reuses 503 retry config — same semantic: upstream temporarily unavailable + const retryConfig = { ...DEFAULT_RETRY_CONFIG, ...this.config.retry }; + const { attempts, delayMs } = resolveRetryEntry(retryConfig[503]); + let attempt = 0; + while (true) { + const result = await super.execute(args); + const peek = await this._peekSseOverloaded(result.response); + if (!peek.matched) { + // Replace body with re-assembled stream (prefix bytes already read + rest) + if (peek.replacementBody) { + result.response = new Response(peek.replacementBody, { + status: result.response.status, + statusText: result.response.statusText, + headers: result.response.headers, + }); + } + return result; + } + if (attempt >= attempts) { + args.log?.warn?.("RETRY", `CODEX | SSE overloaded "${peek.matched}" — retries exhausted (${attempt}/${attempts})`); + // Out of retries → return with replacement body so client gets the error + if (peek.replacementBody) { + result.response = new Response(peek.replacementBody, { + status: result.response.status, + statusText: result.response.statusText, + headers: result.response.headers, + }); + } + return result; + } + attempt++; + args.log?.debug?.("RETRY", `CODEX | SSE "${peek.matched}" retry ${attempt}/${attempts} after ${delayMs / 1000}s`); + dbg("CODEX", `SSE overloaded "${peek.matched}" → retry ${attempt}/${attempts} in ${delayMs}ms`); + try { await result.response.body?.cancel?.(); } catch { /* noop */ } + await new Promise(r => setTimeout(r, delayMs)); + } + } + + // Peek first N bytes of SSE body to detect upstream "overloaded" errors. + // Returns { matched: string|null, replacementBody: ReadableStream|null }. + // Caller MUST use replacementBody (original body has been read). + async _peekSseOverloaded(response) { + if (!response || !response.ok || !response.body) return { matched: null, replacementBody: null }; + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const chunks = []; + let text = ""; + let matched = null; + try { + while (text.length < CODEX_SSE_PEEK_BYTES) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + text += decoder.decode(value, { stream: true }); + const hit = CODEX_SSE_OVERLOADED_PATTERNS.find(p => text.includes(p)); + if (hit) { matched = hit; break; } + } + } catch (e) { + dbg("CODEX", `peek read error: ${e.message}`); + } + reader.releaseLock(); + + // Re-assemble stream: prefix chunks + remaining upstream body + const upstream = response.body; + let upstreamReader = null; + const replacementBody = new ReadableStream({ + start(controller) { + for (const c of chunks) controller.enqueue(c); + upstreamReader = upstream.getReader(); + }, + async pull(controller) { + try { + const { done, value } = await upstreamReader.read(); + if (done) { controller.close(); return; } + controller.enqueue(value); + } catch (e) { controller.error(e); } + }, + cancel(reason) { + try { upstreamReader?.cancel(reason); } catch { /* noop */ } + }, + }); + return { matched, replacementBody }; } // Parse Codex usage_limit_reached to extract precise resetsAtMs; fallback to default otherwise @@ -154,8 +363,8 @@ export class CodexExecutor extends BaseExecutor { transformRequest(model, body, stream, credentials) { this._isCompact = !!body._compact; delete body._compact; - // Resolve conversation-stable session_id from input history + machineId - this._currentSessionId = resolveConversationSessionId(body.input, cachedMachineId); + // Resolve conversation-stable session_id (priority: body → assistant-text → workspace → machine) + this._currentSessionId = resolveCacheSessionId(body, credentials, cachedMachineId); // Convert string input to array format (Codex API requires input as array) const normalized = normalizeResponsesInput(body.input); if (normalized) body.input = normalized; @@ -165,6 +374,13 @@ export class CodexExecutor extends BaseExecutor { body.input = [{ type: "message", role: "user", content: [{ type: "input_text", text: "..." }] }]; } + // Keep system prompts in body.input as role=developer so they stay in the cacheable prefix + convertSystemToDeveloperRole(body); + // Strip server-generated item IDs (rs_/fc_/resp_/msg_) — Codex /responses can't resolve when store=false + stripStoredItemReferences(body); + // Flatten function tools + drop unsupported types + normalizeCodexTools(body); + // Ensure streaming is enabled (Codex API requires it) body.stream = true; @@ -176,6 +392,11 @@ export class CodexExecutor extends BaseExecutor { // Ensure store is false (Codex requirement) body.store = false; + // Inject prompt_cache_key for stable Codex prompt caching + if (!body.prompt_cache_key && this._currentSessionId) { + body.prompt_cache_key = this._currentSessionId; + } + // Map virtual Codex review models to the upstream Codex model before suffix parsing. body.model = getModelUpstreamId("cx", body.model || model); @@ -223,6 +444,12 @@ export class CodexExecutor extends BaseExecutor { delete body.metadata; // Cursor sends this but Codex doesn't support it delete body.stream_options; // Cursor sends this but Codex doesn't support it delete body.safety_identifier; // Droid CLI sends this but Codex doesn't support it + delete body.previous_response_id; // store=false → backend can't resolve previous resp; avoid 404 + + // Final allowlist filter — strip any unknown field that could trigger upstream "routing_unsupported" + for (const k of Object.keys(body)) { + if (!RESPONSES_API_ALLOWLIST.has(k)) delete body[k]; + } return body; } diff --git a/open-sse/translator/response/openai-responses.js b/open-sse/translator/response/openai-responses.js index 184b5396..0cf79cbd 100644 --- a/open-sse/translator/response/openai-responses.js +++ b/open-sse/translator/response/openai-responses.js @@ -568,10 +568,17 @@ export function openaiResponsesToOpenAIResponse(chunk, state) { return null; } - // Reasoning events (convert to content or skip) + // Reasoning summary delta → emit as reasoning_content for client thinking display if (eventType === "response.reasoning_summary_text.delta") { - // Optionally include reasoning as content, or skip - return null; + const delta = data.delta || ""; + if (!delta) return null; + return { + id: state.chatId, + object: "chat.completion.chunk", + created: state.created, + model: state.model || "unknown", + choices: [{ index: 0, delta: { reasoning_content: delta }, finish_reason: null }] + }; } // Ignore other events diff --git a/open-sse/utils/debugLog.js b/open-sse/utils/debugLog.js new file mode 100644 index 00000000..67cdc31f --- /dev/null +++ b/open-sse/utils/debugLog.js @@ -0,0 +1,14 @@ +// Debug logging utility — only active in dev mode (NODE_ENV !== "production") +// Outputs are tagged with [DBG:tag] for easy grep/filter +const isDev = process.env.NODE_ENV !== "production"; + +function ts() { + return new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit", second: "2-digit" }); +} + +export function dbg(tag, msg) { + if (!isDev) return; + console.log(`[${ts()}] 🐛 [DBG:${tag}] ${msg}`); +} + +export const isDebugEnabled = isDev; diff --git a/open-sse/utils/proxyFetch.js b/open-sse/utils/proxyFetch.js index 43b71642..16cac731 100644 --- a/open-sse/utils/proxyFetch.js +++ b/open-sse/utils/proxyFetch.js @@ -1,9 +1,101 @@ import { Readable } from "stream"; import { MEMORY_CONFIG } from "../config/runtimeConfig.js"; +import { dbg } from "./debugLog.js"; const originalFetch = globalThis.fetch; const proxyDispatchers = new Map(); +// ─── TLS fingerprinting via got-scraping (browser-like JA3) ─────────────── +// Lazy-loaded once; if import fails (missing optional native deps in some +// envs) we silently fall back to native fetch — no behavioral change. +let _gotScraping = null; +let _gotScrapingChecked = false; +const _gotScrapingLoggedHosts = new Set(); + +async function getGotScraping() { + if (_gotScrapingChecked) return _gotScraping; + _gotScrapingChecked = true; + try { + const mod = await import("got-scraping"); + _gotScraping = typeof mod.gotScraping === "function" ? mod.gotScraping : null; + if (_gotScraping) dbg("TLS", "got-scraping loaded (browser-like JA3 enabled)"); + } catch (e) { + console.warn(`[ProxyFetch] got-scraping unavailable, falling back to native fetch: ${e.message}`); + _gotScraping = null; + } + return _gotScraping; +} + +// Run a request through got-scraping streaming, return a fetch-compatible Response +async function gotScrapingFetch(url, options) { + const gs = await getGotScraping(); + if (!gs) return null; + + const method = (options.method || "GET").toUpperCase(); + const headersInit = options.headers || {}; + const headers = headersInit instanceof Headers + ? Object.fromEntries(headersInit.entries()) + : { ...headersInit }; + + return new Promise((resolve, reject) => { + let settled = false; + const stream = gs.stream({ + url, + method, + headers, + body: method === "GET" || method === "HEAD" ? undefined : options.body, + throwHttpErrors: false, + retry: { limit: 0 }, + timeout: { request: undefined }, // streaming → no overall timeout + followRedirect: false, + decompress: true, + }); + + if (options.signal) { + const onAbort = () => { try { stream.destroy(new Error("aborted")); } catch { /* noop */ } }; + if (options.signal.aborted) onAbort(); + else options.signal.addEventListener("abort", onAbort, { once: true }); + } + + stream.once("response", (res) => { + if (settled) return; + settled = true; + const resHeaders = new Headers(); + for (const [k, v] of Object.entries(res.headers || {})) { + if (Array.isArray(v)) v.forEach((x) => resHeaders.append(k, String(x))); + else if (v != null) resHeaders.set(k, String(v)); + } + const body = Readable.toWeb(stream); + resolve(new Response(body, { status: res.statusCode, statusText: res.statusMessage || "", headers: resHeaders })); + }); + + stream.once("error", (err) => { + if (settled) return; + settled = true; + reject(err); + }); + }); +} + +async function tryGotScrapingFetch(url, options) { + try { + const res = await gotScrapingFetch(url, options); + if (res) { + try { + const host = new URL(typeof url === "string" ? url : url.toString()).hostname; + if (!_gotScrapingLoggedHosts.has(host)) { + _gotScrapingLoggedHosts.add(host); + dbg("TLS", `using got-scraping for ${host}`); + } + } catch { /* noop */ } + } + return res; + } catch (e) { + console.warn(`[ProxyFetch] got-scraping request failed, fallback to native fetch: ${e.message}`); + return null; + } +} + // DNS cache — use Map to avoid prototype pollution via malformed hostnames const DNS_CACHE = new Map(); const MITM_BYPASS_HOSTS = [ @@ -255,6 +347,8 @@ export async function proxyAwareFetch(url, options = {}, proxyOptions = null) { } } + // got-scraping disabled — use native fetch directly + // (Re-enable per-host by wrapping with tryGotScrapingFetch when needed) return originalFetch(url, options); } diff --git a/open-sse/utils/stream.js b/open-sse/utils/stream.js index 138f2ee3..42829686 100644 --- a/open-sse/utils/stream.js +++ b/open-sse/utils/stream.js @@ -3,6 +3,7 @@ import { FORMATS } from "../translator/formats.js"; import { trackPendingRequest, appendRequestLog } from "@/lib/usageDb.js"; import { extractUsage, hasValidUsage, estimateUsage, logUsage, addBufferToUsage, filterUsageForFormat, COLORS } from "./usageTracking.js"; import { parseSSELine, hasValuableContent, fixInvalidId, formatSSE } from "./streamHelpers.js"; +import { dbg, isDebugEnabled } from "./debugLog.js"; export { COLORS, formatSSE }; @@ -58,11 +59,15 @@ export function createSSEStream(options = {}) { let accumulatedContent = ""; let accumulatedThinking = ""; let ttftAt = null; + let sseLineCount = 0; + let sseEmittedCount = 0; + const eventTypeCounts = {}; return new TransformStream({ transform(chunk, controller) { if (!ttftAt) { ttftAt = Date.now(); + dbg("SSE", `${provider}/${model} | first chunk received | size=${chunk?.byteLength || 0}B`); } const text = decoder.decode(chunk, { stream: true }); buffer += text; @@ -73,6 +78,14 @@ export function createSSEStream(options = {}) { for (const line of lines) { const trimmed = line.trim(); + if (isDebugEnabled && trimmed) { + sseLineCount++; + if (trimmed.startsWith("event:")) { + const evt = trimmed.slice(6).trim(); + eventTypeCounts[evt] = (eventTypeCounts[evt] || 0) + 1; + if (eventTypeCounts[evt] <= 2) dbg("SSE", `recv event: ${evt} (#${eventTypeCounts[evt]})`); + } + } // Passthrough mode: normalize and forward if (mode === STREAM_MODE.PASSTHROUGH) { @@ -248,12 +261,15 @@ export function createSSEStream(options = {}) { const output = formatSSE(item, sourceFormat); reqLogger?.appendConvertedChunk?.(output); controller.enqueue(sharedEncoder.encode(output)); + sseEmittedCount++; } } } }, flush(controller) { + const evtSummary = Object.entries(eventTypeCounts).map(([k, v]) => `${k}=${v}`).join(",") || "none"; + dbg("SSE", `flush | provider=${provider} | model=${model} | recvLines=${sseLineCount} | emitted=${sseEmittedCount} | events=[${evtSummary}]`); trackPendingRequest(model, provider, connectionId, false); try { const remaining = decoder.decode(); diff --git a/open-sse/utils/streamHandler.js b/open-sse/utils/streamHandler.js index f3fc50c8..567b59ac 100644 --- a/open-sse/utils/streamHandler.js +++ b/open-sse/utils/streamHandler.js @@ -1,5 +1,6 @@ // Stream handler with disconnect detection - shared for all providers import { STREAM_STALL_TIMEOUT_MS } from "../config/runtimeConfig.js"; +import { dbg, isDebugEnabled } from "./debugLog.js"; // Get HH:MM:SS timestamp function getTimeString() { @@ -38,6 +39,7 @@ export function createStreamController({ onDisconnect, onError, log, provider, m disconnected = true; logStream(`disconnect: ${reason}`); + dbg("CTRL", `${provider}/${model} | disconnect=${reason} | dur=${Date.now() - startTime}ms`); // Delay abort to allow cleanup abortTimeout = setTimeout(() => { @@ -117,8 +119,23 @@ export function createDisconnectAwareStream(transformStream, streamController) { streamController.handleError(error); reader.cancel().catch(() => {}); writer.abort().catch(() => {}); - - if (!wasConnected || error.name === "AbortError" || error.message?.includes("aborted")) { + + // Treat network resets / socket hang up / abort as graceful close + const msg = error?.message || ""; + const code = error?.code || error?.cause?.code || ""; + const isNetworkClose = + error.name === "AbortError" || + msg.includes("aborted") || + msg.includes("socket hang up") || + msg.includes("ECONNRESET") || + msg.includes("ETIMEDOUT") || + msg.includes("EPIPE") || + code === "ECONNRESET" || + code === "ETIMEDOUT" || + code === "EPIPE" || + code === "UND_ERR_SOCKET"; + + if (!wasConnected || isNetworkClose) { try { controller.close(); } catch (e) { @@ -158,6 +175,11 @@ export function createDisconnectAwareStream(transformStream, streamController) { */ export function pipeWithDisconnect(providerResponse, transformStream, streamController) { let stallTimer = null; + let chunkCount = 0; + let totalBytes = 0; + let lastChunkAt = Date.now(); + const t0 = Date.now(); + const tag = "STREAM"; const clearStall = () => { if (stallTimer) { clearTimeout(stallTimer); stallTimer = null; } }; @@ -165,6 +187,7 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont clearStall(); stallTimer = setTimeout(() => { stallTimer = null; + dbg(tag, `STALL TIMEOUT ${STREAM_STALL_TIMEOUT_MS}ms | chunks=${chunkCount} | bytes=${totalBytes} | sinceLast=${Date.now() - lastChunkAt}ms`); streamController.handleError?.(new Error("stream stall timeout")); streamController.abort?.(); }, STREAM_STALL_TIMEOUT_MS); @@ -177,20 +200,30 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont signal: streamController.signal, startTime: streamController.startTime, isConnected: () => streamController.isConnected(), - handleComplete: () => { clearStall(); streamController.handleComplete(); }, - handleError: (e) => { clearStall(); streamController.handleError(e); }, - handleDisconnect: (r) => { clearStall(); streamController.handleDisconnect(r); }, + handleComplete: () => { dbg(tag, `complete | chunks=${chunkCount} | bytes=${totalBytes} | dur=${Date.now() - t0}ms`); clearStall(); streamController.handleComplete(); }, + handleError: (e) => { dbg(tag, `error: ${e?.message} | chunks=${chunkCount} | bytes=${totalBytes} | dur=${Date.now() - t0}ms`); clearStall(); streamController.handleError(e); }, + handleDisconnect: (r) => { dbg(tag, `disconnect: ${r} | chunks=${chunkCount} | bytes=${totalBytes} | dur=${Date.now() - t0}ms`); clearStall(); streamController.handleDisconnect(r); }, abort: () => { clearStall(); streamController.abort(); } }; armStall(); + dbg(tag, `pipe start | stallTimeout=${STREAM_STALL_TIMEOUT_MS}ms`); const upstreamTap = new TransformStream({ transform(chunk, controller) { + chunkCount++; + const sz = chunk?.byteLength || chunk?.length || 0; + totalBytes += sz; + const now = Date.now(); + const gap = now - lastChunkAt; + lastChunkAt = now; + if (isDebugEnabled && (chunkCount <= 5 || chunkCount % 20 === 0 || gap > 5000)) { + dbg(tag, `chunk #${chunkCount} | size=${sz}B | gap=${gap}ms | total=${totalBytes}B`); + } armStall(); controller.enqueue(chunk); }, - flush() { clearStall(); } + flush() { dbg(tag, `upstream EOF | chunks=${chunkCount} | bytes=${totalBytes} | dur=${Date.now() - t0}ms`); clearStall(); } }); const transformedBody = providerResponse.body diff --git a/package.json b/package.json index 340622f1..c15d0f89 100644 --- a/package.json +++ b/package.json @@ -22,6 +22,7 @@ "confbox": "^0.2.4", "express": "^5.2.1", "fs": "^0.0.1-security", + "got-scraping": "^4.2.1", "http-proxy-middleware": "^3.0.5", "jose": "^6.1.3", "marked": "^18.0.1", diff --git a/src/mitm/antigravityIdeVersion.js b/src/mitm/antigravityIdeVersion.js new file mode 100644 index 00000000..cac9bc47 --- /dev/null +++ b/src/mitm/antigravityIdeVersion.js @@ -0,0 +1,58 @@ +"use strict"; + +// Rewrite Antigravity IDE markers so upstream AG 2.x backend accepts the request. +// User-Agent header (antigravity/) and body.metadata.ideVersion are forced +// to a known-good IDE version. Hardcoded MVP — toggle/version configurable later. + +const ANTIGRAVITY_IDE_VERSION = "1.23.2"; +const ANTIGRAVITY_IDE_VERSION_OVERRIDE_ENABLED = true; + +function shouldRewriteMetadata(metadata) { + if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) return false; + if (String(metadata.ideName || "").toLowerCase() === "antigravity") return true; + if (String(metadata.ideType || "").toUpperCase() === "ANTIGRAVITY") return true; + return Object.prototype.hasOwnProperty.call(metadata, "ideVersion"); +} + +function rewriteAntigravityUserAgent(userAgent, version) { + if (typeof userAgent !== "string" || !userAgent.includes("antigravity/")) return userAgent; + return userAgent.replace(/antigravity\/[^\s]+/, `antigravity/${version}`); +} + +function applyAntigravityIdeVersionOverride(bodyBuffer, headers, log = () => {}) { + if (!ANTIGRAVITY_IDE_VERSION_OVERRIDE_ENABLED) { + return { bodyBuffer, headers, applied: false, version: ANTIGRAVITY_IDE_VERSION }; + } + + const nextHeaders = { ...headers }; + const nextUserAgent = rewriteAntigravityUserAgent(nextHeaders["user-agent"], ANTIGRAVITY_IDE_VERSION); + const userAgentChanged = nextUserAgent !== nextHeaders["user-agent"]; + if (userAgentChanged) nextHeaders["user-agent"] = nextUserAgent; + + try { + const parsed = JSON.parse(bodyBuffer.toString()); + if (!shouldRewriteMetadata(parsed?.metadata)) { + if (userAgentChanged) log(`🛰️ [antigravity] user-agent version override → ${ANTIGRAVITY_IDE_VERSION}`); + return { bodyBuffer, headers: nextHeaders, applied: userAgentChanged, version: ANTIGRAVITY_IDE_VERSION }; + } + + const previousVersion = parsed.metadata.ideVersion; + parsed.metadata.ideVersion = ANTIGRAVITY_IDE_VERSION; + const nextBodyBuffer = Buffer.from(JSON.stringify(parsed)); + log(`🛰️ [antigravity] IDE version override: ${previousVersion || "unknown"} → ${ANTIGRAVITY_IDE_VERSION}`); + return { bodyBuffer: nextBodyBuffer, headers: nextHeaders, applied: true, version: ANTIGRAVITY_IDE_VERSION }; + } catch (e) { + if (userAgentChanged) { + log(`🛰️ [antigravity] user-agent version override → ${ANTIGRAVITY_IDE_VERSION}`); + return { bodyBuffer, headers: nextHeaders, applied: true, version: ANTIGRAVITY_IDE_VERSION }; + } + log(`🛰️ [antigravity] IDE version override skipped: ${e.message}`); + return { bodyBuffer, headers: nextHeaders, applied: false, version: ANTIGRAVITY_IDE_VERSION }; + } +} + +module.exports = { + ANTIGRAVITY_IDE_VERSION, + applyAntigravityIdeVersionOverride, + rewriteAntigravityUserAgent, +}; diff --git a/src/mitm/config.js b/src/mitm/config.js index 3c14adba..eb8aefc9 100644 --- a/src/mitm/config.js +++ b/src/mitm/config.js @@ -31,7 +31,7 @@ const URL_PATTERNS = { // Synonym map: rawModel from request → canonical alias key in mitmAlias DB const MODEL_SYNONYMS = { antigravity: { - "gemini-default": "gemini-3-flash-agent", + "gemini-default": "gemini-3.5-flash-low", "gemini-3.1-pro-high": "gemini-pro-agent", "gemini-3-pro-high": "gemini-pro-agent", "gemini-3-pro-low": "gemini-3.1-pro-low", diff --git a/src/mitm/server.js b/src/mitm/server.js index f334db60..09469ad0 100644 --- a/src/mitm/server.js +++ b/src/mitm/server.js @@ -1,4 +1,6 @@ const https = require("https"); +const http2 = require("http2"); +const tls = require("tls"); const fs = require("fs"); const path = require("path"); const dns = require("dns"); @@ -9,6 +11,7 @@ const { IS_DEV, LSOF_BIN, TARGET_HOSTS, URL_PATTERNS, MODEL_SYNONYMS, MODEL_PATT const { DATA_DIR, MITM_DIR } = require("./paths"); const { getCertForDomain } = require("./cert/generate"); const { getMitmAlias } = require("./dbReader"); +const { applyAntigravityIdeVersionOverride } = require("./antigravityIdeVersion"); const LOCAL_PORT = 443; const IS_WIN = process.platform === "win32"; const ENABLE_FILE_LOG = IS_DEV; @@ -129,16 +132,136 @@ function getMappedModel(tool, model) { */ async function passthrough(req, res, bodyBuffer, onResponse) { const originalHost = (req.headers.host || TARGET_HOSTS[0]).split(":")[0]; - const targetHost = HOST_REWRITE[originalHost] || originalHost; - const targetIP = await resolveTargetIP(targetHost); + // Only rewrite host for chat endpoints — daily-cloudcode-pa rejects auth/login requests + const isChatEndpoint = req.url.includes(":generateContent") || req.url.includes(":streamGenerateContent"); + const targetHost = isChatEndpoint ? (HOST_REWRITE[originalHost] || originalHost) : originalHost; const dumper = ENABLE_FILE_LOG ? createResponseDumper(req, "passthrough") : null; + const tool = getToolForHost(req.headers.host); + const versionOverride = tool === "antigravity" + ? applyAntigravityIdeVersionOverride(bodyBuffer, req.headers, log) + : { bodyBuffer, headers: req.headers }; + const bodyForForwarding = versionOverride.bodyBuffer; + const headersForForwarding = { ...versionOverride.headers, host: targetHost }; + if (bodyForForwarding !== bodyBuffer) { + headersForForwarding["content-length"] = String(bodyForForwarding.length); + } + + // ALPN negotiate: try HTTP/2 first (like browsers/mitmweb), fallback HTTP/1.1 + try { + const proto = await negotiateAlpn(targetHost); + if (proto === "h2") { + return await passthroughHttp2(req, res, bodyForForwarding, headersForForwarding, targetHost, onResponse, dumper); + } + } catch (e) { + err(`[mitm] ALPN negotiate failed: ${e.message}, fallback to HTTP/1.1`); + } + + return passthroughHttps(req, res, bodyForForwarding, headersForForwarding, targetHost, onResponse, dumper); +} + +// ── ALPN negotiation cache ──────────────────────────────────── +const alpnCache = new Map(); // host → "h2" | "http/1.1" +async function negotiateAlpn(host) { + if (alpnCache.has(host)) return alpnCache.get(host); + const ip = await resolveTargetIP(host); + return new Promise((resolve, reject) => { + const socket = tls.connect({ + host: ip, port: 443, servername: host, + ALPNProtocols: ["h2", "http/1.1"], rejectUnauthorized: false, + }, () => { + const proto = socket.alpnProtocol || "http/1.1"; + alpnCache.set(host, proto); + log(`🔗 [mitm] ALPN ${host} → ${proto}`); + socket.end(); + resolve(proto); + }); + socket.once("error", reject); + socket.setTimeout(5000, () => { socket.destroy(new Error("ALPN timeout")); }); + }); +} + +// HTTP/2 passthrough using node:http2 native +async function passthroughHttp2(req, res, bodyBuffer, headers, targetHost, onResponse, dumper) { + const targetIP = await resolveTargetIP(targetHost); + // HTTP/2 pseudo-headers required; strip HTTP/1.1-only headers + const h2Headers = {}; + for (const [k, v] of Object.entries(headers)) { + const lk = k.toLowerCase(); + if (lk === "host" || lk === "connection" || lk === "keep-alive" || + lk === "transfer-encoding" || lk === "upgrade" || lk === "proxy-connection") continue; + h2Headers[lk] = v; + } + h2Headers[":method"] = req.method; + h2Headers[":path"] = req.url; + h2Headers[":scheme"] = "https"; + h2Headers[":authority"] = targetHost; + + return new Promise((resolve) => { + const client = http2.connect(`https://${targetHost}`, { + createConnection: () => tls.connect({ + host: targetIP, port: 443, servername: targetHost, + ALPNProtocols: ["h2"], rejectUnauthorized: false, + }), + }); + client.once("error", (e) => { + err(`[mitm] http2 client error: ${e.message}`); + if (dumper) { dumper.writeChunk(`\n[ERROR h2] ${e.message}\n`); dumper.end(); } + if (!res.headersSent) res.writeHead(502); + if (!res.writableEnded) res.end("Bad Gateway"); + try { client.close(); } catch {} + resolve(); + }); + + const stream = client.request(h2Headers, { endStream: bodyBuffer.length === 0 }); + if (bodyBuffer.length > 0) stream.end(bodyBuffer); + + stream.once("response", (responseHeaders) => { + const status = responseHeaders[":status"]; + // Filter pseudo-headers + connection-specific + const outHeaders = {}; + for (const [k, v] of Object.entries(responseHeaders)) { + if (k.startsWith(":")) continue; + if (k === "connection" || k === "keep-alive" || k === "transfer-encoding") continue; + outHeaders[k] = v; + } + res.writeHead(status, outHeaders); + if (dumper) dumper.writeHeader(status, outHeaders); + + const chunks = []; + stream.on("data", chunk => { + if (dumper) dumper.writeChunk(chunk); + if (onResponse) chunks.push(chunk); + res.write(chunk); + }); + stream.on("end", () => { + if (dumper) dumper.end(); + if (!res.writableEnded) res.end(); + if (onResponse) try { onResponse(Buffer.concat(chunks), outHeaders); } catch {} + try { client.close(); } catch {} + resolve(); + }); + }); + stream.once("error", (e) => { + err(`[mitm] http2 stream error: ${e.message}`); + if (dumper) { dumper.writeChunk(`\n[ERROR h2-stream] ${e.message}\n`); dumper.end(); } + if (!res.headersSent) res.writeHead(502); + if (!res.writableEnded) res.end(); + try { client.close(); } catch {} + resolve(); + }); + }); +} + +// Fallback: raw https.request HTTP/1.1 with custom DNS (bypasses /etc/hosts MITM loop) +async function passthroughHttps(req, res, bodyBuffer, headers, targetHost, onResponse, dumper) { + const targetIP = await resolveTargetIP(targetHost); const forwardReq = https.request({ hostname: targetIP, port: 443, path: req.url, method: req.method, - headers: { ...req.headers, host: targetHost }, + headers, servername: targetHost, rejectUnauthorized: false }, (forwardRes) => { @@ -150,7 +273,6 @@ async function passthrough(req, res, bodyBuffer, onResponse) { return; } - // Tee: forward to client AND optionally buffer + dump const chunks = []; forwardRes.on("data", chunk => { if (dumper) dumper.writeChunk(chunk); diff --git a/src/shared/constants/cliTools.js b/src/shared/constants/cliTools.js index 8f97b8bb..c3aa010c 100644 --- a/src/shared/constants/cliTools.js +++ b/src/shared/constants/cliTools.js @@ -8,12 +8,13 @@ export const MITM_TOOLS = { description: "Google Antigravity IDE with MITM", configType: "mitm", mitmDomain: "daily-cloudcode-pa.googleapis.com", - modelAliases: ["gemini-3-flash-agent", "gemini-3.5-flash-low", "gemini-pro-agent", "gemini-3.1-pro-low", "claude-sonnet-4-6", "claude-opus-4-6-thinking", "gpt-oss-120b-medium", "gemini-3-flash"], + modelAliases: ["gemini-3.5-flash-low", "gemini-3-flash-agent", "gemini-3.5-flash-extra-low", "gemini-3.1-pro-low", "gemini-pro-agent", "claude-sonnet-4-6", "claude-opus-4-6-thinking", "gpt-oss-120b-medium", "gemini-3-flash"], defaultModels: [ - { id: "gemini-3-flash-agent", name: "Gemini 3.5 Flash (High) / Default", alias: "gemini-3-flash-agent" }, - { id: "gemini-3.5-flash-low", name: "Gemini 3.5 Flash (Medium)", alias: "gemini-3.5-flash-low" }, - { id: "gemini-pro-agent", name: "Gemini 3.1 Pro (High)", alias: "gemini-pro-agent" }, + { id: "gemini-3.5-flash-low", name: "Gemini 3.5 Flash (Medium) / Default", alias: "gemini-3.5-flash-low" }, + { id: "gemini-3-flash-agent", name: "Gemini 3.5 Flash (High)", alias: "gemini-3-flash-agent" }, + { id: "gemini-3.5-flash-extra-low", name: "Gemini 3.5 Flash (Low)", alias: "gemini-3.5-flash-extra-low" }, { id: "gemini-3.1-pro-low", name: "Gemini 3.1 Pro (Low)", alias: "gemini-3.1-pro-low" }, + { id: "gemini-pro-agent", name: "Gemini 3.1 Pro (High)", alias: "gemini-pro-agent" }, { id: "claude-sonnet-4-6", name: "Claude Sonnet 4.6 (Thinking)", alias: "claude-sonnet-4-6" }, { id: "claude-opus-4-6-thinking", name: "Claude Opus 4.6 (Thinking)", alias: "claude-opus-4-6-thinking" }, { id: "gpt-oss-120b-medium", name: "GPT-OSS 120B (Medium)", alias: "gpt-oss-120b-medium" },