mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-01 12:52:11 +03:00
refactor(executors): extract pure SSE-collect parsing from antigravity (#5962)
Extract the pure SSE-payload -> collected-stream parser (AntigravityCollectedStream, stripZeroWidth, parseAntigravityTextualToolCall, addAntigravityTextualToolCall, processAntigravitySSEPayload/Text, flushAntigravitySSEText) verbatim into the leaf antigravity/sseCollect.ts. Host imports the helpers it uses and re-exports processAntigravitySSEPayload for external importers (tests). Host 1812 -> 1671 LOC. Byte-identical bodies (verbatim multiset 135/135), leaf does not import the host (no cycle). Credit/quota state, auth, and HTTP dispatch untouched. Adds a split-guard; consumer tests stay green (executor-agy 8, executor-antigravity 26, antigravity-sse-collect-socket-release, copilot-agent-antigravity-parity 6).
This commit is contained in:
Diego Rodrigues de Sa e Souza
committed by
GitHub
GitHub
parent
4eaa2f4c65
commit
81d640dccc
@@ -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,
|
||||
@@ -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;
|
||||
@@ -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").
|
||||
|
||||
148
open-sse/executors/antigravity/sseCollect.ts
Normal file
148
open-sse/executors/antigravity/sseCollect.ts
Normal file
@@ -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
tests/unit/antigravity-executor-split.test.ts
Normal file
48
tests/unit/antigravity-executor-split.test.ts
Normal file
@@ -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("ab"), "ab");
|
||||
assert.deepEqual(stripZeroWidth(42), 42);
|
||||
// A malformed payload must not throw (defensive parse).
|
||||
const collected = { textContent: "" };
|
||||
assert.doesNotThrow(() => processAntigravitySSEPayload("not-json", collected));
|
||||
});
|
||||
Reference in New Issue
Block a user