Skip to content

refactor(executors): extract pure SSE-collect parsing from antigravity (1812→1671) #5962

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
diegosouzapw merged 1 commit into from
Jul 2, 2026
Merged

refactor(executors): extract pure SSE-collect parsing from antigravity (1812→1671) #5962

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 7 additions & 143 deletions open-sse/executors/antigravity.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,13 @@ import {
} from "../services/cloudCodeThinking.ts";
import { buildGeminiTools } from "../translator/helpers/geminiToolsSanitizer.ts";
import { DEFAULT_SAFETY_SETTINGS } from "../translator/helpers/geminiHelper.ts";
import { normalizeOpenAICompatibleFinishReasonString } from "../utils/finishReason.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,
Expand Down Expand Up @@ -155,70 +161,6 @@ function serializeAntigravityRequest(
return applyFingerprint(provider, { ...headers }, serializedBody);
}

type AntigravityCollectedStream = {
textContent: string;
finishReason: string;
toolCalls: Array<{
id: string;
index: number;
type: "function";
function: { name: string; arguments: string };
}>;
usage: Record<string, unknown> | null;
remainingCredits: Array<{ creditType: string; creditAmount: string }> | null;
};

function stripZeroWidth(value: unknown): unknown {
if (typeof value === "string") {
return value.replace(/[\u200B-\u200D\uFEFF]/g, "");
}
if (Array.isArray(value)) {
return value.map((item) => stripZeroWidth(item));
}
if (value && typeof value === "object") {
return Object.fromEntries(
Object.entries(value as Record<string, unknown>).map(([key, item]) => [
key,
stripZeroWidth(item),
])
);
}
return value;
}

function parseAntigravityTextualToolCall(text: unknown): { name: string; args: unknown } | null {
if (typeof text !== "string") return 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;
}
}

function addAntigravityTextualToolCall(
collected: AntigravityCollectedStream,
parsed: { name: string; args: unknown }
): void {
collected.toolCalls.push({
id: `${parsed.name}-${Date.now()}-${collected.toolCalls.length}`,
index: collected.toolCalls.length,
type: "function",
function: {
name: parsed.name,
arguments: JSON.stringify(parsed.args || {}),
},
});
collected.finishReason = "tool_calls";
}

type AntigravityRequestEnvelope = Record<string, unknown> & {
project: string;
model?: string;
Expand Down Expand Up @@ -372,84 +314,6 @@ export function markConnectionQuotaExhausted(connectionId: string, retryAfterMs:
* Accumulate one Antigravity SSE `data:` payload into `collected`. Exported for unit
* tests (the markdown / candidate-parts extraction branches). @internal
*/
export function processAntigravitySSEPayload(
payload: string,
collected: AntigravityCollectedStream,
log?: { debug?: (scope: string, message: string) => void }
) {
if (!payload || payload === "[DONE]") return;
try {
const parsed = JSON.parse(payload);
const markdown =
typeof parsed?.markdown === "string"
? parsed.markdown
: typeof parsed?.response?.markdown === "string"
? parsed.response.markdown
: null;
if (markdown) {
collected.textContent += markdown;
}
const candidate = parsed?.response?.candidates?.[0];
if (candidate?.content?.parts) {
for (const part of candidate.content.parts) {
if (typeof part.text === "string" && !part.thought && !part.thoughtSignature) {
const textualToolCall = parseAntigravityTextualToolCall(part.text);
if (textualToolCall) {
addAntigravityTextualToolCall(collected, textualToolCall);
} else {
collected.textContent += part.text;
}
}
}
}
if (candidate?.finishReason) {
collected.finishReason = normalizeOpenAICompatibleFinishReasonString(
String(candidate.finishReason).toLowerCase()
);
}
if (parsed?.response?.usageMetadata) {
const um = parsed.response.usageMetadata;
collected.usage = {
prompt_tokens: um.promptTokenCount || 0,
completion_tokens: um.candidatesTokenCount || 0,
total_tokens: um.totalTokenCount || 0,
};
}
if (Array.isArray(parsed?.remainingCredits)) {
collected.remainingCredits = parsed.remainingCredits;
}
} catch {
log?.debug?.("SSE_PARSE", `Skipping malformed SSE line: ${payload.slice(0, 80)}`);
}
}

function processAntigravitySSEText(
text: string,
partialLine: { value: string },
collected: AntigravityCollectedStream,
log?: { debug?: (scope: string, message: string) => void }
) {
partialLine.value += text;
const lines = partialLine.value.split("\n");
partialLine.value = lines.pop() || "";

for (const line of lines) {
const trimmed = line.trim();
if (!trimmed.startsWith("data:")) continue;
processAntigravitySSEPayload(trimmed.slice(5).trim(), collected, log);
}
}

function flushAntigravitySSEText(
partialLine: { value: string },
collected: AntigravityCollectedStream,
log?: { debug?: (scope: string, message: string) => void }
) {
const trimmed = partialLine.value.trim();
partialLine.value = "";
if (!trimmed.startsWith("data:")) return;
processAntigravitySSEPayload(trimmed.slice(5).trim(), collected, log);
}

/**
* Strip provider prefixes (e.g. "antigravity/model" → "model").
Expand Down
148 changes: 148 additions & 0 deletions open-sse/executors/antigravity/sseCollect.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
// Pure SSE-payload -> collected-stream parsing for the Antigravity executor.
// Extracted verbatim from antigravity.ts (no host state, no fetch/auth).
import { normalizeOpenAICompatibleFinishReasonString } from "../../utils/finishReason.ts";

export type AntigravityCollectedStream = {
textContent: string;
finishReason: string;
toolCalls: Array<{
id: string;
index: number;
type: "function";
function: { name: string; arguments: string };
}>;
usage: Record<string, unknown> | null;
remainingCredits: Array<{ creditType: string; creditAmount: string }> | null;
};

export function stripZeroWidth(value: unknown): unknown {
if (typeof value === "string") {
return value.replace(/[\u200B-\u200D\uFEFF]/g, "");
}
if (Array.isArray(value)) {
return value.map((item) => stripZeroWidth(item));
}
if (value && typeof value === "object") {
return Object.fromEntries(
Object.entries(value as Record<string, unknown>).map(([key, item]) => [
key,
stripZeroWidth(item),
])
);
}
return value;
}

export function parseAntigravityTextualToolCall(
text: unknown
): { name: string; args: unknown } | null {
if (typeof text !== "string") return 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;
}
}

export function addAntigravityTextualToolCall(
collected: AntigravityCollectedStream,
parsed: { name: string; args: unknown }
): void {
collected.toolCalls.push({
id: `${parsed.name}-${Date.now()}-${collected.toolCalls.length}`,
index: collected.toolCalls.length,
type: "function",
function: {
name: parsed.name,
arguments: JSON.stringify(parsed.args || {}),
},
});
collected.finishReason = "tool_calls";
}

export function processAntigravitySSEPayload(
payload: string,
collected: AntigravityCollectedStream,
log?: { debug?: (scope: string, message: string) => void }
) {
if (!payload || payload === "[DONE]") return;
try {
const parsed = JSON.parse(payload);
const markdown =
typeof parsed?.markdown === "string"
? parsed.markdown
: typeof parsed?.response?.markdown === "string"
? parsed.response.markdown
: null;
if (markdown) {
collected.textContent += markdown;
}
const candidate = parsed?.response?.candidates?.[0];
if (candidate?.content?.parts) {
for (const part of candidate.content.parts) {
if (typeof part.text === "string" && !part.thought && !part.thoughtSignature) {
const textualToolCall = parseAntigravityTextualToolCall(part.text);
if (textualToolCall) {
addAntigravityTextualToolCall(collected, textualToolCall);
} else {
collected.textContent += part.text;
}
}
}
}
if (candidate?.finishReason) {
collected.finishReason = normalizeOpenAICompatibleFinishReasonString(
String(candidate.finishReason).toLowerCase()
);
}
if (parsed?.response?.usageMetadata) {
const um = parsed.response.usageMetadata;
collected.usage = {
prompt_tokens: um.promptTokenCount || 0,
completion_tokens: um.candidatesTokenCount || 0,
total_tokens: um.totalTokenCount || 0,
};
}
if (Array.isArray(parsed?.remainingCredits)) {
collected.remainingCredits = parsed.remainingCredits;
}
} catch {
log?.debug?.("SSE_PARSE", `Skipping malformed SSE line: ${payload.slice(0, 80)}`);
}
}

export function processAntigravitySSEText(
text: string,
partialLine: { value: string },
collected: AntigravityCollectedStream,
log?: { debug?: (scope: string, message: string) => void }
) {
partialLine.value += text;
const lines = partialLine.value.split("\n");
partialLine.value = lines.pop() || "";

for (const line of lines) {
const trimmed = line.trim();
if (!trimmed.startsWith("data:")) continue;
processAntigravitySSEPayload(trimmed.slice(5).trim(), collected, log);
}
}

export function flushAntigravitySSEText(
partialLine: { value: string },
collected: AntigravityCollectedStream,
log?: { debug?: (scope: string, message: string) => void }
) {
const trimmed = partialLine.value.trim();
partialLine.value = "";
if (!trimmed.startsWith("data:")) return;
processAntigravitySSEPayload(trimmed.slice(5).trim(), collected, log);
}
48 changes: 48 additions & 0 deletions tests/unit/antigravity-executor-split.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
import { test } from "node:test";
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import { fileURLToPath } from "node:url";
import { dirname, join } from "node:path";

// Split-guard for the antigravity executor SSE-collect extraction.
// The pure SSE-payload -> collected-stream parser lives in antigravity/sseCollect.ts
// (no host state, no fetch/auth). Host imports the helpers it uses and re-exports
// processAntigravitySSEPayload for external importers (tests).
const HERE = dirname(fileURLToPath(import.meta.url));
const EXE = join(HERE, "../../open-sse/executors");
const HOST = join(EXE, "antigravity.ts");
const LEAF = join(EXE, "antigravity/sseCollect.ts");

test("leaf hosts the SSE-collect helpers and does not import the host", () => {
const src = readFileSync(LEAF, "utf8");
for (const sym of [
"processAntigravitySSEPayload",
"processAntigravitySSEText",
"flushAntigravitySSEText",
"stripZeroWidth",
]) {
assert.match(src, new RegExp(`export (function|type) ${sym}\\b`));
}
assert.doesNotMatch(src, /from "\.\.\/antigravity\.ts"/);
});

test("host re-exports processAntigravitySSEPayload", () => {
const host = readFileSync(HOST, "utf8");
assert.match(
host,
/export \{ processAntigravitySSEPayload \} from "\.\/antigravity\/sseCollect\.ts"/
);
assert.match(host, /from "\.\/antigravity\/sseCollect\.ts"/);
});

test("SSE-collect helpers are callable and tolerate empty/garbage input", async () => {
const { processAntigravitySSEPayload, stripZeroWidth } =
await import("../../open-sse/executors/antigravity/sseCollect.ts");
assert.equal(typeof processAntigravitySSEPayload, "function");
// stripZeroWidth removes zero-width markers from strings, passes through non-strings.
assert.equal(stripZeroWidth("a​b"), "ab");
assert.deepEqual(stripZeroWidth(42), 42);
// A malformed payload must not throw (defensive parse).
const collected = { textContent: "" };
assert.doesNotThrow(() => processAntigravitySSEPayload("not-json", collected));
});
Loading