wehub-resource-sync
1717 lines
67 KiB
TypeScript
1717 lines
67 KiB
TypeScript
import crypto, { randomUUID } from "crypto";
|
||
import {
|
||
BaseExecutor,
|
||
mergeAbortSignals,
|
||
mergeUpstreamExtraHeaders,
|
||
type ExecuteInput,
|
||
type ExecutorLog,
|
||
type ProviderCredentials,
|
||
} from "./base.ts";
|
||
import { applyFingerprint, isCliCompatEnabled } from "../config/cliFingerprints.ts";
|
||
import { buildAntigravityUpstreamError } from "./antigravityUpstreamError.ts";
|
||
import {
|
||
PROVIDERS,
|
||
OAUTH_ENDPOINTS,
|
||
HTTP_STATUS,
|
||
STREAM_READINESS_TIMEOUT_MS,
|
||
ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE,
|
||
} from "../config/constants.ts";
|
||
import { scrubProxyAndFingerprintHeaders } from "../services/antigravityHeaderScrub.ts";
|
||
import {
|
||
antigravityNativeOAuthUserAgent,
|
||
antigravityUserAgent,
|
||
} from "../services/antigravityHeaders.ts";
|
||
import { classify429, decide429, type Decision } from "../services/antigravity429Engine.ts";
|
||
import {
|
||
injectCreditsField,
|
||
shouldRetryWithCredits,
|
||
shouldUseCreditsFirst,
|
||
getCreditsMode,
|
||
handleCreditsFailure,
|
||
} from "../services/antigravityCredits.ts";
|
||
import { persistCreditBalance, getAllPersistedCreditBalances } from "@/lib/db/creditBalance";
|
||
import { setConnectionRateLimitUntil } from "@/lib/db/providers";
|
||
import { getMitmAlias } from "@/lib/db/models";
|
||
import { obfuscateSensitiveWords } from "../services/antigravityObfuscation.ts";
|
||
import { resolveAntigravityVersion } from "../services/antigravityVersion.ts";
|
||
import { ensureAntigravityProjectAssigned } from "../services/antigravityProjectBootstrap.ts";
|
||
import {
|
||
resolveAntigravityModelId,
|
||
getAntigravityModelFallbacks,
|
||
} from "../config/antigravityModelAliases.ts";
|
||
import { cloakAntigravityToolPayload } from "../config/toolCloaking.ts";
|
||
import {
|
||
shouldStripCloudCodeThinking,
|
||
stripCloudCodeThinkingConfig,
|
||
} from "../services/cloudCodeThinking.ts";
|
||
import { buildGeminiTools } from "../translator/helpers/geminiToolsSanitizer.ts";
|
||
import { DEFAULT_SAFETY_SETTINGS } from "../translator/helpers/geminiHelper.ts";
|
||
import {
|
||
type AntigravityCollectedStream,
|
||
processAntigravitySSEText,
|
||
flushAntigravitySSEText,
|
||
} from "./antigravity/sseCollect.ts";
|
||
// processAntigravitySSEPayload re-exported for external importers (tests).
|
||
export { processAntigravitySSEPayload } from "./antigravity/sseCollect.ts";
|
||
import {
|
||
applyAntigravityClientProfileHeaders,
|
||
removeHeaderCaseInsensitive,
|
||
} from "../services/antigravityClientProfile.ts";
|
||
import {
|
||
generateAntigravityRequestId,
|
||
getAntigravityEnvelopeUserAgent,
|
||
getAntigravitySessionId,
|
||
} from "../services/antigravityIdentity.ts";
|
||
import * as prl from "../utils/providerRequestLogging.ts";
|
||
|
||
const MAX_RETRY_AFTER_MS = 60_000;
|
||
const LONG_RETRY_THRESHOLD_MS = 60_000;
|
||
const CREDITS_EXHAUSTED_TTL_MS = 5 * 60 * 60 * 1000; // 5 hours
|
||
// Cap for transient 5xx backoff — shorter than the 429 cap to avoid long stalls on
|
||
// infra hiccups ("Agent execution terminated", "high traffic", capacity errors).
|
||
const ANTIGRAVITY_TRANSIENT_RETRY_MAX_MS = 15_000;
|
||
|
||
const ANTIGRAVITY_TRANSIENT_ERROR_PATTERNS: RegExp[] = [
|
||
/high\s+traffic/i,
|
||
/agent\s+(execution\s+)?terminated\s+due\s+to\s+error/i,
|
||
/capacity/i,
|
||
/temporarily\s+unavailable/i,
|
||
/timeout/i,
|
||
/stream\s+(ended|closed|terminated|interrupted)/i,
|
||
/empty\s+response/i,
|
||
];
|
||
|
||
const ANTIGRAVITY_TRANSIENT_STATUSES = new Set([
|
||
HTTP_STATUS.SERVER_ERROR,
|
||
HTTP_STATUS.BAD_GATEWAY,
|
||
HTTP_STATUS.SERVICE_UNAVAILABLE,
|
||
HTTP_STATUS.GATEWAY_TIMEOUT,
|
||
]);
|
||
// The upstream API uses plain model IDs (no -high/-low suffix).
|
||
// Tier suffixes were speculative and caused 404 for gemini-3.x models — the
|
||
// bare-Pro→Low normalization was retired (the set stayed empty, making the guard
|
||
// dead code). Only keep models that are live-proven via streamGenerateContent.
|
||
|
||
interface AntigravityContent {
|
||
role: string;
|
||
parts: unknown[];
|
||
[key: string]: unknown;
|
||
}
|
||
|
||
type AntigravityCredentials = ProviderCredentials & {
|
||
projectId?: string | null;
|
||
expiresIn?: number;
|
||
};
|
||
|
||
type AntigravityChunkContent = Record<string, unknown> & {
|
||
role?: string;
|
||
parts?: Array<
|
||
Record<string, unknown> & {
|
||
text?: unknown;
|
||
functionCall?: Record<string, unknown>;
|
||
functionResponse?: unknown;
|
||
thought?: unknown;
|
||
thoughtSignature?: unknown;
|
||
}
|
||
>;
|
||
};
|
||
|
||
type AntigravityCreditEntry = {
|
||
creditType?: string;
|
||
creditAmount?: string;
|
||
};
|
||
|
||
function getChunkedOrFixedBody(bodyStr: string, stream: boolean): BodyInit {
|
||
if (stream) {
|
||
return new ReadableStream(
|
||
{
|
||
async start(controller) {
|
||
controller.enqueue(new TextEncoder().encode(bodyStr));
|
||
controller.close();
|
||
},
|
||
},
|
||
{ highWaterMark: 16384 }
|
||
);
|
||
}
|
||
return bodyStr;
|
||
}
|
||
|
||
function cloneAntigravityRequestBody(body: unknown): unknown {
|
||
if (!body || typeof body !== "object") {
|
||
return body;
|
||
}
|
||
|
||
try {
|
||
return structuredClone(body);
|
||
} catch {
|
||
return JSON.parse(JSON.stringify(body));
|
||
}
|
||
}
|
||
|
||
function serializeAntigravityRequest(
|
||
provider: string,
|
||
headers: Record<string, string>,
|
||
body: unknown
|
||
): { headers: Record<string, string>; bodyString: string } {
|
||
const serializedBody = cloneAntigravityRequestBody(body);
|
||
|
||
if (!isCliCompatEnabled(provider)) {
|
||
return { headers, bodyString: JSON.stringify(serializedBody) };
|
||
}
|
||
return applyFingerprint(provider, { ...headers }, serializedBody);
|
||
}
|
||
|
||
type AntigravityRequestEnvelope = Record<string, unknown> & {
|
||
project: string;
|
||
model?: string;
|
||
userAgent: "antigravity" | "jetski";
|
||
requestType: "agent" | "image_gen";
|
||
requestId: string;
|
||
request: Record<string, unknown>;
|
||
enabledCreditTypes?: string[];
|
||
};
|
||
|
||
class AntigravityPreResponseTimeoutError extends Error {
|
||
code = ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE;
|
||
status = HTTP_STATUS.GATEWAY_TIMEOUT;
|
||
|
||
constructor(timeoutMs: number, url: string) {
|
||
super(`Antigravity upstream did not return response headers within ${timeoutMs}ms: ${url}`);
|
||
this.name = "TimeoutError";
|
||
}
|
||
}
|
||
|
||
function getAbortErrorCode(error: unknown): string | null {
|
||
if (!error || typeof error !== "object") return null;
|
||
const value = (error as { code?: unknown }).code;
|
||
return typeof value === "string" ? value : null;
|
||
}
|
||
|
||
function isAntigravityPreResponseTimeout(error: unknown): boolean {
|
||
return getAbortErrorCode(error) === ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE;
|
||
}
|
||
|
||
/**
|
||
* Per-account GOOGLE_ONE_AI credits-exhausted tracker.
|
||
* Key: accountId (OAuth subject / email). Value: expiry timestamp.
|
||
* When credits hit 0 we skip the credit retry for CREDITS_EXHAUSTED_TTL_MS.
|
||
*/
|
||
const MAX_CREDITS_EXHAUSTED_ENTRIES = 50;
|
||
const creditsExhaustedUntil = new Map<string, number>();
|
||
|
||
const _creditsExhaustedSweep = setInterval(() => {
|
||
const now = Date.now();
|
||
for (const [key, until] of creditsExhaustedUntil) {
|
||
if (now >= until) creditsExhaustedUntil.delete(key);
|
||
}
|
||
}, 60_000);
|
||
if (typeof _creditsExhaustedSweep === "object" && "unref" in _creditsExhaustedSweep) {
|
||
(_creditsExhaustedSweep as { unref?: () => void }).unref?.();
|
||
}
|
||
|
||
const MAX_CREDIT_BALANCE_ENTRIES = 50;
|
||
const CREDIT_BALANCE_TTL_MS = 5 * 60 * 1000;
|
||
const creditBalanceCache = new Map<string, { balance: number; updatedAt: number }>();
|
||
let creditCacheHydrated = false;
|
||
|
||
function hydrateCreditCacheFromDb(): void {
|
||
if (creditCacheHydrated) return;
|
||
creditCacheHydrated = true;
|
||
try {
|
||
const persisted = getAllPersistedCreditBalances();
|
||
for (const [accountId, balance] of persisted) {
|
||
if (!creditBalanceCache.has(accountId)) {
|
||
creditBalanceCache.set(accountId, { balance, updatedAt: Date.now() });
|
||
}
|
||
}
|
||
} catch {}
|
||
}
|
||
|
||
function evictStaleCreditBalanceEntries(): void {
|
||
const now = Date.now();
|
||
for (const [key, entry] of creditBalanceCache) {
|
||
if (now - entry.updatedAt > CREDIT_BALANCE_TTL_MS) {
|
||
creditBalanceCache.delete(key);
|
||
}
|
||
}
|
||
while (creditBalanceCache.size > MAX_CREDIT_BALANCE_ENTRIES) {
|
||
const oldestKey = creditBalanceCache.keys().next().value;
|
||
if (oldestKey !== undefined) creditBalanceCache.delete(oldestKey);
|
||
else break;
|
||
}
|
||
}
|
||
|
||
const _creditBalanceSweep = setInterval(evictStaleCreditBalanceEntries, 60_000);
|
||
if (typeof _creditBalanceSweep === "object" && "unref" in _creditBalanceSweep) {
|
||
(_creditBalanceSweep as { unref?: () => void }).unref?.();
|
||
}
|
||
|
||
export function getAntigravityRemainingCredits(accountId: string): number | null {
|
||
hydrateCreditCacheFromDb();
|
||
const entry = creditBalanceCache.get(accountId);
|
||
if (!entry) return null;
|
||
if (Date.now() - entry.updatedAt > CREDIT_BALANCE_TTL_MS) {
|
||
creditBalanceCache.delete(accountId);
|
||
return null;
|
||
}
|
||
return entry.balance;
|
||
}
|
||
|
||
export function updateAntigravityRemainingCredits(accountId: string, balance: number): void {
|
||
if (creditBalanceCache.size >= MAX_CREDIT_BALANCE_ENTRIES && !creditBalanceCache.has(accountId)) {
|
||
const oldestKey = creditBalanceCache.keys().next().value;
|
||
if (oldestKey !== undefined) creditBalanceCache.delete(oldestKey);
|
||
}
|
||
creditBalanceCache.set(accountId, { balance, updatedAt: Date.now() });
|
||
try {
|
||
persistCreditBalance(accountId, balance);
|
||
} catch {}
|
||
}
|
||
|
||
function isCreditsExhausted(accountId: string): boolean {
|
||
const until = creditsExhaustedUntil.get(accountId);
|
||
if (!until) return false;
|
||
if (Date.now() >= until) {
|
||
creditsExhaustedUntil.delete(accountId);
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
function markCreditsExhausted(accountId: string): void {
|
||
if (
|
||
creditsExhaustedUntil.size >= MAX_CREDITS_EXHAUSTED_ENTRIES &&
|
||
!creditsExhaustedUntil.has(accountId)
|
||
) {
|
||
const now = Date.now();
|
||
for (const [key, until] of creditsExhaustedUntil) {
|
||
if (now >= until) {
|
||
creditsExhaustedUntil.delete(key);
|
||
}
|
||
}
|
||
if (creditsExhaustedUntil.size >= MAX_CREDITS_EXHAUSTED_ENTRIES) {
|
||
const oldestKey = creditsExhaustedUntil.keys().next().value;
|
||
if (oldestKey !== undefined) creditsExhaustedUntil.delete(oldestKey);
|
||
}
|
||
}
|
||
creditsExhaustedUntil.set(accountId, Date.now() + CREDITS_EXHAUSTED_TTL_MS);
|
||
}
|
||
|
||
/**
|
||
* Persist a quota-exhausted cooldown to the DB for `connectionId` so that
|
||
* cross-request and post-restart routing skips this connection until the
|
||
* cooldown expires. Exported for unit testing. @internal
|
||
*/
|
||
export function markConnectionQuotaExhausted(connectionId: string, retryAfterMs: number): void {
|
||
try {
|
||
setConnectionRateLimitUntil(connectionId, Date.now() + retryAfterMs);
|
||
} catch {
|
||
// DB write failure must never crash the request path
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Accumulate one Antigravity SSE `data:` payload into `collected`. Exported for unit
|
||
* tests (the markdown / candidate-parts extraction branches). @internal
|
||
*/
|
||
|
||
/**
|
||
* Strip provider prefixes (e.g. "antigravity/model" → "model").
|
||
* Ensures the model name sent to the upstream API never contains a routing prefix.
|
||
*
|
||
* `modelIdOverride` (#3786): when the per-request Pro-family fallback chain forces a
|
||
* specific upstream id, pass it here. It is an ALREADY-RESOLVED upstream id, so it bypasses
|
||
* the MITM/static alias resolution and is used verbatim (after prefix stripping).
|
||
*/
|
||
async function cleanModelName(model: string, modelIdOverride?: string): Promise<string> {
|
||
if (modelIdOverride) {
|
||
return modelIdOverride.includes("/") ? modelIdOverride.split("/").pop()! : modelIdOverride;
|
||
}
|
||
if (!model) return model;
|
||
const stripped = model.includes("/") ? model.split("/").pop()! : model;
|
||
let clean = stripped;
|
||
|
||
// 1. Check dynamic MITM aliases first (authoritative after first sync).
|
||
// Built during model sync — contains ONLY currently-available models.
|
||
// Obsolete/removed models are automatically excluded.
|
||
try {
|
||
const mitmAliases = await getMitmAlias("antigravity");
|
||
if (mitmAliases && typeof mitmAliases === "object") {
|
||
const aliases = mitmAliases as Record<string, unknown>;
|
||
const raw = aliases[stripped];
|
||
// Only honor string aliases; corrupted/non-string DB values fall through
|
||
// to the static alias resolution below (never return undefined here).
|
||
if (typeof raw === "string" && raw) {
|
||
// Strip the "antigravity/" prefix if present; use the raw model ID otherwise.
|
||
const PREFIX = "antigravity/";
|
||
clean = raw.startsWith(PREFIX) ? raw.slice(PREFIX.length) : raw;
|
||
}
|
||
}
|
||
} catch {
|
||
// DB not available (build phase, transient error) — fall through to static aliases
|
||
}
|
||
|
||
// 2. Fall back to static aliases if MITM didn't resolve
|
||
if (clean === stripped) {
|
||
clean = resolveAntigravityModelId(clean);
|
||
}
|
||
|
||
return clean;
|
||
}
|
||
|
||
function attachToolNameMap<T>(payload: T, toolNameMap: Map<string, string> | null): T {
|
||
if (!toolNameMap?.size || !payload || typeof payload !== "object") {
|
||
return payload;
|
||
}
|
||
|
||
const copy = Array.isArray(payload) ? ([...payload] as T) : ({ ...(payload as object) } as T);
|
||
Object.defineProperty(copy, "_toolNameMap", {
|
||
value: toolNameMap,
|
||
enumerable: false,
|
||
configurable: true,
|
||
writable: true,
|
||
});
|
||
return copy;
|
||
}
|
||
|
||
function getRequestTargetModel(body: Record<string, unknown>): string {
|
||
const target = body.model;
|
||
return typeof target === "string" && target.length > 0 ? target : "unknown";
|
||
}
|
||
|
||
/**
|
||
* Hard ceiling on `generationConfig.maxOutputTokens` for Antigravity Cloud Code.
|
||
*
|
||
* Ports decolua/9router#779 (lukmanfauzie): VS Code GitHub Copilot Chat in
|
||
* Agent mode regularly requests 32K–65K output tokens, which the Antigravity
|
||
* backend rejects with HTTP 400 "Invalid Argument". 16384 matches the
|
||
* upstream-accepted ceiling confirmed via successful 200 OK runs with
|
||
* claude-sonnet-4-6 and gemini-3.1-pro-high across both Ask and Agent modes.
|
||
*/
|
||
export const MAX_ANTIGRAVITY_OUTPUT_TOKENS = 16384;
|
||
|
||
function applyAntigravityGenerationDefaults(request: Record<string, unknown>): void {
|
||
const generationConfig =
|
||
request.generationConfig && typeof request.generationConfig === "object"
|
||
? (request.generationConfig as Record<string, unknown>)
|
||
: {};
|
||
|
||
if (generationConfig.topK === undefined) {
|
||
generationConfig.topK = 40;
|
||
}
|
||
if (generationConfig.topP === undefined) {
|
||
generationConfig.topP = 1.0;
|
||
}
|
||
|
||
const thinkingConfig =
|
||
generationConfig.thinkingConfig && typeof generationConfig.thinkingConfig === "object"
|
||
? (generationConfig.thinkingConfig as Record<string, unknown>)
|
||
: null;
|
||
const thinkingBudget = Number(thinkingConfig?.thinkingBudget);
|
||
const maxOutputTokens = Number(generationConfig.maxOutputTokens);
|
||
if (
|
||
Number.isFinite(thinkingBudget) &&
|
||
thinkingBudget > 0 &&
|
||
(!Number.isFinite(maxOutputTokens) || maxOutputTokens <= thinkingBudget)
|
||
) {
|
||
generationConfig.maxOutputTokens = Math.floor(thinkingBudget) + 1;
|
||
}
|
||
|
||
// Final cap (after the thinkingBudget bump may have raised the value):
|
||
// GitHub Copilot Agent envelopes commonly carry oversized maxOutputTokens
|
||
// (32K–65K) that trigger upstream 400 "Invalid Argument". Clamp silently
|
||
// — the cap is provider-driven, not client-driven, and only matters when
|
||
// the request would otherwise be rejected outright.
|
||
const finalMax = Number(generationConfig.maxOutputTokens);
|
||
if (Number.isFinite(finalMax) && finalMax > MAX_ANTIGRAVITY_OUTPUT_TOKENS) {
|
||
generationConfig.maxOutputTokens = MAX_ANTIGRAVITY_OUTPUT_TOKENS;
|
||
}
|
||
|
||
request.generationConfig = generationConfig;
|
||
}
|
||
|
||
// Test-only export so the unit suite can exercise the cap logic in isolation
|
||
// without spinning up the full executor.
|
||
export const __test_applyAntigravityGenerationDefaults = applyAntigravityGenerationDefaults;
|
||
|
||
function asRecord(value: unknown): Record<string, unknown> | null {
|
||
return value && typeof value === "object" && !Array.isArray(value)
|
||
? (value as Record<string, unknown>)
|
||
: null;
|
||
}
|
||
|
||
function sanitizeAntigravityGeminiRequest(
|
||
request: Record<string, unknown>
|
||
): Record<string, unknown> {
|
||
const clean: Record<string, unknown> = {};
|
||
|
||
if (Array.isArray(request.contents)) {
|
||
clean.contents = request.contents;
|
||
}
|
||
|
||
if (asRecord(request.systemInstruction)) {
|
||
clean.systemInstruction = request.systemInstruction;
|
||
}
|
||
|
||
clean.generationConfig = asRecord(request.generationConfig)
|
||
? { ...(request.generationConfig as Record<string, unknown>) }
|
||
: {};
|
||
|
||
const geminiTools = buildGeminiTools(request.tools);
|
||
if (geminiTools) {
|
||
clean.tools = geminiTools;
|
||
clean.toolConfig = { functionCallingConfig: { mode: "VALIDATED" } };
|
||
} else if (asRecord(request.toolConfig)) {
|
||
clean.toolConfig = request.toolConfig;
|
||
}
|
||
|
||
if (typeof request.sessionId === "string") {
|
||
clean.sessionId = request.sessionId;
|
||
}
|
||
|
||
// #5003: preserve safetySettings through the Claude-path whitelist so the all-OFF
|
||
// default (or a caller-supplied value) actually reaches Google Cloud Code. Without
|
||
// this the field is dropped and Google applies its own safety defaults that
|
||
// false-flag benign technical prompts as `prohibited_content`.
|
||
if (Array.isArray(request.safetySettings)) {
|
||
clean.safetySettings = request.safetySettings;
|
||
}
|
||
|
||
return clean;
|
||
}
|
||
|
||
/**
|
||
* Ported from decolua/9router#2321 (anki1kr): Vertex AI (used by Antigravity for
|
||
* Claude-branded models) rejects a conversation ending on an assistant turn —
|
||
* "This model does not support assistant message prefill" — so the request must
|
||
* always end on a user turn. Upstream patched `openaiToClaudeRequestForAntigravity`
|
||
* (dead code here, zero callers — see `open-sse/translator/request/openai-to-claude.ts`);
|
||
* this relocates the same strip to the LIVE Antigravity dispatch path, where Claude
|
||
* requests are converted to Gemini `contents` (assistant role is `"model"`, not
|
||
* `"assistant"`). Mirrors the trailing-strip pop-loop already used for Mistral
|
||
* (#3396), Copilot (#5802), and the CC-bridge in `claudeCodeCompatible.ts`.
|
||
*
|
||
* Scoped strictly to the Claude path by the caller (`isClaude` branch only) — native
|
||
* Gemini models via Antigravity must be unaffected, since Vertex-Claude is the only
|
||
* documented rejection surface.
|
||
*
|
||
* Guard: never strip `contents` down to empty — an empty `contents` array is itself
|
||
* an invalid request, so at least one entry (even a lone trailing "model" turn) is
|
||
* always preserved.
|
||
*/
|
||
function stripTrailingAntigravityAssistantTurn(
|
||
request: Record<string, unknown>
|
||
): Record<string, unknown> {
|
||
const contents = request.contents;
|
||
if (!Array.isArray(contents) || contents.length === 0) {
|
||
return request;
|
||
}
|
||
|
||
while (
|
||
contents.length > 1 &&
|
||
(contents[contents.length - 1] as AntigravityContent)?.role === "model"
|
||
) {
|
||
contents.pop();
|
||
}
|
||
|
||
return request;
|
||
}
|
||
|
||
// Test-only export so the unit suite can exercise the strip logic directly.
|
||
export const __test_stripTrailingAntigravityAssistantTurn = stripTrailingAntigravityAssistantTurn;
|
||
|
||
export class AntigravityExecutor extends BaseExecutor {
|
||
constructor() {
|
||
super("antigravity", PROVIDERS.antigravity);
|
||
}
|
||
|
||
buildUrl(model: string, _stream: boolean, urlIndex = 0): string {
|
||
void model;
|
||
const baseUrls = this.getBaseUrls();
|
||
const baseUrl = baseUrls[urlIndex] || baseUrls[0];
|
||
// Always use streaming endpoint — the non-streaming `generateContent` causes
|
||
// upstream 400 errors for some models (e.g. gpt-oss-120b-medium) because the
|
||
// Cloud Code API internally converts to OpenAI format and injects
|
||
// stream_options without setting stream=true. chatCore already handles
|
||
// SSE→JSON conversion for non-streaming client requests.
|
||
return `${baseUrl}/v1internal:streamGenerateContent?alt=sse`;
|
||
}
|
||
|
||
buildHeaders(credentials: AntigravityCredentials, _stream = true): Record<string, string> {
|
||
const raw = {
|
||
"Content-Type": "application/json",
|
||
Authorization: `Bearer ${credentials.accessToken}`,
|
||
"User-Agent": antigravityUserAgent(),
|
||
Accept: "text/event-stream",
|
||
"X-OmniRoute-Source": "omniroute",
|
||
};
|
||
// Scrub proxy/fingerprint headers that reveal non-native traffic
|
||
return scrubProxyAndFingerprintHeaders(raw);
|
||
}
|
||
|
||
async transformRequest(
|
||
model: string,
|
||
body: unknown,
|
||
_stream: boolean,
|
||
credentials: AntigravityCredentials,
|
||
modelIdOverride?: string
|
||
): Promise<AntigravityRequestEnvelope | Response> {
|
||
// Project ID resolution: prefer OAuth-stored projectId over incoming body.project
|
||
// to avoid stale/wrong client-side values causing 404/403 from Cloud Code endpoints.
|
||
// Opt-in escape hatch: set OMNIROUTER_ALLOW_BODY_PROJECT_OVERRIDE=1.
|
||
const normalizeProjectId = (value: unknown): string | null => {
|
||
if (typeof value !== "string") return null;
|
||
const trimmedValue = value.trim();
|
||
return trimmedValue ? trimmedValue : null;
|
||
};
|
||
const bodyRecord = asRecord(body) ?? {};
|
||
const bodyProjectId = normalizeProjectId(bodyRecord.project);
|
||
const credentialsProjectId = normalizeProjectId(credentials?.projectId);
|
||
const providerSpecificProjectId = normalizeProjectId(
|
||
(credentials?.providerSpecificData as Record<string, unknown> | undefined)?.projectId
|
||
);
|
||
const allowBodyProjectOverride = process.env.OMNIROUTE_ALLOW_BODY_PROJECT_OVERRIDE === "1";
|
||
|
||
// Default: prefer OAuth-stored projectId over incoming body.project to avoid
|
||
// stale/wrong client-side values causing 404/403 from Cloud Code endpoints.
|
||
// Opt-in escape hatch: set OMNIROUTE_ALLOW_BODY_PROJECT_OVERRIDE=1.
|
||
let projectId =
|
||
allowBodyProjectOverride && bodyProjectId
|
||
? bodyProjectId
|
||
: credentialsProjectId || providerSpecificProjectId || bodyProjectId;
|
||
|
||
// Auto-discover a missing projectId via loadCodeAssist before failing (#2334/#2541).
|
||
// A freshly re-added Antigravity account can have an empty stored projectId even when
|
||
// its Google account already owns a Cloud Code project (the OAuth-time loadCodeAssist
|
||
// returned empty/transiently failed). Mirror the Cloud Code bootstrap to recover it
|
||
// here — the helper memoizes per access-token, so this is a one-time round-trip.
|
||
if (!projectId && credentials?.accessToken) {
|
||
const discovered = await ensureAntigravityProjectAssigned(credentials.accessToken);
|
||
if (discovered) projectId = discovered;
|
||
}
|
||
|
||
if (!projectId) {
|
||
// (#489) Return a structured error instead of throwing — gives the client a clear signal
|
||
// to show a "Reconnect OAuth" prompt rather than an opaque "Internal Server Error".
|
||
const errorMsg =
|
||
"Missing Google projectId for Antigravity account. Auto-discovery via loadCodeAssist " +
|
||
"found no Cloud Code project. Please reconnect OAuth in Providers → Antigravity (and " +
|
||
"ensure the Google account has completed Gemini Code Assist onboarding).";
|
||
const errorBody = {
|
||
error: {
|
||
message: errorMsg,
|
||
type: "oauth_missing_project_id",
|
||
code: "missing_project_id",
|
||
},
|
||
};
|
||
const resp = new Response(JSON.stringify(errorBody), {
|
||
status: 422,
|
||
headers: { "Content-Type": "application/json" },
|
||
});
|
||
// Returning a Response object signals the executor to stop and forward it
|
||
return resp as unknown as never;
|
||
}
|
||
|
||
// Validate projectId is non-empty and not just whitespace
|
||
const trimmedProjectId = typeof projectId === "string" ? projectId.trim() : projectId;
|
||
if (!trimmedProjectId) {
|
||
const resp = new Response(
|
||
JSON.stringify({
|
||
error: {
|
||
message:
|
||
"Invalid (empty) Google projectId for Antigravity account. " +
|
||
"Please reconnect OAuth in Providers → Antigravity.",
|
||
type: "oauth_missing_project_id",
|
||
code: "missing_project_id",
|
||
},
|
||
}),
|
||
{ status: 422, headers: { "Content-Type": "application/json" } }
|
||
);
|
||
return resp as unknown as never;
|
||
}
|
||
|
||
const upstreamModel = await cleanModelName(model, modelIdOverride);
|
||
const isClaude = upstreamModel.toLowerCase().includes("claude");
|
||
const baseBody = bodyRecord;
|
||
const normalizedBody = shouldStripCloudCodeThinking(this.provider, upstreamModel)
|
||
? stripCloudCodeThinkingConfig(baseBody)
|
||
: baseBody;
|
||
const normalizedRequest = asRecord(normalizedBody.request);
|
||
const rawContents = Array.isArray(normalizedRequest?.contents)
|
||
? normalizedRequest.contents
|
||
: [];
|
||
|
||
// Fix contents for Gemini-compatible Cloud Code requests via Antigravity.
|
||
// Claude-branded Antigravity models use the same streamGenerateContent schema.
|
||
const normalizedContents: AntigravityContent[] =
|
||
rawContents.map((content): AntigravityContent => {
|
||
const c = content as AntigravityChunkContent;
|
||
let role = typeof c.role === "string" ? c.role : "user";
|
||
if (c.parts?.some((p) => p.functionResponse)) {
|
||
role = "user";
|
||
}
|
||
|
||
const hasFunctionCall = c.parts?.some((p) => p.functionCall) || false;
|
||
|
||
const parts =
|
||
c.parts?.filter((p) => {
|
||
if (typeof p.text === "string" && p.text === "") return false;
|
||
if (p.functionCall && !p.functionCall.name) return false;
|
||
|
||
// Only strip if it's NOT our bypass sentinel.
|
||
// Antigravity models (like Gemini) need this sentinel to bypass 400 errors.
|
||
return (
|
||
!p.thought &&
|
||
(hasFunctionCall ||
|
||
!p.thoughtSignature ||
|
||
p.thoughtSignature === "skip_thought_signature_validator")
|
||
);
|
||
}) || [];
|
||
return { ...c, role, parts };
|
||
}) || [];
|
||
|
||
const contents: AntigravityContent[] = [];
|
||
for (const c of normalizedContents) {
|
||
if (!Array.isArray(c.parts) || c.parts.length === 0) continue;
|
||
if (contents.length > 0 && contents[contents.length - 1].role === c.role) {
|
||
contents[contents.length - 1].parts.push(...c.parts);
|
||
} else {
|
||
contents.push(c);
|
||
}
|
||
}
|
||
|
||
const rawTransformedRequest = {
|
||
...normalizedRequest,
|
||
...(contents.length > 0 && { contents }),
|
||
sessionId: getAntigravitySessionId(
|
||
credentials,
|
||
typeof normalizedRequest?.sessionId === "string" ? normalizedRequest.sessionId : undefined
|
||
),
|
||
// #5003: default to all-OFF safety for parity with the native Gemini paths
|
||
// (claude-to-gemini / openai-to-gemini both default to DEFAULT_SAFETY_SETTINGS).
|
||
// Previously this was `undefined`, which JSON.stringify drops, so Google Cloud Code
|
||
// applied its server-side defaults that false-flag benign technical prompts as
|
||
// `prohibited_content` (HTTP 200 + blocked body → terminal combo failover).
|
||
safetySettings: normalizedRequest?.safetySettings ?? DEFAULT_SAFETY_SETTINGS,
|
||
toolConfig:
|
||
Array.isArray(normalizedRequest?.tools) && normalizedRequest.tools.length > 0
|
||
? { functionCallingConfig: { mode: "VALIDATED" } }
|
||
: normalizedRequest?.toolConfig,
|
||
};
|
||
|
||
const transformedRequest = isClaude
|
||
? stripTrailingAntigravityAssistantTurn(sanitizeAntigravityGeminiRequest(rawTransformedRequest))
|
||
: rawTransformedRequest;
|
||
|
||
// Obfuscate sensitive client names in user content (e.g. "OpenCode", "Cursor")
|
||
const requestContents = transformedRequest.contents;
|
||
if (Array.isArray(requestContents)) {
|
||
for (const msg of requestContents) {
|
||
if (Array.isArray(msg.parts)) {
|
||
for (const part of msg.parts) {
|
||
if (typeof part.text === "string") {
|
||
part.text = obfuscateSensitiveWords(part.text);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
applyAntigravityGenerationDefaults(transformedRequest);
|
||
|
||
const {
|
||
project: _project,
|
||
model: _model,
|
||
userAgent: _userAgent,
|
||
requestType: _requestType,
|
||
requestId: _requestId,
|
||
request: _request,
|
||
// #1944: output_config (and the legacy output_format) are Anthropic/Claude-Code-only
|
||
// fields. Google's Cloud Code envelope rejects unknown top-level fields with a 400
|
||
// ("Invalid JSON payload received. Unknown name \"output_config\""), which broke every
|
||
// Claude model served via Antigravity. Drop them so they never reach the envelope.
|
||
output_config: _outputConfig,
|
||
output_format: _outputFormat,
|
||
// #1926: the unified thinking adapter can also set Claude/OpenAI-native thinking fields
|
||
// at the body root. Google rejects them with `400 Bad input: oneOf at '/' not met`
|
||
// (or `Unknown name "thinking"`), breaking every reasoning/thinking model served via
|
||
// Antigravity (e.g. claude-opus-4-x-thinking). Strip the whole thinking family too.
|
||
thinking: _thinking,
|
||
reasoning_effort: _reasoningEffort,
|
||
reasoning: _reasoning,
|
||
enable_thinking: _enableThinking,
|
||
thinking_budget: _thinkingBudget,
|
||
...passthroughFields
|
||
} = normalizedBody;
|
||
|
||
const requestType = _requestType === "image_gen" ? "image_gen" : "agent";
|
||
const envelope: AntigravityRequestEnvelope = {
|
||
project: projectId,
|
||
requestId: generateAntigravityRequestId(),
|
||
request: transformedRequest,
|
||
model: upstreamModel,
|
||
userAgent: getAntigravityEnvelopeUserAgent(credentials),
|
||
requestType,
|
||
...passthroughFields,
|
||
};
|
||
|
||
if (requestType === "agent" && envelope.enabledCreditTypes === undefined) {
|
||
envelope.enabledCreditTypes = ["GOOGLE_ONE_AI"];
|
||
}
|
||
|
||
return envelope;
|
||
}
|
||
|
||
async refreshCredentials(
|
||
credentials: AntigravityCredentials,
|
||
log?: ExecutorLog | null
|
||
): Promise<AntigravityCredentials | null> {
|
||
if (!credentials.refreshToken) return null;
|
||
|
||
try {
|
||
const bodyParams: Record<string, string> = {
|
||
grant_type: "refresh_token",
|
||
refresh_token: credentials.refreshToken,
|
||
};
|
||
// Only include non-empty client_id/client_secret — Google OAuth rejects
|
||
// empty params which raw URLSearchParams produces (buildFormParams semantics).
|
||
if (this.config.clientId) bodyParams.client_id = this.config.clientId;
|
||
if (this.config.clientSecret) bodyParams.client_secret = this.config.clientSecret;
|
||
|
||
const response = await fetch(OAUTH_ENDPOINTS.google.token, {
|
||
method: "POST",
|
||
headers: {
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
Accept: "application/json",
|
||
"User-Agent": antigravityNativeOAuthUserAgent(),
|
||
},
|
||
body: new URLSearchParams(bodyParams),
|
||
});
|
||
|
||
if (!response.ok) {
|
||
// Detect unrecoverable token (invalid_grant = revoked / expired refresh token)
|
||
try {
|
||
const errorBody = (await response.json()) as Record<string, unknown>;
|
||
if (errorBody.error === "invalid_grant") {
|
||
log?.error?.("TOKEN", "Antigravity refresh token revoked. Re-authentication required.");
|
||
return { error: "unrecoverable_refresh_error" } as unknown as AntigravityCredentials;
|
||
}
|
||
} catch {
|
||
// not JSON — fall through
|
||
}
|
||
return null;
|
||
}
|
||
|
||
const tokens = (await response.json()) as Record<string, unknown>;
|
||
log?.info?.("TOKEN", "Antigravity refreshed");
|
||
|
||
return {
|
||
accessToken: typeof tokens.access_token === "string" ? tokens.access_token : undefined,
|
||
refreshToken:
|
||
typeof tokens.refresh_token === "string" && tokens.refresh_token
|
||
? tokens.refresh_token
|
||
: credentials.refreshToken,
|
||
expiresIn: typeof tokens.expires_in === "number" ? tokens.expires_in : undefined,
|
||
projectId: credentials.projectId,
|
||
// Preserve providerSpecificData so a projectId stored there survives the refresh
|
||
// (the onCredentialsRefreshed DB write) instead of being dropped → 422 (#2480).
|
||
providerSpecificData: credentials.providerSpecificData,
|
||
};
|
||
} catch (error) {
|
||
const message = error instanceof Error ? error.message : String(error);
|
||
log?.error?.("TOKEN", `Antigravity refresh error: ${message}`);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
generateSessionId(): string {
|
||
return `-${parseInt(randomUUID().replace(/-/g, "").substring(0, 8), 16) % 9_000_000_000_000_000_000}`;
|
||
}
|
||
|
||
parseRetryHeaders(headers: Headers | null | undefined): number | null {
|
||
if (!headers?.get) return null;
|
||
|
||
const retryAfter = headers.get("retry-after");
|
||
if (retryAfter) {
|
||
const seconds = parseInt(retryAfter, 10);
|
||
if (!isNaN(seconds) && seconds > 0) return seconds * 1000;
|
||
|
||
const date = new Date(retryAfter);
|
||
if (!isNaN(date.getTime())) {
|
||
const diff = date.getTime() - Date.now();
|
||
return diff > 0 ? diff : null;
|
||
}
|
||
}
|
||
|
||
const resetAfter = headers.get("x-ratelimit-reset-after");
|
||
if (resetAfter) {
|
||
const seconds = parseInt(resetAfter, 10);
|
||
if (!isNaN(seconds) && seconds > 0) return seconds * 1000;
|
||
}
|
||
|
||
const resetTimestamp = headers.get("x-ratelimit-reset");
|
||
if (resetTimestamp) {
|
||
const ts = parseInt(resetTimestamp, 10) * 1000;
|
||
const diff = ts - Date.now();
|
||
return diff > 0 ? diff : null;
|
||
}
|
||
|
||
return null;
|
||
}
|
||
|
||
// Parse retry time from Antigravity error message body
|
||
// Format: "Your quota will reset after 2h7m23s" or "Resets in 160h27m24s" or
|
||
// "1h30m" or "45m" or "30s". The optional plural ("resets in") must match too (#1308).
|
||
parseRetryFromErrorMessage(errorMessage: unknown): number | null {
|
||
if (!errorMessage || typeof errorMessage !== "string") return null;
|
||
|
||
const match = errorMessage.match(/resets? (?:after|in) (\d+h)?(\d+m)?(\d+s)?/i);
|
||
if (!match) return null;
|
||
|
||
let totalMs = 0;
|
||
if (match[1]) totalMs += parseInt(match[1]) * 3600 * 1000; // hours
|
||
if (match[2]) totalMs += parseInt(match[2]) * 60 * 1000; // minutes
|
||
if (match[3]) totalMs += parseInt(match[3]) * 1000; // seconds
|
||
|
||
// "reset after 0s" = burst/RPM limit, not quota exhaustion.
|
||
// Return a minimum backoff so the auto-retry loop handles it
|
||
// instead of falling through to the 24h exhaustion classifier.
|
||
if (totalMs === 0) return 2_000; // 2s minimum burst-limit backoff
|
||
|
||
return totalMs;
|
||
}
|
||
|
||
/**
|
||
* Flatten an Antigravity error JSON + raw body text into a single string so
|
||
* isTransientAntigravityError can match against body patterns.
|
||
*/
|
||
extractErrorMessage(errorJson: unknown, bodyText = ""): string {
|
||
const candidates: string[] = [];
|
||
if (errorJson && typeof errorJson === "object") {
|
||
const obj = errorJson as Record<string, unknown>;
|
||
const errField = obj.error;
|
||
if (errField && typeof errField === "object") {
|
||
const msg = (errField as Record<string, unknown>).message;
|
||
if (typeof msg === "string") candidates.push(msg);
|
||
else if (msg != null) candidates.push(JSON.stringify(msg));
|
||
} else if (typeof errField === "string") {
|
||
candidates.push(errField);
|
||
}
|
||
if (typeof obj.message === "string") candidates.push(obj.message);
|
||
}
|
||
if (bodyText) candidates.push(bodyText);
|
||
return candidates.filter(Boolean).join("\n");
|
||
}
|
||
|
||
/**
|
||
* Return true when a status + error message combination should be retried
|
||
* with exponential backoff instead of immediately failing-over to the next URL.
|
||
* 429 is always transient. Transient 5xx statuses (500/502/503/504) are also
|
||
* retried when the body contains a known capacity/traffic/agent pattern.
|
||
*/
|
||
isTransientAntigravityError(status: number, message: string): boolean {
|
||
if (status === HTTP_STATUS.RATE_LIMITED) return true;
|
||
if (ANTIGRAVITY_TRANSIENT_STATUSES.has(status)) return true;
|
||
return ANTIGRAVITY_TRANSIENT_ERROR_PATTERNS.some((p) => p.test(message || ""));
|
||
}
|
||
|
||
/**
|
||
* Collect an SSE streaming response into a single non-streaming JSON response.
|
||
* Parses Gemini-format SSE chunks and assembles text content + usage into one
|
||
* OpenAI-format chat.completion payload.
|
||
*/
|
||
collectStreamToResponse(
|
||
response: Response,
|
||
model: string,
|
||
url: string,
|
||
headers: Record<string, string>,
|
||
transformedBody: Record<string, unknown>,
|
||
log?: ExecutorLog | null,
|
||
signal?: AbortSignal | null
|
||
) {
|
||
if (!response.body) {
|
||
return Promise.resolve({ response, url, headers, transformedBody });
|
||
}
|
||
|
||
const reader = response.body.getReader();
|
||
const decoder = new TextDecoder();
|
||
const logger = log || undefined;
|
||
|
||
const SSE_COLLECT_TIMEOUT_MS = 120_000;
|
||
|
||
const collect = async () => {
|
||
const collected: AntigravityCollectedStream = {
|
||
textContent: "",
|
||
finishReason: "stop",
|
||
toolCalls: [],
|
||
usage: null,
|
||
remainingCredits: null,
|
||
};
|
||
const partialLine = { value: "" };
|
||
let timedOut = false;
|
||
const timeout = AbortSignal.timeout(SSE_COLLECT_TIMEOUT_MS);
|
||
try {
|
||
while (true) {
|
||
if (signal?.aborted) throw new Error("Request aborted during SSE collection");
|
||
const { done, value } = await Promise.race([
|
||
reader.read(),
|
||
new Promise<never>((_, reject) =>
|
||
timeout.addEventListener(
|
||
"abort",
|
||
() => reject(new Error("SSE collection timed out")),
|
||
{ once: true }
|
||
)
|
||
),
|
||
]);
|
||
if (done) break;
|
||
processAntigravitySSEText(
|
||
decoder.decode(value, { stream: true }),
|
||
partialLine,
|
||
collected,
|
||
logger
|
||
);
|
||
}
|
||
} catch (err) {
|
||
const msg = err?.message || String(err);
|
||
timedOut = msg.includes("timed out");
|
||
log?.warn?.("SSE_COLLECT", `Error collecting SSE stream: ${msg}`);
|
||
// Cancel the stream to prevent locking the socket in Undici pool
|
||
try {
|
||
reader.releaseLock();
|
||
} catch (_) {}
|
||
try {
|
||
response.body?.cancel().catch(() => {});
|
||
} catch (_) {}
|
||
} finally {
|
||
try {
|
||
reader.releaseLock();
|
||
} catch (_) {}
|
||
}
|
||
processAntigravitySSEText(decoder.decode(), partialLine, collected, logger);
|
||
flushAntigravitySSEText(partialLine, collected, logger);
|
||
|
||
const result = {
|
||
id: `chatcmpl-${Date.now()}-${crypto.randomUUID().slice(0, 8)}`,
|
||
object: "chat.completion",
|
||
created: Math.floor(Date.now() / 1000),
|
||
model,
|
||
choices: [
|
||
{
|
||
index: 0,
|
||
message:
|
||
collected.toolCalls.length > 0
|
||
? {
|
||
role: "assistant",
|
||
content: collected.textContent || null,
|
||
tool_calls: collected.toolCalls,
|
||
}
|
||
: { role: "assistant", content: collected.textContent },
|
||
finish_reason: timedOut
|
||
? "length"
|
||
: collected.toolCalls.length > 0
|
||
? "tool_calls"
|
||
: collected.finishReason,
|
||
},
|
||
],
|
||
...(collected.usage && { usage: collected.usage }),
|
||
// Expose credit balance for upstream consumers (usage service, dashboard)
|
||
...(collected.remainingCredits && { _remainingCredits: collected.remainingCredits }),
|
||
};
|
||
|
||
const syntheticStatus = timedOut ? 504 : response.status;
|
||
const syntheticResponse = new Response(JSON.stringify(result), {
|
||
status: syntheticStatus,
|
||
statusText: timedOut ? "Gateway Timeout" : response.statusText,
|
||
headers: [["Content-Type", "application/json"]],
|
||
});
|
||
|
||
return { response: syntheticResponse, url, headers, transformedBody };
|
||
};
|
||
|
||
return collect();
|
||
}
|
||
|
||
/**
|
||
* #3786 — Drive the per-request Pro-family upstream-id FALLBACK CHAIN.
|
||
*
|
||
* The upstream silently renamed the Gemini 3.1 Pro-high id (HTTP 400 on the old id) and the
|
||
* live id cannot be known from static analysis (competitor proxies disagree). When the
|
||
* resolved upstream id has a fallback chain (see ANTIGRAVITY_PRO_FALLBACK_CHAINS) we try the
|
||
* requested id first and, ONLY on a 400, retry the next candidate until one succeeds (2xx)
|
||
* or the chain is exhausted — then the original 400 surfaces (sanitized, hard rule #12).
|
||
*
|
||
* Off the happy path entirely: a model with no chain, or whose first id is not a 400, makes
|
||
* exactly the same single call as before (zero extra upstream requests).
|
||
*/
|
||
async execute(input: ExecuteInput) {
|
||
await resolveAntigravityVersion();
|
||
|
||
// Look up the chain by the NORMALLY-resolved upstream id (honours MITM/static aliases).
|
||
// If a MITM alias remapped the id away from a known Pro tier, no chain applies → fast path.
|
||
const resolvedUpstreamId = await cleanModelName(input.model);
|
||
const chain = getAntigravityModelFallbacks(resolvedUpstreamId);
|
||
|
||
if (chain.length <= 1) {
|
||
// No fallback chain (flash, claude, plain pro, unknown) → single attempt, unchanged.
|
||
return this.executeOnce(input);
|
||
}
|
||
|
||
let firstResult: Awaited<ReturnType<AntigravityExecutor["executeOnce"]>> | null = null;
|
||
for (let i = 0; i < chain.length; i++) {
|
||
const candidate = chain[i];
|
||
const result = await this.executeOnce(input, candidate);
|
||
|
||
// Success (or any non-400) on a candidate → return immediately.
|
||
if (result.response.status !== HTTP_STATUS.BAD_REQUEST) {
|
||
return result;
|
||
}
|
||
|
||
// Remember the FIRST 400 so the exhausted-chain case surfaces the original error.
|
||
if (i === 0) firstResult = result;
|
||
|
||
const isLast = i === chain.length - 1;
|
||
if (!isLast) {
|
||
input.log?.debug?.(
|
||
"AG_PRO_FALLBACK",
|
||
`400 on "${candidate}" — retrying with next Pro candidate "${chain[i + 1]}"`
|
||
);
|
||
continue;
|
||
}
|
||
|
||
// Chain exhausted: surface the FIRST candidate's sanitized 400.
|
||
input.log?.warn?.(
|
||
"AG_PRO_FALLBACK",
|
||
`Pro fallback chain exhausted (all ${chain.length} candidates 400'd) for "${resolvedUpstreamId}"`
|
||
);
|
||
return firstResult ?? result;
|
||
}
|
||
|
||
// Unreachable (loop always returns), but keeps the type checker happy.
|
||
return firstResult ?? this.executeOnce(input);
|
||
}
|
||
|
||
/**
|
||
* #3786 — Run the request once for a SINGLE resolved upstream model id. The Pro-family
|
||
* fallback chain in `execute()` calls this per candidate (`modelIdOverride`), retrying the
|
||
* next id on a 400. `modelIdOverride === undefined` is the normal (non-chain) path and
|
||
* preserves the prior behavior exactly. Returns the executor result plus the upstream
|
||
* status of the first response so `execute()` can decide whether to fall through. @internal
|
||
*/
|
||
private async executeOnce(
|
||
{ model, body, stream, credentials, signal, log, upstreamExtraHeaders }: ExecuteInput,
|
||
modelIdOverride?: string
|
||
) {
|
||
await resolveAntigravityVersion();
|
||
const fallbackCount = this.getFallbackCount();
|
||
let lastError = null;
|
||
let lastStatus = 0;
|
||
const MAX_AUTO_RETRIES = 3;
|
||
const retryAttemptsByUrl: Record<number, number> = {}; // Track retry attempts per URL
|
||
|
||
// Always stream upstream — buildUrl always returns the streaming endpoint.
|
||
// For non-streaming clients, we collect the SSE below and return a synthetic
|
||
// non-streaming Response so chatCore's non-streaming path stays unchanged.
|
||
const upstreamStream = true;
|
||
|
||
// Account ID for credits tracking.
|
||
// Use connectionId as the stable cache key — it's available in both the executor
|
||
// (via credentials.connectionId) and the usage fetcher (via connection.id).
|
||
// The email-based key was unreliable because email isn't always on the credentials object.
|
||
const accountId: string = credentials?.connectionId || "unknown";
|
||
|
||
// Resolve credits mode once per execute() call. "always" injects
|
||
// enabledCreditTypes: ["GOOGLE_ONE_AI"] on the first request so the
|
||
// preflight normal call is skipped entirely.
|
||
const creditsMode = getCreditsMode();
|
||
const useCreditsFirst = shouldUseCreditsFirst(credentials?.accessToken || "", creditsMode);
|
||
|
||
const fetchWithReadinessTimeout = async (
|
||
url: string,
|
||
init: RequestInit,
|
||
timeoutMs = STREAM_READINESS_TIMEOUT_MS
|
||
): Promise<Response> => {
|
||
const boundedTimeoutMs = Math.max(0, Math.floor(timeoutMs));
|
||
if (boundedTimeoutMs <= 0) {
|
||
return fetch(url, init);
|
||
}
|
||
|
||
const timeoutController = new AbortController();
|
||
let timeoutId: ReturnType<typeof setTimeout> | null = setTimeout(() => {
|
||
timeoutController.abort(new AntigravityPreResponseTimeoutError(boundedTimeoutMs, url));
|
||
}, boundedTimeoutMs);
|
||
|
||
const existingSignal = init.signal instanceof AbortSignal ? init.signal : null;
|
||
const combinedSignal = existingSignal
|
||
? mergeAbortSignals(existingSignal, timeoutController.signal)
|
||
: timeoutController.signal;
|
||
|
||
try {
|
||
return await fetch(url, { ...init, signal: combinedSignal });
|
||
} catch (error) {
|
||
if (
|
||
timeoutController.signal.aborted &&
|
||
isAntigravityPreResponseTimeout(timeoutController.signal.reason)
|
||
) {
|
||
throw timeoutController.signal.reason;
|
||
}
|
||
throw error;
|
||
} finally {
|
||
if (timeoutId) {
|
||
clearTimeout(timeoutId);
|
||
timeoutId = null;
|
||
}
|
||
}
|
||
};
|
||
|
||
for (let urlIndex = 0; urlIndex < fallbackCount; urlIndex++) {
|
||
const url = this.buildUrl(model, upstreamStream, urlIndex);
|
||
const headers = this.buildHeaders(credentials, upstreamStream);
|
||
mergeUpstreamExtraHeaders(headers, upstreamExtraHeaders);
|
||
const transformed = await this.transformRequest(
|
||
model,
|
||
body,
|
||
upstreamStream,
|
||
credentials,
|
||
modelIdOverride
|
||
);
|
||
let requestToolNameMap: Map<string, string> | null = null;
|
||
|
||
if (transformed instanceof Response) {
|
||
return { response: transformed, url, headers, transformedBody: body };
|
||
}
|
||
|
||
let transformedBody: Record<string, unknown> = transformed;
|
||
|
||
if (transformedBody && typeof transformedBody === "object") {
|
||
const cloaked = cloakAntigravityToolPayload(transformedBody);
|
||
transformedBody = cloaked.body;
|
||
requestToolNameMap = cloaked.toolNameMap;
|
||
}
|
||
|
||
// Credits-first: inject GOOGLE_ONE_AI upfront so we never try the normal
|
||
// quota path. If credits are exhausted / disabled shouldUseCreditsFirst()
|
||
// returns false and we fall back to the legacy retry-on-429 flow.
|
||
if (useCreditsFirst) {
|
||
transformedBody = injectCreditsField(transformedBody);
|
||
log?.debug?.("AG_CREDITS", "Credits-first enabled (ANTIGRAVITY_CREDITS=always)");
|
||
}
|
||
|
||
// Initialize retry counter for this URL
|
||
if (!retryAttemptsByUrl[urlIndex]) {
|
||
retryAttemptsByUrl[urlIndex] = 0;
|
||
}
|
||
|
||
try {
|
||
const serializedRequest = serializeAntigravityRequest(
|
||
this.provider,
|
||
headers,
|
||
transformedBody
|
||
);
|
||
let finalHeaders = serializedRequest.headers;
|
||
const capture = (h: Record<string, string>, s: string) =>
|
||
prl.captureCurrentProviderBody(url, h, s, log);
|
||
const clientProfile = applyAntigravityClientProfileHeaders(
|
||
finalHeaders,
|
||
credentials,
|
||
transformedBody
|
||
);
|
||
|
||
log?.debug?.(
|
||
"TELEMETRY",
|
||
`[Antigravity] Execute - URL: ${url}, Model: ${model}, Target: ${getRequestTargetModel(transformedBody)}, RetryAttempt: ${retryAttemptsByUrl[urlIndex]}`
|
||
);
|
||
|
||
// Dump outgoing headers (mask Authorization) and envelope shape for debugging
|
||
if (log?.debug) {
|
||
const safeHeaders = { ...finalHeaders };
|
||
if (safeHeaders["Authorization"]) safeHeaders["Authorization"] = "Bearer ***";
|
||
log.debug("AG_REQUEST_HEADERS", JSON.stringify(safeHeaders));
|
||
|
||
const envelope = transformedBody as Record<string, unknown>;
|
||
const requestInner = envelope.request as Record<string, unknown> | undefined;
|
||
log.debug(
|
||
"AG_REQUEST_ENVELOPE",
|
||
JSON.stringify({
|
||
fieldOrder: Object.keys(envelope),
|
||
project: envelope.project,
|
||
requestId: envelope.requestId,
|
||
model: envelope.model,
|
||
userAgent: envelope.userAgent,
|
||
requestType: envelope.requestType,
|
||
enabledCreditTypes: envelope.enabledCreditTypes,
|
||
clientProfile,
|
||
sessionId: requestInner?.sessionId,
|
||
generationConfig: requestInner?.generationConfig,
|
||
})
|
||
);
|
||
}
|
||
|
||
await capture(finalHeaders, serializedRequest.bodyString);
|
||
let response = await fetchWithReadinessTimeout(url, {
|
||
method: "POST",
|
||
headers: finalHeaders,
|
||
body: getChunkedOrFixedBody(serializedRequest.bodyString, stream),
|
||
...(stream ? { duplex: "half" } : {}),
|
||
signal,
|
||
});
|
||
|
||
if (response.status === HTTP_STATUS.FORBIDDEN && finalHeaders["x-goog-user-project"]) {
|
||
const retryHeaders = { ...finalHeaders };
|
||
removeHeaderCaseInsensitive(retryHeaders, "x-goog-user-project");
|
||
log?.debug?.("RETRY", "403 with x-goog-user-project, retrying once without it");
|
||
await capture(retryHeaders, serializedRequest.bodyString);
|
||
response = await fetchWithReadinessTimeout(url, {
|
||
method: "POST",
|
||
headers: retryHeaders,
|
||
body: getChunkedOrFixedBody(serializedRequest.bodyString, stream),
|
||
...(stream ? { duplex: "half" } : {}),
|
||
signal,
|
||
});
|
||
finalHeaders = retryHeaders;
|
||
}
|
||
|
||
if (!response.ok) {
|
||
log?.warn?.(
|
||
"TELEMETRY",
|
||
`[Antigravity] Error Response - URL: ${url}, Status: ${response.status}, Model: ${model}`
|
||
);
|
||
}
|
||
|
||
// Parse retry time for 429/503 responses
|
||
let retryMs: number | null = null;
|
||
|
||
if (
|
||
response.status === HTTP_STATUS.RATE_LIMITED ||
|
||
response.status === HTTP_STATUS.SERVICE_UNAVAILABLE
|
||
) {
|
||
// Try to get retry time from headers first
|
||
retryMs = this.parseRetryHeaders(response.headers);
|
||
|
||
// If no retry time in headers, try to parse from error message body
|
||
if (!retryMs) {
|
||
try {
|
||
const errorBody = await response.clone().text();
|
||
const errorJson = JSON.parse(errorBody);
|
||
let errorMessage = errorJson?.error?.message || errorJson?.message || "";
|
||
if (errorJson?.error?.details && Array.isArray(errorJson.error.details)) {
|
||
for (const detail of errorJson.error.details) {
|
||
if (detail?.reason) {
|
||
errorMessage += ` ${detail.reason}`;
|
||
}
|
||
}
|
||
}
|
||
|
||
// 1. Try to parse explicit retry time from message
|
||
const parsedRetryMs = this.parseRetryFromErrorMessage(errorMessage);
|
||
|
||
// 2. Classify 429 (pass header-parsed retry hint as fallback
|
||
// signal — multi-hour Retry-After upgrades rate_limited to
|
||
// quota_exhausted so the GOOGLE_ONE_AI credits retry fires).
|
||
const effectiveRetryHintMs = retryMs ?? parsedRetryMs ?? null;
|
||
const category = classify429(errorMessage);
|
||
|
||
// 3. Decide final retry time BEFORE the credits retry so that
|
||
// full_quota_exhausted can skip the credits attempt entirely
|
||
// (avoids ~41s hold on an already-exhausted account) and
|
||
// persist the cooldown to DB for post-restart routing.
|
||
const decision: Decision = decide429(category, parsedRetryMs);
|
||
retryMs = decision.retryAfterMs;
|
||
log?.debug?.(
|
||
"AG_429",
|
||
`Category: ${category}, Decision: ${decision.kind} — ${decision.reason}`
|
||
);
|
||
|
||
if (decision.kind === "full_quota_exhausted" && retryMs) {
|
||
markConnectionQuotaExhausted(accountId, retryMs);
|
||
}
|
||
|
||
const creditsAlreadyInjected =
|
||
(transformedBody as { enabledCreditTypes?: unknown }).enabledCreditTypes != null;
|
||
|
||
if (category === "quota_exhausted" && creditsAlreadyInjected) {
|
||
handleCreditsFailure(credentials?.accessToken || "");
|
||
log?.warn?.("AG_CREDITS", "Credits-first request 429'd — credits likely exhausted");
|
||
markCreditsExhausted(accountId);
|
||
}
|
||
|
||
if (
|
||
category === "quota_exhausted" &&
|
||
decision.kind !== "full_quota_exhausted" &&
|
||
!creditsAlreadyInjected &&
|
||
shouldRetryWithCredits(credentials?.accessToken || "", creditsMode !== "off")
|
||
) {
|
||
log?.info?.("AG_CREDITS", "Retrying with Google One AI credits");
|
||
const creditsBody = injectCreditsField(transformedBody);
|
||
const serializedCreditsRequest = serializeAntigravityRequest(
|
||
this.provider,
|
||
headers,
|
||
creditsBody
|
||
);
|
||
const finalCreditsHeaders = serializedCreditsRequest.headers;
|
||
try {
|
||
await capture(finalCreditsHeaders, serializedCreditsRequest.bodyString);
|
||
const creditsResp = await fetchWithReadinessTimeout(url, {
|
||
method: "POST",
|
||
headers: finalCreditsHeaders,
|
||
body: getChunkedOrFixedBody(serializedCreditsRequest.bodyString, stream),
|
||
...(stream ? { duplex: "half" } : {}),
|
||
signal,
|
||
});
|
||
if (creditsResp.ok || creditsResp.status !== HTTP_STATUS.RATE_LIMITED) {
|
||
log?.info?.("AG_CREDITS", `Credits retry succeeded: ${creditsResp.status}`);
|
||
if (!stream) {
|
||
const collected = await this.collectStreamToResponse(
|
||
creditsResp,
|
||
model,
|
||
url,
|
||
finalCreditsHeaders,
|
||
creditsBody,
|
||
log,
|
||
signal
|
||
);
|
||
// Parse _remainingCredits from the synthetic response and cache
|
||
try {
|
||
const syntheticJson = await collected.response.clone().json();
|
||
const rc = syntheticJson?._remainingCredits;
|
||
if (Array.isArray(rc)) {
|
||
const googleCredit = rc.find((c) => c.creditType === "GOOGLE_ONE_AI");
|
||
if (googleCredit) {
|
||
const balance = parseInt(googleCredit.creditAmount, 10);
|
||
if (!isNaN(balance))
|
||
updateAntigravityRemainingCredits(accountId, balance);
|
||
}
|
||
}
|
||
} catch {
|
||
/**/
|
||
}
|
||
return {
|
||
...collected,
|
||
transformedBody: attachToolNameMap(creditsBody, requestToolNameMap),
|
||
};
|
||
}
|
||
return {
|
||
response: creditsResp,
|
||
url,
|
||
headers: finalCreditsHeaders,
|
||
transformedBody: attachToolNameMap(creditsBody, requestToolNameMap),
|
||
};
|
||
}
|
||
|
||
// Credit retry also 429'd
|
||
handleCreditsFailure(credentials?.accessToken || "");
|
||
log?.warn?.("AG_CREDITS", "Credits retry also 429'd");
|
||
|
||
// Also mark in our legacy exhaustion map to avoid retrying other routes
|
||
markCreditsExhausted(accountId);
|
||
} catch (creditsErr) {
|
||
handleCreditsFailure(credentials?.accessToken || "");
|
||
log?.warn?.("AG_CREDITS", `Credits retry failed: ${creditsErr}`);
|
||
}
|
||
}
|
||
} catch (e) {
|
||
// Ignore parse errors, will fall back to exponential backoff
|
||
}
|
||
}
|
||
|
||
// Bounded short-retry: a non-null retryAfterMs ≤ 60s covers nearly every
|
||
// 429 (decide429 returns 2s/5s/60s defaults), so this branch MUST share the
|
||
// per-URL attempt counter. Without the bound a persistent 429 loops forever
|
||
// on the same endpoint/account (urlIndex-- cancels the loop's urlIndex++) and
|
||
// never returns the 429 to the account-fallback layer in chat.ts.
|
||
if (
|
||
retryMs &&
|
||
retryMs <= LONG_RETRY_THRESHOLD_MS &&
|
||
retryAttemptsByUrl[urlIndex] < MAX_AUTO_RETRIES
|
||
) {
|
||
retryAttemptsByUrl[urlIndex]++;
|
||
const effectiveRetryMs = Math.min(retryMs, MAX_RETRY_AFTER_MS);
|
||
log?.debug?.(
|
||
"RETRY",
|
||
`${response.status} retry ${retryAttemptsByUrl[urlIndex]}/${MAX_AUTO_RETRIES} with Retry-After: ${Math.ceil(effectiveRetryMs / 1000)}s, waiting...`
|
||
);
|
||
await new Promise((resolve) => setTimeout(resolve, effectiveRetryMs));
|
||
urlIndex--;
|
||
continue;
|
||
}
|
||
|
||
// Auto retry for 429 (no Retry-After) or transient 5xx errors.
|
||
// For 5xx we read the body to detect known transient patterns
|
||
// ("Agent execution terminated due to error", "high traffic", "capacity").
|
||
if ((!retryMs || retryMs === 0) && retryAttemptsByUrl[urlIndex] < MAX_AUTO_RETRIES) {
|
||
let shouldAutoRetry = response.status === HTTP_STATUS.RATE_LIMITED;
|
||
if (!shouldAutoRetry && ANTIGRAVITY_TRANSIENT_STATUSES.has(response.status)) {
|
||
try {
|
||
const errBody = await response.clone().text();
|
||
let errJson: unknown = null;
|
||
try {
|
||
errJson = errBody ? JSON.parse(errBody) : null;
|
||
} catch {
|
||
// non-JSON body — fall through to pattern match against raw text
|
||
}
|
||
const errMsg = this.extractErrorMessage(errJson, errBody);
|
||
shouldAutoRetry = this.isTransientAntigravityError(response.status, errMsg);
|
||
} catch {
|
||
// ignore body read errors
|
||
}
|
||
}
|
||
if (shouldAutoRetry) {
|
||
retryAttemptsByUrl[urlIndex]++;
|
||
// Exponential backoff: 2s, 4s, 8s… capped per-status
|
||
const cap =
|
||
response.status === HTTP_STATUS.RATE_LIMITED
|
||
? MAX_RETRY_AFTER_MS
|
||
: ANTIGRAVITY_TRANSIENT_RETRY_MAX_MS;
|
||
const backoffMs = Math.min(1000 * 2 ** retryAttemptsByUrl[urlIndex], cap);
|
||
log?.debug?.(
|
||
"RETRY",
|
||
`${response.status} transient auto retry ${retryAttemptsByUrl[urlIndex]}/${MAX_AUTO_RETRIES} after ${backoffMs / 1000}s`
|
||
);
|
||
await new Promise((resolve) => setTimeout(resolve, backoffMs));
|
||
urlIndex--;
|
||
continue;
|
||
}
|
||
}
|
||
|
||
log?.debug?.(
|
||
"RETRY",
|
||
`${response.status}, Retry-After ${retryMs ? `too long (${Math.ceil(retryMs / 1000)}s)` : "missing"}, trying fallback`
|
||
);
|
||
lastStatus = response.status;
|
||
|
||
if (urlIndex + 1 < fallbackCount) {
|
||
continue;
|
||
}
|
||
}
|
||
|
||
if (this.shouldRetry(response.status, urlIndex)) {
|
||
log?.debug?.("RETRY", `${response.status} on ${url}, trying fallback ${urlIndex + 1}`);
|
||
lastStatus = response.status;
|
||
continue;
|
||
}
|
||
|
||
// If we have a 429 with long retry time, embed it in response body
|
||
if (
|
||
response.status === HTTP_STATUS.RATE_LIMITED &&
|
||
retryMs &&
|
||
retryMs > LONG_RETRY_THRESHOLD_MS
|
||
) {
|
||
try {
|
||
const respBody = await response.clone().text();
|
||
let obj;
|
||
try {
|
||
obj = JSON.parse(respBody);
|
||
} catch {
|
||
obj = {};
|
||
}
|
||
obj.retryAfterMs = retryMs;
|
||
const modifiedBody = JSON.stringify(obj);
|
||
const modifiedResponse = new Response(modifiedBody, {
|
||
status: response.status,
|
||
headers: response.headers,
|
||
});
|
||
return {
|
||
response: modifiedResponse,
|
||
url,
|
||
headers: finalHeaders,
|
||
transformedBody: attachToolNameMap(transformedBody, requestToolNameMap),
|
||
};
|
||
} catch (err) {
|
||
log?.warn?.("RETRY", `Failed to embed retryAfterMs: ${err}`);
|
||
// Fall back to original response
|
||
}
|
||
}
|
||
|
||
// For non-streaming clients, collect the SSE stream and return a synthetic
|
||
// non-streaming Response so chatCore doesn't need to handle SSE conversion.
|
||
if (!stream) {
|
||
// #3229: surface a real upstream error instead of masking a 4xx/5xx as an
|
||
// empty `chat.completion` envelope (collectStreamToResponse synthesizes a
|
||
// success-shaped body when the upstream returned no SSE data).
|
||
if (!response.ok) {
|
||
const rawBody = await response
|
||
.clone()
|
||
.text()
|
||
.catch(() => "");
|
||
const errorBody = buildAntigravityUpstreamError(
|
||
response.status,
|
||
response.statusText,
|
||
rawBody
|
||
);
|
||
return {
|
||
response: new Response(JSON.stringify(errorBody), {
|
||
status: response.status,
|
||
headers: { "Content-Type": "application/json" },
|
||
}),
|
||
url,
|
||
headers: finalHeaders,
|
||
transformedBody: attachToolNameMap(transformedBody, requestToolNameMap),
|
||
};
|
||
}
|
||
const collected = await this.collectStreamToResponse(
|
||
response,
|
||
model,
|
||
url,
|
||
finalHeaders,
|
||
transformedBody,
|
||
log,
|
||
signal
|
||
);
|
||
// When credits were injected (credits-first or credits-retry), the
|
||
// synthetic body contains _remainingCredits — mirror it into the
|
||
// balance cache so the dashboard stays fresh.
|
||
try {
|
||
const syntheticJson = await collected.response.clone().json();
|
||
const rc = syntheticJson?._remainingCredits;
|
||
if (Array.isArray(rc)) {
|
||
const googleCredit = rc.find(
|
||
(c: { creditType?: string }) => c?.creditType === "GOOGLE_ONE_AI"
|
||
);
|
||
if (googleCredit) {
|
||
const balance = parseInt(googleCredit.creditAmount, 10);
|
||
if (!isNaN(balance)) updateAntigravityRemainingCredits(accountId, balance);
|
||
}
|
||
}
|
||
} catch {
|
||
/* balance cache is best-effort */
|
||
}
|
||
return {
|
||
...collected,
|
||
transformedBody: attachToolNameMap(transformedBody, requestToolNameMap),
|
||
};
|
||
}
|
||
|
||
// Streaming path: wrap the response body in a pass-through TransformStream
|
||
// that extracts remainingCredits from the final SSE chunk(s) without
|
||
// consuming the stream. The client receives the unmodified SSE data.
|
||
if (response.body) {
|
||
// If the downstream client aborts, cancel the upstream fetch body immediately
|
||
// to release the socket back to the Undici agent pool and prevent memory leaks.
|
||
if (signal) {
|
||
const abortHandler = () => {
|
||
try {
|
||
response.body?.cancel().catch(() => {});
|
||
} catch (_) {}
|
||
};
|
||
if (signal.aborted) {
|
||
abortHandler();
|
||
} else {
|
||
signal.addEventListener("abort", abortHandler, { once: true });
|
||
}
|
||
}
|
||
|
||
let sseBuffer = "";
|
||
const decoder = new TextDecoder(); // Singleton for correct streaming decode
|
||
const MAX_BUFFER_SIZE = 16 * 1024; // Limit to prevent OOM on large streams
|
||
|
||
const passThrough = new TransformStream(
|
||
{
|
||
transform(chunk, controller) {
|
||
controller.enqueue(chunk);
|
||
// Accumulate text to scan for remainingCredits
|
||
try {
|
||
const text = decoder.decode(chunk, { stream: true });
|
||
sseBuffer += text;
|
||
// Limit buffer size to prevent unbounded growth
|
||
// Truncate only after a complete newline to avoid splitting SSE lines mid-payload
|
||
if (sseBuffer.length > MAX_BUFFER_SIZE) {
|
||
const lastNewline = sseBuffer.lastIndexOf(
|
||
"\n",
|
||
sseBuffer.length - MAX_BUFFER_SIZE
|
||
);
|
||
if (lastNewline !== -1) {
|
||
sseBuffer = sseBuffer.slice(lastNewline + 1);
|
||
} else {
|
||
// No newline found in discard region — buffer contains an incomplete SSE line.
|
||
// Discard it entirely to avoid returning malformed data; the remainingCredits
|
||
// parser won't find valid data in a truncated line anyway.
|
||
sseBuffer = "";
|
||
}
|
||
}
|
||
} catch {
|
||
/* decoding best-effort */
|
||
}
|
||
},
|
||
flush() {
|
||
// Final decode for any remaining bytes
|
||
try {
|
||
const text = decoder.decode(); // Flush pending bytes
|
||
sseBuffer += text;
|
||
} catch {
|
||
/* decoding best-effort */
|
||
}
|
||
|
||
// Parse the accumulated SSE data for remainingCredits
|
||
try {
|
||
const lines = sseBuffer.split("\n");
|
||
for (const line of lines) {
|
||
const trimmed = line.trim();
|
||
if (!trimmed.startsWith("data:")) continue;
|
||
const payload = trimmed.slice(5).trim();
|
||
if (!payload || payload === "[DONE]") continue;
|
||
try {
|
||
const parsed = JSON.parse(payload);
|
||
if (Array.isArray(parsed?.remainingCredits)) {
|
||
const googleCredit = parsed.remainingCredits.find((c: unknown) => {
|
||
const credit = asRecord(c);
|
||
return credit?.creditType === "GOOGLE_ONE_AI";
|
||
}) as AntigravityCreditEntry | undefined;
|
||
if (googleCredit) {
|
||
const balance = parseInt(String(googleCredit.creditAmount ?? ""), 10);
|
||
if (!isNaN(balance)) {
|
||
updateAntigravityRemainingCredits(accountId, balance);
|
||
}
|
||
}
|
||
}
|
||
} catch {
|
||
/* skip malformed lines */
|
||
}
|
||
}
|
||
} catch {
|
||
/* credits extraction is best-effort */
|
||
}
|
||
sseBuffer = "";
|
||
},
|
||
},
|
||
{ highWaterMark: 16384 },
|
||
{ highWaterMark: 16384 }
|
||
);
|
||
const tappedBody = response.body.pipeThrough(passThrough);
|
||
const tappedResponse = new Response(tappedBody, {
|
||
status: response.status,
|
||
statusText: response.statusText,
|
||
headers: response.headers,
|
||
});
|
||
return {
|
||
response: tappedResponse,
|
||
url,
|
||
headers: finalHeaders,
|
||
transformedBody: attachToolNameMap(transformedBody, requestToolNameMap),
|
||
};
|
||
}
|
||
|
||
return {
|
||
response,
|
||
url,
|
||
headers: finalHeaders,
|
||
transformedBody: attachToolNameMap(transformedBody, requestToolNameMap),
|
||
};
|
||
} catch (error) {
|
||
lastError = error;
|
||
log?.error?.(
|
||
"TELEMETRY",
|
||
`[Antigravity] Network/Fetch Error - URL: ${url}, Model: ${model}, Error: ${error instanceof Error ? error.message : String(error)}`
|
||
);
|
||
if (urlIndex + 1 < fallbackCount) {
|
||
log?.debug?.("RETRY", `Error on ${url}, trying fallback ${urlIndex + 1}`);
|
||
continue;
|
||
}
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
throw lastError || new Error(`All ${fallbackCount} URLs failed with status ${lastStatus}`);
|
||
}
|
||
}
|
||
|
||
export default AntigravityExecutor;
|