pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-26 09:52:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(antigravity): streaming passthrough for non-streaming clients (#7408)

Browse Source 
* chore(ci): add .mergify.yml to main — Mergify only reads config from the default branch (#7168)

* fix(ci): add the auto-enqueue pull_request_rule to the Mergify config (queue_conditions alone are eligibility-only) (#7179)

* fix(ci): migrate Mergify auto-enqueue to merge_protections_settings.auto_merge_conditions (rules-based path is EOL 2026-07-16) (#7216)

* fix(ci): drop Mergify batch settings (batching is a paid-tier feature; free plan queue is serial) (#7220)

* fix(ci): merge queue tolerates the advisory dast-smoke failure (its GH-hosted build hang dequeued every attempt) (#7225)

* test(ci): make the #6634 selfref guard hermetic — main's copy hard-fails every PR (#7341)

main's copy of this test still does git I/O inside a unit test:

    const baseSrc = git(['show', 'origin/main:' + FILE]);

Runners check out a shallow single ref, so origin/main does not resolve and the
test dies with 'fatal: invalid object name origin/main'. Every PR into main
fails Unit Tests (7/8) on it — today that is #7313, #7315, #7316, #7334, #7336
and #7337, six PRs red on a defect none of them introduced. #7313 has no other
red at all.

release/v3.8.49 already carries a fix (2e42b8efc, #7174: try/catch, fetch
origin/main on demand, t.skip() when unreachable), but it only reaches main at
release time — so main stays broken for the whole cycle. Cherry-picking it would
also import a new problem: PR Test Policy classifies t.skip() as a silenced
assertion, which we watched it correctly catch on #7300 today.

This is the hermetic version instead (ported from #7327, which does the same for
the release branch): read the file straight off disk, compare against an empty
base so baseTaut/baseExtTaut are 0 — the strictest possible comparison point —
and call evaluateMasking() directly. No git ref, no fetch, no skip, nothing the
runner's checkout depth can break.

The #6634 regression stays covered: the guard's logic lives in
SELF_TEST_FIXTURE_RE (check-test-masking.mjs:337), not in the test. Proven both
ways on main before committing — neutralise SELF_TEST_FIXTURE_RE to /$^/ and
the test FAILS; restore it and it passes 2/2, with check-test-masking.mjs left
byte-identical.

Co-authored-by: growab <nekron@icloud.com>

* chore(quality): tighten main's coverage baseline to the CI's real numbers (#7347)

main's ratchet had been failing --require-tighten on every PR: 11 metrics
improved but the baseline was never tightened. Same class as the #6634
selfref guard — an infra fix that lands only on the release branch leaves
main red for the whole cycle, and every PR into main pays for it.

Values are the merged-coverage numbers from a run on main itself (a local
run measures ~68% vs CI's ~80%; the baseline's own note warns about that
gap). Only the 11 coverage values change — gitleaks and semgrepFindings
keep main's own state.

No changelog fragment: #7326 carries it on release/v3.8.49, and a second
one here would double the entry at release time.

* fix(antigravity): remove hardcoded 120s SSE collect timeout

The SSE collection in collectStreamToResponse had a hardcoded 120 s
timeout.  Reasoning-heavy models like gemini-3.1-pro-high on large
prompts (>30 KB) regularly exceed 120 s of generation time, causing
the executor to return a synthetic 504 before the model finishes.

Replace the hardcoded value with FETCH_TIMEOUT_MS (default 600 s,
overridable via FETCH_TIMEOUT_MS env var), which is the standard
upstream-request budget across all OmniRoute providers.

Signed-off-by: Minxi Hou <houminxi@gmail.com>

* fix(antigravity): streaming passthrough for non-streaming clients

When a client sends stream: false to the Antigravity executor
(Gemini models), OmniRoute buffered the entire SSE stream before
responding. Long-thinking models exceeded the 120s timeout.

Remove hardcoded SSE_COLLECT_TIMEOUT_MS. Extract shared
createCreditsExtractionTransform with 16KB buffer cap and abort
handling for client disconnect. Add parseSSEToGeminiResponse for
the non-streaming drain path. Fix hasGeminiTerminalFinishReason
to check top-level candidates (no response wrapper). Add signal
null guards for credits retry path. Return 499 on early abort
instead of piping cancelled body.

Also remove duplicate SKILLS_SANDBOX_RUNTIME from .env.example
and clarify .artifacts/ vs _artifacts/ in .gitignore.

Signed-off-by: Minxi Hou <houminxi@gmail.com>

* refactor(antigravity): extract streaming passthrough to module (file-size cap)

#7408 added the non-streaming SSE pass-through (createCreditsExtractionTransform
plus its two call sites: the credits-retry path and the main non-streaming
path) inline in antigravity.ts, growing it to 1806 lines. Combined with two
other authorized PRs touching the same file (#6979 +11, #7290 +30), the
projected total exceeds the frozen file-size gate (1813).

Extract the new streaming-passthrough logic verbatim into
open-sse/executors/antigravity/streamingPassthrough.ts
(createCreditsExtractionTransform + a new buildSsePassthroughResult that
deduplicates the two near-identical call sites), following the existing
sseCollect.ts submodule pattern -- pure, no host state, no fetch/auth.
antigravity.ts keeps a thin wrapper for createCreditsExtractionTransform
(same public signature the existing unit tests import) that injects
updateAntigravityRemainingCredits so the two modules don't import each
other.

No behavior change: same abort handling, same 499-on-early-disconnect,
same 16KB credits sliding-window cap. antigravity.ts: 1806 -> 1693 lines
(under the 1755 pre-PR baseline, with margin). New module: 176 lines
(cap 800).

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* refactor(antigravity): split incremental parser + move new tests to own file (file-size caps)

Two remaining frozen file-size violations from #7408, resolved by
extraction/move with zero behavior or assert changes:

- open-sse/handlers/sseParser.ts (979 > frozen 830): the PR appended
  parseSSEToGeminiResponse (+153, the Gemini buffered-SSE ->
  chat.completion parser). Moved verbatim to
  open-sse/handlers/sseParser/geminiResponse.ts, following the handlers
  submodule pattern (chatCore/, responseSanitizer/). sseParser.ts is now
  byte-identical to its pre-PR content (825 lines; PR delta 0). Importers
  (chatCore/nonStreamingSse.ts, tests) point at the new module.

- tests/unit/executor-antigravity.test.ts (1058 > testFrozen 942): the
  PR's new streaming-passthrough tests moved verbatim (same tests, same
  asserts) to tests/unit/antigravity-streaming-passthrough.test.ts:
  the 3 createCreditsExtractionTransform tests plus the non-streaming
  passthrough drain test ("auto-retries short 429 ... collects SSE for
  non-stream clients"), which the PR rewired onto the new raw-SSE path.
  The frozen file drops to 888 lines (below its pre-PR 941).

New files: geminiResponse.ts 156 lines, passthrough test 202 lines (caps
800). Also fixes the stale sseParser.ts path in collectStreamToResponse's
deprecation note.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* refactor(antigravity): decompose execute + gemini parser below complexity gate

executeOnce() (complexity 127, 436 lines) and parseSSEToGeminiResponse()
(complexity 39, 117 lines) were both over the check-complexity.mjs gate
(complexity>15, max-lines-per-function>80). Decomposed each into small
named helpers, no behavior change:

- geminiResponse.ts: split into pure per-concern functions (markdown
  shortcut, candidate-parts walk, finishReason, usageMetadata, final
  response assembly).
- antigravity.ts: extracted the per-url-index attempt pipeline
  (runAntigravityAttempt, handleAntigravityRateLimit,
  tryResolveRetryFromErrorBody, shouldAutoRetryTransient) and moved the
  request/result-building helpers (send, credits-retry, embed-retry,
  non-streaming/streaming result builders) into a new
  antigravity/executeAttempt.ts submodule, mirroring the existing
  streamingPassthrough.ts/sseCollect.ts pattern. Also fixes the
  antigravity.ts file-size cap (was pushed to 2084 lines > 1813 frozen
  ceiling by the decomposition itself; now 1428).

check-complexity.mjs: 2054 violations (baseline 2058) — net improvement.
execute/executeOnce/parseSSEToGeminiResponse no longer appear with
ruleId complexity or max-lines-per-function.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* Merge branch 'release/v3.8.49' into fix/antigravity-streaming-passthrough

Resolves conflict in open-sse/executors/antigravity.ts between this
branch's streaming-passthrough decomposition and #7290's fallback-chain
decomposition (already merged into release/v3.8.49) — both sides added
imports from the same new antigravity/ submodule files, kept both.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* fix(antigravity): keep buffered JSON contract for non-streaming callers

#3786's Pro-family fallback-chain retry loop (execute()) calls executeOnce()
per candidate and inspects result.response directly, expecting a
synthesized chat.completion JSON body on success. The streaming-passthrough
migration made ALL non-streaming (stream: false) responses a raw SSE
pass-through instead, so a successful retry candidate's response.json()
threw ("data: {...}" is not valid JSON) — breaking the fallback chain
(tests/unit/agy-pro-fallback-chain-3786.test.ts, 3 of 13 red).

Route non-streaming (stream: false) responses back through
collectStreamToResponse (buffered collect-to-JSON), which already uses
FETCH_TIMEOUT_MS with no hardcoded 120s ceiling, so long-thinking models
are not penalized. Passthrough is reserved for actual streaming clients
(stream: true), which was the PR's real target scenario.

Extracted the branch into buildAntigravityAttemptResult() to keep
runAntigravityAttempt under the 80-line ratchet cap.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Signed-off-by: Minxi Hou <houminxi@gmail.com>
Co-authored-by: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com>
Co-authored-by: growab <nekron@icloud.com>
Co-authored-by: HouMinXi <1000+HouMinXi@users.noreply.github.com>
Co-authored-by: HouMinXi <19586012+HouMinXi@users.noreply.github.com>
This commit is contained in:
Bob.Hou
2026-07-18 20:19:20 -04:00
committed by GitHub
parent c6598fdd19
commit 60580ffeb7
11 changed files with 1872 additions and 801 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example
View File

@@ -222,7 +222,7 @@ CONTAINER_HOST=docker
# - orbstack: OrbStack (high-perf Linux VM + docker shim on macOS)
# - podman: Podman (rootless, daemonless)
# - docker: Docker (default fallback)
SKILLS_SANDBOX_RUNTIME=auto
# (defined under SKILLS & SANDBOXING section below)
# ═══════════════════════════════════════════════════════════════════════════════
# 4. SECURITY & AUTHENTICATION

4
.gitignore vendored
View File

@@ -233,7 +233,7 @@ omniroute.md
# mise configuration
mise.toml
_artifacts/
_artifacts/ # release-green artifacts
.claude-flow/
# ESLint file cache (npm run lint --cache / complexity ratchets)
@@ -241,7 +241,7 @@ _artifacts/
.eslintcache-complexity
# CI/local quality artifacts (eslint-results.json, etc.)
# CI/local quality artifacts (eslint-results.json, quality-ratchet.md, etc.)
.artifacts/
# Homologation E2E suite (npm run homolog) — real-environment credentials + report output

5
config/quality/eslint-suppressions.json
View File

@@ -2047,11 +2047,6 @@
"count": 13
}
},
"tests/unit/sse-parser.test.ts": {
"@typescript-eslint/no-explicit-any": {
"count": 15
}
},
"tests/unit/startup-stale-cooldown-recovery.test.ts": {
"@typescript-eslint/no-explicit-any": {
"count": 1

1158
open-sse/executors/antigravity.ts
View File

File diff suppressed because it is too large Load Diff

687
open-sse/executors/antigravity/executeAttempt.ts Normal file
View File

@@ -0,0 +1,687 @@
// Pure-ish per-attempt request/result helpers for the Antigravity executor (#7408
// complexity-gate decomposition): building + sending one upstream request, and
// building the final non-streaming/streaming result. No host state of their own —
// callers inject `provider` and `onCreditsUpdate` so this module doesn't need to
// import the executor's credit-balance cache. Extracted from antigravity.ts
// (file-size cap), mirroring the existing antigravity/streamingPassthrough.ts and
// antigravity/sseCollect.ts submodule pattern.
import { mergeAbortSignals, type ExecutorLog } from "../base.ts";
import { applyFingerprint, isCliCompatEnabled } from "../../config/cliFingerprints.ts";
import { buildAntigravityUpstreamError } from "../antigravityUpstreamError.ts";
import {
HTTP_STATUS,
STREAM_READINESS_TIMEOUT_MS,
ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE,
} from "../../config/constants.ts";
import { injectCreditsField, handleCreditsFailure } from "../../services/antigravityCredits.ts";
import { cloakAntigravityToolPayload } from "../../config/toolCloaking.ts";
import {
applyAntigravityClientProfileHeaders,
removeHeaderCaseInsensitive,
} from "../../services/antigravityClientProfile.ts";
import * as prl from "../../utils/providerRequestLogging.ts";
import {
createCreditsExtractionTransform as createCreditsExtractionTransformImpl,
buildSsePassthroughResult,
type SsePassthroughResult,
} from "./streamingPassthrough.ts";
import type { AntigravityCredentials } from "../antigravity.ts";
const LONG_RETRY_THRESHOLD_MS = 60_000;
const CREDITS_EXHAUSTED_TTL_MS = 5 * 60 * 60 * 1000; // 5 hours
/** Invoked with a fresh GOOGLE_ONE_AI credit balance to persist in the caller's cache. */
export type OnAntigravityCreditsUpdate = (accountId: string, balance: number) => void;
/**
* 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.
* Lives here (not antigravity.ts) so both this module's tryCreditsRetry and
* antigravity.ts's tryResolveRetryFromErrorBody can share it via a single import
* direction (antigravity.ts -> executeAttempt.ts), avoiding a circular import.
*/
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?.();
}
/** True while `accountId`'s Google One AI credits are marked exhausted. @internal */
export 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;
}
/** Mark an account's Google One AI credits as exhausted for CREDITS_EXHAUSTED_TTL_MS. */
export 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);
}
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;
}
/**
* `fetch()` wrapper that aborts if the upstream never returns response headers
* within `timeoutMs` (default STREAM_READINESS_TIMEOUT_MS) — distinct from the
* overall FETCH_TIMEOUT_MS, which bounds the whole request including body streaming.
* Shared by every fetch attempt in executeOnce() (initial, 403-retry, credits-retry).
*/
export async function fetchAntigravityWithReadinessTimeout(
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;
}
}
}
/** ExecutorLog with every method always callable — see toSafeAntigravityLog(). */
export type SafeAntigravityLog = Required<ExecutorLog>;
function noopLogFn(): void {}
/**
* Normalize a possibly-null/undefined ExecutorLog into an object with all four
* methods always callable, so the request/retry helpers below can call
* `l.debug(...)` directly instead of repeating `log?.debug?.(...)` at every call
* site. This isn't just style: the complexity linter (eslint `complexity` rule)
* weighs each `?.` link in a chain as its own branch — a doubly-chained
* `log?.debug?.(...)` costs +2 — so a logging-heavy helper can rack up a large
* complexity score with zero real decision points. Resolving once here keeps
* the actual branch count legible in the functions that matter.
*/
export function toSafeAntigravityLog(log: ExecutorLog | null | undefined): SafeAntigravityLog {
return {
debug: log?.debug ? log.debug.bind(log) : noopLogFn,
info: log?.info ? log.info.bind(log) : noopLogFn,
warn: log?.warn ? log.warn.bind(log) : noopLogFn,
error: log?.error ? log.error.bind(log) : noopLogFn,
};
}
/** Flatten a 429/503 error JSON body (message + `error.details[].reason`) into one string. */
export function buildAntigravity429ErrorMessage(errorJson: unknown): string {
const obj = errorJson as
| { error?: { message?: unknown; details?: unknown }; message?: unknown }
| null
| undefined;
let errorMessage = String(obj?.error?.message || obj?.message || "");
const details = obj?.error?.details;
if (Array.isArray(details)) {
for (const detail of details) {
const reason = (detail as { reason?: unknown } | null)?.reason;
if (reason) errorMessage += ` ${reason}`;
}
}
return errorMessage;
}
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);
}
function getRequestTargetModel(body: Record<string, unknown>): string {
const target = body.model;
return typeof target === "string" && target.length > 0 ? target : "unknown";
}
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;
}
/** Cloak the tool-name payload, then apply credits-first injection, for one attempt. */
export function finalizeAntigravityRequestBody(
transformed: Record<string, unknown>,
useCreditsFirst: boolean,
log: SafeAntigravityLog
): {
transformedBody: Record<string, unknown>;
requestToolNameMap: Map<string, string> | null;
} {
let transformedBody: Record<string, unknown> = transformed;
let requestToolNameMap: Map<string, string> | null = null;
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)");
}
return { transformedBody, requestToolNameMap };
}
/** Debug-only dump of outgoing headers (mask Authorization) and envelope shape. */
function dumpAntigravityRequestDebug(
finalHeaders: Record<string, string>,
transformedBody: Record<string, unknown>,
clientProfile: unknown,
log: SafeAntigravityLog
): void {
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,
})
);
}
/**
* Send one Antigravity request attempt: serialize + apply the client-profile
* fingerprint, debug-dump the outgoing envelope, fetch with a readiness timeout,
* and transparently retry once without `x-goog-user-project` on a 403 (some
* projects reject that header). Returns the (possibly 403-retried) response
* plus the headers actually used for it.
*/
export async function sendAntigravityRequest(
provider: string,
url: string,
model: string,
headers: Record<string, string>,
transformedBody: Record<string, unknown>,
credentials: AntigravityCredentials,
stream: boolean,
signal: AbortSignal | null | undefined,
log: SafeAntigravityLog,
retryAttempt: number
): Promise<{ response: Response; finalHeaders: Record<string, string> }> {
const serializedRequest = serializeAntigravityRequest(provider, headers, transformedBody);
let finalHeaders = serializedRequest.headers;
const clientProfile = applyAntigravityClientProfileHeaders(
finalHeaders,
credentials,
transformedBody
);
log.debug(
"TELEMETRY",
`[Antigravity] Execute - URL: ${url}, Model: ${model}, Target: ${getRequestTargetModel(transformedBody)}, RetryAttempt: ${retryAttempt}`
);
// Dump outgoing headers (mask Authorization) and envelope shape for debugging.
// Gated behind an explicit typeof check (not just calling log.debug() unconditionally)
// so the JSON.stringify work below is skipped entirely when debug logging is off.
if (typeof log.debug === "function") {
dumpAntigravityRequestDebug(finalHeaders, transformedBody, clientProfile, log);
}
await prl.captureCurrentProviderBody(url, finalHeaders, serializedRequest.bodyString, log);
let response = await fetchAntigravityWithReadinessTimeout(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 prl.captureCurrentProviderBody(url, retryHeaders, serializedRequest.bodyString, log);
response = await fetchAntigravityWithReadinessTimeout(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}`
);
}
return { response, finalHeaders };
}
/**
* Retry the SAME url with `enabledCreditTypes: ["GOOGLE_ONE_AI"]` injected, for a
* quota_exhausted 429 that hasn't already tried credits. Returns the result to hand
* back to the caller of execute() on success (or a non-429 status), or null if the
* credits retry also failed/429'd (caller falls through to the normal retry logic).
*/
export async function tryCreditsRetry(
provider: string,
url: string,
headers: Record<string, string>,
transformedBody: Record<string, unknown>,
requestToolNameMap: Map<string, string> | null,
credentials: AntigravityCredentials,
stream: boolean,
signal: AbortSignal | null | undefined,
log: SafeAntigravityLog,
accountId: string,
onCreditsUpdate: OnAntigravityCreditsUpdate
): Promise<SsePassthroughResult | null> {
log.info("AG_CREDITS", "Retrying with Google One AI credits");
const creditsBody = injectCreditsField(transformedBody);
const serializedCreditsRequest = serializeAntigravityRequest(provider, headers, creditsBody);
const finalCreditsHeaders = serializedCreditsRequest.headers;
try {
await prl.captureCurrentProviderBody(
url,
finalCreditsHeaders,
serializedCreditsRequest.bodyString,
log
);
const creditsResp = await fetchAntigravityWithReadinessTimeout(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 && creditsResp.body) {
// Raw SSE pass-through + credits extraction (see
// streamingPassthrough.ts); 499s early if the client
// already disconnected instead of piping a cancelled body.
return buildSsePassthroughResult(
creditsResp.body,
creditsResp,
accountId,
onCreditsUpdate,
url,
finalCreditsHeaders,
attachToolNameMap(creditsBody, requestToolNameMap),
signal
);
}
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);
return null;
} catch (creditsErr) {
handleCreditsFailure(credentials?.accessToken || "");
log.warn("AG_CREDITS", `Credits retry failed: ${creditsErr}`);
return null;
}
}
/**
* If we have a 429 with a long retry time (> LONG_RETRY_THRESHOLD_MS), embed
* `retryAfterMs` in the response body so the caller (combo/account-fallback
* layer) can read it back out. Returns null (fall back to the original
* response handling) when the status/retryMs don't qualify, or on error.
*/
export async function tryEmbedLongRetryAfter(
response: Response,
retryMs: number | null,
url: string,
finalHeaders: Record<string, string>,
transformedBody: Record<string, unknown>,
requestToolNameMap: Map<string, string> | null,
log: ExecutorLog | null | undefined
): Promise<SsePassthroughResult | null> {
if (
response.status !== HTTP_STATUS.RATE_LIMITED ||
!retryMs ||
retryMs <= LONG_RETRY_THRESHOLD_MS
) {
return null;
}
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}`);
return null;
}
}
/** Build the sanitized JSON error result shared by the non-streaming and streaming paths. */
async function buildUpstreamErrorResult(
response: Response,
url: string,
finalHeaders: Record<string, string>,
transformedBody: Record<string, unknown>,
requestToolNameMap: Map<string, string> | null
): Promise<SsePassthroughResult> {
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),
};
}
/**
* For non-streaming clients, return the raw SSE stream with a
* credits-extraction TransformStream. chatCore's non-streaming path
* (readNonStreamingResponseBody + parseNonStreamingSSEPayload with
* Gemini format support) handles draining and conversion to JSON.
* This replaces the previous collectStreamToResponse() approach which
* had an artificial timeout (now the standard FETCH_BODY_TIMEOUT_MS
* of 10 min applies).
*/
async function buildNonStreamingExecuteOnceResult(
response: Response,
url: string,
finalHeaders: Record<string, string>,
transformedBody: Record<string, unknown>,
requestToolNameMap: Map<string, string> | null,
accountId: string,
signal: AbortSignal | null | undefined,
onCreditsUpdate: OnAntigravityCreditsUpdate
): Promise<SsePassthroughResult> {
// #3229: surface a real upstream error instead of masking a 4xx/5xx as an
// empty `chat.completion` envelope.
if (!response.ok) {
return buildUpstreamErrorResult(response, url, finalHeaders, transformedBody, requestToolNameMap);
}
if (response.body) {
// Raw SSE pass-through + credits extraction (see
// streamingPassthrough.ts); 499s early if the client already
// disconnected instead of piping a cancelled body.
return buildSsePassthroughResult(
response.body,
response,
accountId,
onCreditsUpdate,
url,
finalHeaders,
attachToolNameMap(transformedBody, requestToolNameMap),
signal
);
}
// No body -- return as-is
return {
response,
url,
headers: finalHeaders,
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.
*
* #2461: a non-ok upstream response (e.g. 403) must never be piped through the
* streaming pass-through below as if it were an SSE body. Google occasionally
* returns non-UTF8/binary error bodies (observed: gzip-magic-byte payloads) for
* 403s on this endpoint; reading/forwarding those raw bytes corrupts the
* client-visible error message. Mirror the non-streaming branch above and build
* a sanitized JSON error via buildAntigravityUpstreamError (hard rule #12)
* instead of streaming unknown bytes straight through.
*/
async function buildStreamingExecuteOnceResult(
response: Response,
url: string,
finalHeaders: Record<string, string>,
transformedBody: Record<string, unknown>,
requestToolNameMap: Map<string, string> | null,
accountId: string,
signal: AbortSignal | null | undefined,
onCreditsUpdate: OnAntigravityCreditsUpdate
): Promise<SsePassthroughResult> {
if (!response.ok) {
return buildUpstreamErrorResult(response, url, finalHeaders, transformedBody, requestToolNameMap);
}
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 });
}
}
const passThrough = createCreditsExtractionTransformImpl(
accountId,
onCreditsUpdate,
16 * 1024 // 16KB sliding-window cap to prevent OOM
);
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),
};
}
/** Dispatch to the non-streaming or streaming final-result builder. */
export async function buildFinalAntigravityResult(
stream: boolean,
response: Response,
url: string,
finalHeaders: Record<string, string>,
transformedBody: Record<string, unknown>,
requestToolNameMap: Map<string, string> | null,
accountId: string,
signal: AbortSignal | null | undefined,
onCreditsUpdate: OnAntigravityCreditsUpdate
): Promise<SsePassthroughResult> {
if (!stream) {
return buildNonStreamingExecuteOnceResult(
response,
url,
finalHeaders,
transformedBody,
requestToolNameMap,
accountId,
signal,
onCreditsUpdate
);
}
return buildStreamingExecuteOnceResult(
response,
url,
finalHeaders,
transformedBody,
requestToolNameMap,
accountId,
signal,
onCreditsUpdate
);
}

176
open-sse/executors/antigravity/streamingPassthrough.ts Normal file
View File

@@ -0,0 +1,176 @@
// Pure streaming pass-through helpers for the Antigravity executor (#7408):
// tap an upstream Gemini SSE Response through a credits-extraction
// TransformStream instead of buffering the whole body in the executor, so
// long-thinking models aren't killed by an artificial collection timeout.
// Extracted from antigravity.ts (no host state, no fetch/auth) -- the
// credit-balance cache itself stays in antigravity.ts; callers inject the
// update function below so the two modules don't import each other.
/** Shape of one entry in a Gemini `remainingCredits` SSE payload array. */
export type AntigravityCreditEntry = {
creditType?: string;
creditAmount?: string;
};
function asCreditRecord(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
/**
* Create a pass-through TransformStream that extracts `remainingCredits`
* from SSE data without consuming the stream. The downstream client
* receives the unmodified bytes.
*
* @param accountId Provider account ID for credit-balance persistence.
* @param onCreditsUpdate Invoked with the parsed GOOGLE_ONE_AI balance.
* Injected by the caller (antigravity.ts's
* updateAntigravityRemainingCredits) to avoid this module
* importing back the executor's credit-balance cache.
* @param bufferSize Optional sliding-window buffer cap in bytes.
* Pass 0 or omit for unlimited (non-streaming callers
* where the full body is already buffered upstream).
* The streaming path uses 16384 (16 KB) to prevent OOM
* on long-lived SSE connections. Credit-balance data
* appears near the end of the SSE stream (after
* content), so the sliding window captures it even at
* 16 KB -- only truly massive responses (>16 KB of
* consecutive non-newline content) would lose credits.
*/
export function createCreditsExtractionTransform(
accountId: string,
onCreditsUpdate: (accountId: string, balance: number) => void,
bufferSize = 0
): TransformStream<Uint8Array, Uint8Array> {
let buffer = "";
const decoder = new TextDecoder();
return new TransformStream(
{
transform(chunk, controller) {
controller.enqueue(chunk);
try {
buffer += decoder.decode(chunk, { stream: true });
// Sliding-window cap: truncate after the last complete newline
// in the discard region so SSE lines are never split mid-payload.
if (bufferSize > 0 && buffer.length > bufferSize) {
const lastNewline = buffer.lastIndexOf("\n", buffer.length - bufferSize);
if (lastNewline !== -1) {
buffer = buffer.slice(lastNewline + 1);
} else {
// No newline in the discard region -- incomplete line, discard entirely.
buffer = "";
}
}
} catch {
/* decoding best-effort */
}
},
flush() {
try {
buffer += decoder.decode();
} catch {
/* decoding best-effort */
}
try {
const lines = buffer.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 = asCreditRecord(c);
return credit?.creditType === "GOOGLE_ONE_AI";
}) as AntigravityCreditEntry | undefined;
if (googleCredit) {
const balance = parseInt(String(googleCredit.creditAmount ?? ""), 10);
if (!isNaN(balance)) onCreditsUpdate(accountId, balance);
}
}
} catch {
/* skip malformed lines */
}
}
} catch {
/* credits extraction is best-effort */
}
buffer = "";
},
},
{ highWaterMark: 16384 },
{ highWaterMark: 16384 }
);
}
/** Result shape returned to callers of AntigravityExecutor.execute(). */
export type SsePassthroughResult = {
response: Response;
url: string;
headers: Record<string, string>;
transformedBody: unknown;
};
/** Cancel `body` when `signal` aborts, releasing the upstream connection. */
function cancelBodyOnAbort(body: ReadableStream<Uint8Array>, signal: AbortSignal): void {
signal.addEventListener(
"abort",
() => {
body.cancel().catch(() => {});
},
{ once: true }
);
}
/**
* Build the non-streaming pass-through result: tap `body` through
* createCreditsExtractionTransform and wrap it in a same-status Response so
* chatCore's non-streaming path (readNonStreamingResponseBody +
* parseNonStreamingSSEPayload) can drain and parse the Gemini SSE without
* this executor buffering the whole stream itself.
*
* If the client already disconnected (`signal.aborted`), cancels the
* upstream body immediately and returns a bare 499 instead of piping a
* cancelled body through.
*/
export function buildSsePassthroughResult(
body: ReadableStream<Uint8Array>,
upstream: { status: number; statusText: string; headers: Headers },
accountId: string,
onCreditsUpdate: (accountId: string, balance: number) => void,
url: string,
outHeaders: Record<string, string>,
transformedBody: unknown,
signal: AbortSignal | null | undefined
): SsePassthroughResult {
// Client already disconnected — skip pipe
if (signal?.aborted) {
body.cancel().catch(() => {});
return {
response: new Response(null, { status: 499 }),
url,
headers: outHeaders,
transformedBody: null,
};
}
// Cancel upstream body on client disconnect
if (signal) cancelBodyOnAbort(body, signal);
const tapped = body.pipeThrough(
createCreditsExtractionTransform(accountId, onCreditsUpdate, 16 * 1024)
);
return {
response: new Response(tapped, {
status: upstream.status,
statusText: upstream.statusText,
headers: upstream.headers,
}),
url,
headers: outHeaders,
transformedBody,
};
}

43
open-sse/handlers/chatCore/nonStreamingSse.ts
View File

@@ -4,6 +4,7 @@ import {
parseSSEToClaudeResponse,
parseSSEToOpenAIResponse,
} from "../sseParser.ts";
import { parseSSEToGeminiResponse } from "../sseParser/geminiResponse.ts";
import { getHeaderValueCaseInsensitive } from "./headers.ts";
export function parseNonStreamingSSEPayload(
@@ -20,6 +21,7 @@ export function parseNonStreamingSSEPayload(
};
queueFormat(preferredFormat);
queueFormat(FORMATS.GEMINI);
queueFormat(FORMATS.OPENAI_RESPONSES);
queueFormat(FORMATS.CLAUDE);
queueFormat(FORMATS.OPENAI);
@@ -30,7 +32,9 @@ export function parseNonStreamingSSEPayload(
? parseSSEToResponsesOutput(rawBody, fallbackModel)
: format === FORMATS.CLAUDE
? parseSSEToClaudeResponse(rawBody, fallbackModel)
: parseSSEToOpenAIResponse(rawBody, fallbackModel);
: format === FORMATS.GEMINI || format === FORMATS.ANTIGRAVITY
? parseSSEToGeminiResponse(rawBody, fallbackModel)
: parseSSEToOpenAIResponse(rawBody, fallbackModel);
if (parsed && typeof parsed === "object") {
return {
body: parsed as Record<string, unknown>,
@@ -107,6 +111,21 @@ function hasClaudeTerminalMessageDelta(parsed: unknown, eventType: string): bool
return typeof stopReason === "string" ? stopReason.length > 0 : stopReason != null;
}
// Non-empty finishReason is terminal. Gemini SSE payloads from
// streamGenerateContent have candidates at the top level (no
// "response" wrapper). Any non-empty string signals stream end.
function hasGeminiTerminalFinishReason(parsed: unknown): boolean {
if (!parsed || typeof parsed !== "object") return false;
// Top-level candidates (streamGenerateContent?alt=sse)
const obj = parsed as Record<string, unknown>;
const candidates = obj.candidates as unknown[] | undefined;
if (!Array.isArray(candidates) || candidates.length === 0) return false;
const candidate = candidates[0] as Record<string, unknown> | undefined;
if (!candidate || typeof candidate !== "object") return false;
const finishReason = candidate.finishReason;
return typeof finishReason === "string" && finishReason.length > 0;
}
function processNonStreamingSseTerminalLine(
state: NonStreamingSseTerminalState,
rawLine: string
@@ -130,12 +149,22 @@ function processNonStreamingSseTerminalLine(
// Hot-path optimization: the terminal SSE events we look for (message_stop,
// response.completed, …) all carry a top-level "type" field, OR are signalled by a
// preceding `event:` line (Claude). OpenAI chat.completion chunks carry neither and
// terminate with `[DONE]` (handled above), so parsing every one of them here is pure
// waste that compounds into the CPU-runaway on large buffered responses. Skip the
// JSON.parse unless the line could actually be a typed terminal.
// preceding `event:` line (Claude). Gemini signals completion via
// "finishReason" inside response.candidates[0]. OpenAI chat.completion chunks
// carry none of these and terminate with `[DONE]` (handled above), so parsing
// every one of them here is pure waste that compounds into the CPU-runaway on
// large buffered responses. Skip the JSON.parse unless the line could actually
// be a typed terminal.
if (
!data.includes('"type"') &&
// NOTE: "finishReason" is a superset match -- it triggers JSON.parse on
// every Gemini chunk that happens to contain the string (e.g. partial
// candidate payloads), not just the terminal one. This is intentional:
// the extra parses are cheap compared to the CPU-runaway we'd get from
// parsing ALL chunks unconditionally on large buffered responses, and
// the superset is safe (false positives just parse a non-terminal chunk
// and fall through to `return false`).
!data.includes('"finishReason"') &&
!(state.currentEvent === "message_delta" && data.includes("stop_reason"))
) {
return isNonStreamingSseTerminalType(state.currentEvent);
@@ -148,7 +177,9 @@ function processNonStreamingSseTerminalLine(
? parsed.type
: state.currentEvent;
return (
isNonStreamingSseTerminalType(eventType) || hasClaudeTerminalMessageDelta(parsed, eventType)
isNonStreamingSseTerminalType(eventType) ||
hasClaudeTerminalMessageDelta(parsed, eventType) ||
hasGeminiTerminalFinishReason(parsed)
);
} catch {
// Keep reading malformed data so the parser can report a useful upstream error.

214
open-sse/handlers/sseParser/geminiResponse.ts Normal file
View File

@@ -0,0 +1,214 @@
// Gemini/Antigravity buffered-SSE -> chat.completion conversion (#7408).
// Extracted verbatim from sseParser.ts (file-size cap): pure parsing, no host
// state, following the handlers submodule pattern (chatCore/, responseSanitizer/).
import { normalizeOpenAICompatibleFinishReasonString } from "../../utils/finishReason.ts";
type AccumulatedToolCall = {
id: string;
index: number;
type: "function";
function: { name: string; arguments: string };
};
/** Mutable accumulator threaded through one SSE payload's worth of parsing. */
type GeminiSSEAccumulator = {
textContent: string;
finishReason: string;
usage: Record<string, unknown> | null;
sawContent: boolean;
toolCalls: AccumulatedToolCall[];
};
function stripZeroWidth(value: unknown): unknown {
if (typeof value === "string") return value.replace(/[\u200B-\u200D\uFEFF]/g, "");
return value;
}
/**
* Detect the `[Tool call: name]\nArguments: {...}` textual convention some
* Gemini/Antigravity models emit instead of a native functionCall part.
*/
function tryParseTextualToolCall(text: string): { name: string; args: unknown } | null {
const normalized = text.replace(/[\u200B-\u200D\uFEFF]/g, "");
const match = normalized.match(
/^[\s\S]*?\[Tool call:\s*([^\]\n]+)\]\s*\nArguments:\s*([\s\S]+?)\s*$/
);
if (!match) return null;
const name = match[1]?.trim();
const rawArgs = match[2]?.trim();
if (!name || !rawArgs) return null;
try {
return { name, args: stripZeroWidth(JSON.parse(rawArgs)) };
} catch {
return null;
}
}
/** Extract the markdown shortcut some Gemini variants send (top-level or nested). */
function extractGeminiMarkdownShortcut(parsed: Record<string, unknown>): string | null {
if (typeof parsed.markdown === "string") return parsed.markdown;
const response = parsed.response as Record<string, unknown> | undefined;
return typeof response?.markdown === "string" ? response.markdown : null;
}
/** Append one candidate content part (text or textual tool call) onto the accumulator. */
function applyCandidatePart(part: Record<string, unknown>, acc: GeminiSSEAccumulator): void {
if (typeof part.text !== "string" || part.thought || part.thoughtSignature) return;
const textualToolCall = tryParseTextualToolCall(part.text);
if (textualToolCall) {
acc.toolCalls.push({
id: `${textualToolCall.name}-${Date.now()}-${acc.toolCalls.length}`,
index: acc.toolCalls.length,
type: "function",
function: {
name: textualToolCall.name,
arguments: JSON.stringify(textualToolCall.args || {}),
},
});
} else {
acc.textContent += part.text;
}
acc.sawContent = true;
}
/** Walk the first candidate's content parts, if present, mutating the accumulator. */
function applyCandidateContentParts(
candidate: Record<string, unknown> | undefined,
acc: GeminiSSEAccumulator
): void {
const content = candidate?.content as Record<string, unknown> | undefined;
const parts = content?.parts;
if (!Array.isArray(parts)) return;
for (const part of parts) {
applyCandidatePart(part as Record<string, unknown>, acc);
}
}
/** Normalize and apply the candidate's finishReason, if present. */
function applyFinishReason(
candidate: Record<string, unknown> | undefined,
acc: GeminiSSEAccumulator
): void {
if (!candidate?.finishReason) return;
acc.finishReason = normalizeOpenAICompatibleFinishReasonString(
String(candidate.finishReason).toLowerCase()
);
}
/** Extract usageMetadata into the OpenAI-shaped usage object, if present. */
function applyUsageMetadata(parsed: Record<string, unknown>, acc: GeminiSSEAccumulator): void {
const response = parsed.response as Record<string, unknown> | undefined;
const um = response?.usageMetadata as Record<string, unknown> | undefined;
if (!um) return;
acc.usage = {
prompt_tokens: um.promptTokenCount || 0,
completion_tokens: um.candidatesTokenCount || 0,
total_tokens: um.totalTokenCount || 0,
};
}
/** Parse one `data:` line's JSON payload and fold it into the accumulator (best-effort). */
function applyGeminiSSEDataLine(payload: string, acc: GeminiSSEAccumulator): void {
try {
const parsed = JSON.parse(payload) as Record<string, unknown>;
const markdown = extractGeminiMarkdownShortcut(parsed);
if (markdown) {
acc.textContent += markdown;
acc.sawContent = true;
}
const response = parsed.response as Record<string, unknown> | undefined;
const candidates = response?.candidates;
const candidate = Array.isArray(candidates)
? (candidates[0] as Record<string, unknown> | undefined)
: undefined;
applyCandidateContentParts(candidate, acc);
applyFinishReason(candidate, acc);
applyUsageMetadata(parsed, acc);
} catch {
// Ignore malformed lines
}
}
/** Assemble the final non-streaming chat.completion payload from the accumulator. */
function buildChatCompletionFromAccumulator(
acc: GeminiSSEAccumulator,
fallbackModel: string
): Record<string, unknown> {
const message: Record<string, unknown> = {
role: "assistant",
content: acc.textContent || null,
};
let finishReason = acc.finishReason;
if (acc.toolCalls.length > 0) {
message.tool_calls = acc.toolCalls;
finishReason = "tool_calls";
}
const result: Record<string, unknown> = {
id: `chatcmpl-${Date.now()}`,
object: "chat.completion",
created: Math.floor(Date.now() / 1000),
model: fallbackModel || "unknown",
choices: [
{
index: 0,
message,
finish_reason: finishReason,
},
],
};
if (acc.usage) {
result.usage = acc.usage;
}
return result;
}
/**
* Convert Gemini/Antigravity SSE chunks into a single non-streaming OpenAI
* chat.completion JSON response. Gemini SSE carries payloads like:
*
* data: {"markdown":"...chunk..."}
* data: {"response":{"candidates":[{"content":{"parts":[{"text":"..."}]},"finishReason":"STOP"}],"usageMetadata":{...}}}
* data: {"remainingCredits":[...]}
*
* Reuses the same parsing logic as processAntigravitySSEPayload() in sseCollect.ts
* so that format conversion is functionally equivalent to the previous
* collectStreamToResponse() approach. Intentional differences:
* - remainingCredits is NOT embedded into the result (handled separately
* by the credits-extraction TransformStream in antigravity.ts).
* - The synthetic `id` uses `chatcmpl-${Date.now()}` (no UUID suffix)
* because this path runs once per response, not per chunk.
*/
export function parseSSEToGeminiResponse(
rawSSE: string,
fallbackModel: string
): Record<string, unknown> | null {
const lines = String(rawSSE || "").split("\n");
const acc: GeminiSSEAccumulator = {
textContent: "",
finishReason: "stop",
usage: null,
sawContent: false,
toolCalls: [],
};
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;
applyGeminiSSEDataLine(payload, acc);
}
if (!acc.sawContent && acc.toolCalls.length === 0) return null;
return buildChatCompletionFromAccumulator(acc, fallbackModel);
}

202
tests/unit/antigravity-streaming-passthrough.test.ts Normal file
View File

@@ -0,0 +1,202 @@
// Antigravity streaming-passthrough behavior (#7408): the non-streaming drain
// path (raw SSE + chatCore-side parse) and the credits-extraction pass-through
// TransformStream. Moved verbatim from tests/unit/executor-antigravity.test.ts
// (frozen file-size cap) — same tests, same asserts.
import { test } from "node:test";
import assert from "node:assert/strict";
import {
AntigravityExecutor,
createCreditsExtractionTransform,
} from "../../open-sse/executors/antigravity.ts";
import { parseSSEToGeminiResponse } from "../../open-sse/handlers/sseParser/geminiResponse.ts";
import {
clearAntigravityVersionCache,
seedAntigravityVersionCache,
} from "../../open-sse/services/antigravityVersion.ts";
type ChatCompletionPayload = {
object?: string;
choices: Array<{
message: { content: string };
finish_reason: string;
}>;
usage?: {
prompt_tokens: number;
completion_tokens: number;
total_tokens: number;
};
};
test.afterEach(() => {
clearAntigravityVersionCache();
});
test("AntigravityExecutor.execute auto-retries short 429 responses and collects SSE for non-stream clients", async () => {
const executor = new AntigravityExecutor();
const originalFetch = globalThis.fetch;
const originalSetTimeout = globalThis.setTimeout;
const calls = [];
seedAntigravityVersionCache("2026.04.17-test");
globalThis.fetch = async (url) => {
calls.push(String(url));
if (calls.length === 1) {
return new Response(JSON.stringify({ error: { message: "rate limited" } }), {
status: 429,
headers: { "Content-Type": "application/json" },
});
}
return new Response(
[
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"Hello "}]},"finishReason":"STOP"}]}}\n\n',
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"again"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":2,"candidatesTokenCount":3,"totalTokenCount":5}}}\n\n',
].join(""),
{
status: 200,
headers: { "Content-Type": "text/event-stream" },
}
);
};
globalThis.setTimeout = ((callback) => {
(callback as () => void)();
return 0;
}) as typeof setTimeout;
try {
const result = await executor.execute({
model: "antigravity/gemini-2.5-flash",
body: { request: { contents: [] } },
stream: false,
credentials: { accessToken: "token", projectId: "project-1" },
log: { debug() {}, warn() {} },
});
// Non-streaming now returns raw SSE; parse it the way chatCore would.
const rawSSE = await result.response.text();
const parsed = parseSSEToGeminiResponse(rawSSE, "antigravity/gemini-2.5-flash");
assert.ok(parsed, "parseSSEToGeminiResponse should parse the SSE");
const payload = parsed as ChatCompletionPayload;
assert.equal(calls.length, 2);
assert.equal(result.response.status, 200);
assert.equal(payload.choices[0].message.content, "Hello again");
assert.deepEqual(payload.usage, {
prompt_tokens: 2,
completion_tokens: 3,
total_tokens: 5,
});
} finally {
globalThis.fetch = originalFetch;
globalThis.setTimeout = originalSetTimeout;
}
});
// ---------------------------------------------------------------------------
// createCreditsExtractionTransform -- credits extraction with buffer cap
// ---------------------------------------------------------------------------
test("createCreditsExtractionTransform extracts remainingCredits from SSE data", async () => {
const encoder = new TextEncoder();
const sseData = [
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"hello"}]},"finishReason":"STOP"}]}}\n\n',
'data: {"remainingCredits":[{"creditType":"GOOGLE_ONE_AI","creditAmount":"42"}]}\n\n',
].join("");
const transform = createCreditsExtractionTransform("test-account");
const readable = new ReadableStream({
start(controller) {
controller.enqueue(encoder.encode(sseData));
controller.close();
},
});
// Consume the stream through the transform
const output = readable.pipeThrough(transform);
const reader = output.getReader();
const chunks: Uint8Array[] = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value);
}
// Data must pass through unmodified
const collected = new TextDecoder().decode(
new Uint8Array(
chunks.reduce((acc, c) => acc + c.length, 0) > 0 ? Buffer.concat(chunks) : new Uint8Array(0)
)
);
assert.ok(collected.includes("hello"));
assert.ok(collected.includes("remainingCredits"));
});
test("createCreditsExtractionTransform with buffer cap truncates large buffers", async () => {
const encoder = new TextEncoder();
// Build a payload larger than 1KB
const largeText = "x".repeat(2000);
const ssePayload = JSON.stringify({
response: {
candidates: [{ content: { parts: [{ text: largeText }] } }],
},
});
const sseLine = `data: ${ssePayload}\n\n`;
// Append a credits line at the end
const creditsLine =
'data: {"remainingCredits":[{"creditType":"GOOGLE_ONE_AI","creditAmount":"99"}]}\n\n';
const fullData = sseLine + creditsLine;
// Use a 512-byte buffer cap -- the large text line should be discarded
const transform = createCreditsExtractionTransform("test-account", 512);
const readable = new ReadableStream({
start(controller) {
// Send in small chunks to exercise the sliding-window logic
const encoded = encoder.encode(fullData);
const chunkSize = 256;
for (let i = 0; i < encoded.length; i += chunkSize) {
controller.enqueue(encoded.slice(i, i + chunkSize));
}
controller.close();
},
});
const output = readable.pipeThrough(transform);
const reader = output.getReader();
while (true) {
const { done } = await reader.read();
if (done) break;
}
// The transform should not throw -- buffer cap just limits what the
// flush handler can see. If the credits line was within the last 512
// bytes it will be found; otherwise it's a graceful no-op.
// Either way, no crash or OOM.
assert.ok(true);
});
test("createCreditsExtractionTransform handles malformed SSE gracefully", async () => {
const encoder = new TextEncoder();
const badData = "not valid sse\ndata: {broken json\n\ndata: [DONE]\n\n";
const transform = createCreditsExtractionTransform("test-account");
const readable = new ReadableStream({
start(controller) {
controller.enqueue(encoder.encode(badData));
controller.close();
},
});
const output = readable.pipeThrough(transform);
const reader = output.getReader();
const chunks: Uint8Array[] = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value);
}
// Data passes through unmodified, no crash on malformed input
const collected = new TextDecoder().decode(Buffer.concat(chunks));
assert.ok(collected.includes("not valid sse"));
});

59
tests/unit/executor-antigravity.test.ts
View File

@@ -631,62 +631,9 @@ test("AntigravityExecutor.refreshCredentials refreshes Google OAuth tokens", asy
}
});
test("AntigravityExecutor.execute auto-retries short 429 responses and collects SSE for non-stream clients", async () => {
const executor = new AntigravityExecutor();
const originalFetch = globalThis.fetch;
const originalSetTimeout = globalThis.setTimeout;
const calls = [];
seedAntigravityVersionCache("2026.04.17-test");
globalThis.fetch = async (url) => {
calls.push(String(url));
if (calls.length === 1) {
return new Response(JSON.stringify({ error: { message: "rate limited" } }), {
status: 429,
headers: { "Content-Type": "application/json" },
});
}
return new Response(
[
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"Hello "}]},"finishReason":"STOP"}]}}\n\n',
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"again"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":2,"candidatesTokenCount":3,"totalTokenCount":5}}}\n\n',
].join(""),
{
status: 200,
headers: { "Content-Type": "text/event-stream" },
}
);
};
globalThis.setTimeout = ((callback) => {
(callback as () => void)();
return 0;
}) as typeof setTimeout;
try {
const result = await executor.execute({
model: "antigravity/gemini-2.5-flash",
body: { request: { contents: [] } },
stream: false,
credentials: { accessToken: "token", projectId: "project-1" },
log: { debug() {}, warn() {} },
});
const payload = (await result.response.json()) as ChatCompletionPayload;
assert.equal(calls.length, 2);
assert.equal(result.response.status, 200);
assert.equal(payload.choices[0].message.content, "Hello again");
assert.deepEqual(payload.usage, {
prompt_tokens: 2,
completion_tokens: 3,
total_tokens: 5,
});
} finally {
globalThis.fetch = originalFetch;
globalThis.setTimeout = originalSetTimeout;
}
});
// The non-streaming passthrough drain test ("auto-retries short 429 responses and
// collects SSE for non-stream clients") lives in
// tests/unit/antigravity-streaming-passthrough.test.ts with the other passthrough tests.
test("AntigravityExecutor.execute embeds retryAfterMs when the upstream asks for a long wait", async () => {
const executor = new AntigravityExecutor();

123
tests/unit/sse-parser.test.ts
View File

@@ -3,6 +3,8 @@ import assert from "node:assert/strict";
const { parseSSEToOpenAIResponse, parseSSEToClaudeResponse, parseSSEToResponsesOutput } =
await import("../../open-sse/handlers/sseParser.ts");
const { parseSSEToGeminiResponse } =
await import("../../open-sse/handlers/sseParser/geminiResponse.ts");
test("parseSSEToOpenAIResponse parses a single SSE event with a done marker", () => {
const rawSSE = [
@@ -103,12 +105,12 @@ test("parseSSEToClaudeResponse parses text, thinking, tool_use, and usage events
assert.equal(parsed.id, "msg_1");
assert.equal(parsed.model, "claude-3-5-sonnet");
assert.equal((parsed.content[0] as any).type, "thinking");
assert.equal((parsed as any).content[0].thinking, "step 1");
assert.equal((parsed as any).content[0].signature, "sig-1");
(assert as any).equal((parsed.content[1] as any).text, "Hello");
assert.equal((parsed.content[2] as any).type, "tool_use");
(assert as any).deepEqual((parsed.content[2] as any).input, { q: "docs" });
assert.equal((parsed.content[0] as { type: string }).type, "thinking");
assert.equal((parsed.content[0] as { thinking: string }).thinking, "step 1");
assert.equal((parsed.content[0] as { signature: string }).signature, "sig-1");
assert.equal((parsed.content[1] as { text: string }).text, "Hello");
assert.equal((parsed.content[2] as { type: string }).type, "tool_use");
assert.deepEqual((parsed.content[2] as { input: unknown }).input, { q: "docs" });
assert.equal(parsed.stop_reason, "tool_use");
assert.equal(parsed.stop_sequence, "END");
assert.deepEqual(parsed.usage, { input_tokens: 10, output_tokens: 4 });
@@ -130,7 +132,7 @@ test("parseSSEToClaudeResponse tolerates event-only types and missing blank sepa
assert.equal(parsed.id, "msg_event_fallback");
assert.equal(parsed.model, "claude-sonnet-4-6");
assert.equal((parsed.content[0] as any).text, "event fallback ok");
assert.equal((parsed.content[0] as { text: string }).text, "event fallback ok");
assert.deepEqual(parsed.usage, { input_tokens: 3, output_tokens: 2 });
});
@@ -152,9 +154,9 @@ test("parseSSEToClaudeResponse merges signature_delta into an existing thinking
const parsed = parseSSEToClaudeResponse(rawSSE, "fallback-model");
assert.equal((parsed.content[0] as any).type, "thinking");
assert.equal((parsed.content[0] as any).thinking, "first second");
assert.equal((parsed.content[0] as any).signature, "sig-1");
assert.equal((parsed.content[0] as { type: string }).type, "thinking");
assert.equal((parsed.content[0] as { thinking: string }).thinking, "first second");
assert.equal((parsed.content[0] as { signature: string }).signature, "sig-1");
});
test("parseSSEToClaudeResponse preserves signature_delta when it arrives before thinking_delta", () => {
@@ -172,9 +174,9 @@ test("parseSSEToClaudeResponse preserves signature_delta when it arrives before
const parsed = parseSSEToClaudeResponse(rawSSE, "fallback-model");
assert.equal((parsed.content[0] as any).type, "thinking");
assert.equal((parsed.content[0] as any).thinking, "later thinking");
assert.equal((parsed.content[0] as any).signature, "sig-before");
assert.equal((parsed.content[0] as { type: string }).type, "thinking");
assert.equal((parsed.content[0] as { thinking: string }).thinking, "later thinking");
assert.equal((parsed.content[0] as { signature: string }).signature, "sig-before");
});
test("parseSSEToClaudeResponse ignores malformed payloads and returns null when nothing valid remains", () => {
@@ -334,3 +336,98 @@ test("parseSSEToOpenAIResponse deduplicates repeated tool call snapshots", () =>
assert.equal(toolCall.function.arguments, args);
assert.equal(JSON.parse(toolCall.function.arguments).command, "find /tmp -name test.txt");
});
// ---------------------------------------------------------------------------
// parseSSEToGeminiResponse
// ---------------------------------------------------------------------------
test("parseSSEToGeminiResponse extracts text content from candidate parts", () => {
const rawSSE = [
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"Hello "}]}}]}}',
'data: {"response":{"candidates":[{"content":{"parts":[{"text":"world"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":5,"candidatesTokenCount":3,"totalTokenCount":8}}}',
].join("\n");
const parsed = parseSSEToGeminiResponse(rawSSE, "gemini-2.5-flash");
assert.ok(parsed);
assert.equal(parsed.object, "chat.completion");
assert.equal(parsed.choices[0].message.content, "Hello world");
assert.equal(parsed.choices[0].finish_reason, "stop");
assert.deepEqual(parsed.usage, {
prompt_tokens: 5,
completion_tokens: 3,
total_tokens: 8,
});
});
test("parseSSEToGeminiResponse handles markdown shortcut format", () => {
const rawSSE = [
'data: {"markdown":"Hello "}',
'data: {"markdown":"world"}',
'data: {"response":{"candidates":[{"finishReason":"STOP"}]}}',
].join("\n");
const parsed = parseSSEToGeminiResponse(rawSSE, "gemini-2.5-flash");
assert.ok(parsed);
assert.equal(parsed.choices[0].message.content, "Hello world");
assert.equal(parsed.choices[0].finish_reason, "stop");
});
test("parseSSEToGeminiResponse extracts tool calls from textual format", () => {
const rawSSE = [
`data: ${JSON.stringify({
response: {
candidates: [
{
content: {
parts: [
{
text: '[Tool call: search_files]\nArguments: {"path":"/tmp"}',
},
],
},
finishReason: "STOP",
},
],
},
})}`,
].join("\n");
const parsed = parseSSEToGeminiResponse(rawSSE, "gemini-3.5-flash-low");
assert.ok(parsed);
assert.equal(parsed.choices[0].finish_reason, "tool_calls");
const toolCalls = parsed.choices[0].message.tool_calls;
assert.equal(toolCalls.length, 1);
assert.equal(toolCalls[0].function.name, "search_files");
assert.deepEqual(JSON.parse(toolCalls[0].function.arguments), { path: "/tmp" });
});
test("parseSSEToGeminiResponse returns null for empty or non-content SSE", () => {
assert.equal(parseSSEToGeminiResponse("", "model"), null);
assert.equal(parseSSEToGeminiResponse("data: [DONE]\n", "model"), null);
assert.equal(parseSSEToGeminiResponse("event: ping\n", "model"), null);
});
test("parseSSEToGeminiResponse ignores thought/thoughtSignature parts", () => {
const rawSSE = [
`data: ${JSON.stringify({
response: {
candidates: [
{
content: {
parts: [{ text: "internal reasoning", thought: true }, { text: "visible answer" }],
},
finishReason: "STOP",
},
],
},
})}`,
].join("\n");
const parsed = parseSSEToGeminiResponse(rawSSE, "model");
assert.ok(parsed);
assert.equal(parsed.choices[0].message.content, "visible answer");
});