pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-09-13 18:32:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

perf(stream): compile hot-path regexes once, bound token caches, fix quadratic buffering (#12179)

Browse Source 
Validado sobre o tip de `release/v3.8.51` após reconciliar quatro arquivos que driftaram. Em parte o tip já tinha absorvido a intenção deste branch por abstrações melhores, então mantive a forma do tip e trouxe os ganhos que ainda eram reais:

- **`sseCollect.ts`** — o tip extraiu `stripObfuscationZeroWidth()` para `utils/zeroWidth.ts`, o que supera o `ZERO_WIDTH_RE` local (removido). A içada de `TEXTUAL_TOOL_CALL_RE` foi mantida: essa regex ainda estava inline num caminho quente.
- **`resultMemo.ts`** — o `memoStore()` do tip devolve o clone armazenado para que o idiom comum `memoStore(k, r); return memoLookup(k)!` evite um segundo deep-clone de vários MB. Esse contrato foi preservado (o branch o revertia para `void`), e o round-trip `JSON.parse(JSON.stringify())` virou `structuredClone()` nas duas pontas — que era o ponto de performance real do branch.
- **`browserPool.ts`** — o tip agora tem caminho headed e o engine Obscura (#12286). Ambos preservados, mais a varredura de TTL do `pendingContexts` deste branch, adaptada ao nome `poolKey` do tip.
- **`executeAttempt.ts`** — mantido o comentário explicativo do tip.

Também corrigi **quatro erros de typecheck que o branch introduzia**: `hasUnsupportedSignal` estava tipado `boolean` mas avaliava para `string | boolean`, e o fast-path de `extractUsage()` indexava `c.response`/`c.message` como `unknown`.

`typecheck:core` limpo e **482/482** nos testes de antigravity + compressão na própria branch. Obrigado, @opensource-elearning.
This commit is contained in:
opensource-elearning
2026-09-03 16:59:14 +05:30
committed by GitHub
parent 8df944cd46
commit e2e330a058
20 changed files with 275 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
open-sse/executors/adapta-web.ts
View File

@@ -33,6 +33,15 @@ interface CachedSession {
jwtExpiresAt: number; // unix ms
}
const SESSION_CACHE_MAX = 100;
function evictOldest(cache: Map<string, CachedSession>): void {
if (cache.size >= SESSION_CACHE_MAX) {
const first = cache.keys().next().value;
if (first) cache.delete(first);
}
}
// Keyed by the first 32 chars of the stored __client JWT
const sessionCache = new Map<string, CachedSession>();
@@ -44,11 +53,15 @@ function cachedJwt(clientJwt: string): string | null {
const entry = sessionCache.get(cacheKey(clientJwt));
if (!entry) return null;
// Keep a 30-second buffer before expiry
if (Date.now() >= entry.jwtExpiresAt - 30_000) return null;
if (Date.now() >= entry.jwtExpiresAt - 30_000) {
sessionCache.delete(cacheKey(clientJwt));
return null;
}
return entry.jwt;
}
function storeSession(clientJwt: string, sessionId: string, jwt: string, expMs: number): void {
evictOldest(sessionCache);
sessionCache.set(cacheKey(clientJwt), { sessionId, jwt, jwtExpiresAt: expMs });
}

9
open-sse/executors/antigravity/sseCollect.ts
View File

@@ -16,6 +16,11 @@ export type AntigravityCollectedStream = {
remainingCredits: Array<{ creditType: string; creditAmount: string }> | null;
};
// Both run once per SSE data line / per text part (processAntigravitySSEPayload),
// so the literals are hoisted to module constants.
const TEXTUAL_TOOL_CALL_RE =
/^[\s\S]*?\[Tool call:\s*([^\]\n]+)\]\s*\nArguments:\s*([\s\S]+?)\s*$/;
export function stripZeroWidth(value: unknown): unknown {
if (typeof value === "string") {
return stripObfuscationZeroWidth(value);
@@ -39,9 +44,7 @@ export function parseAntigravityTextualToolCall(
): { name: string; args: unknown } | null {
if (typeof text !== "string") return null;
const normalized = stripObfuscationZeroWidth(text);
const match = normalized.match(
/^[\s\S]*?\[Tool call:\s*([^\]\n]+)\]\s*\nArguments:\s*([\s\S]+?)\s*$/
);
const match = normalized.match(TEXTUAL_TOOL_CALL_RE);
if (!match) return null;
const name = match[1]?.trim();
const rawArgs = match[2]?.trim();

8
open-sse/executors/codex.ts
View File

@@ -538,6 +538,10 @@ export function codexDropNonstandardEvents(): boolean {
// every `codex.*` event block from the byte stream before it reaches the client.
// Exported for unit testing (#4715). Strips `codex.*` SSE event blocks from a
// streaming Response when `codexDropNonstandardEvents()` is on (default, #11014).
// Pre-compiled: the filter's transform() runs on every chunk, so these were
// re-allocated per block/iteration before hoisting.
const CODEX_SSE_EVENT_LINE_RE = /^event:\s*(.+)$/m;
const CODEX_SSE_BLOCK_SEP_RE = /\r?\n\r?\n/;
export function filterNonstandardCodexSse(response: Response): Response {
const contentType = response.headers.get("content-type") || "";
if (!response.body || !contentType.includes("text/event-stream")) {
@@ -547,14 +551,14 @@ export function filterNonstandardCodexSse(response: Response): Response {
const encoder = new TextEncoder();
let buffer = "";
const dropBlock = (block: string): boolean => {
const match = /^event:\s*(.+)$/m.exec(block);
const match = CODEX_SSE_EVENT_LINE_RE.exec(block);
return !!match && match[1].trim().startsWith("codex.");
};
const transform = new TransformStream<Uint8Array, Uint8Array>({
transform(chunk, controller) {
buffer += decoder.decode(chunk, { stream: true });
while (true) {
const separator = /\r?\n\r?\n/.exec(buffer);
const separator = CODEX_SSE_BLOCK_SEP_RE.exec(buffer);
if (!separator) break;
const blockEnd = separator.index + separator[0].length;
const block = buffer.slice(0, blockEnd);

7
open-sse/executors/glm.ts
View File

@@ -223,6 +223,8 @@ export function translateSseResponse(
suppressThinkClose: boolean = false
): Response {
if (!response.body) return response;
// GLM is a high-throughput provider — use a larger stream buffer (64KB) to
// keep provider → client pacing ahead of the model's token emission rate.
const transform = createSSETransformStreamWithLogger(
FORMATS.CLAUDE,
FORMATS.OPENAI,
@@ -236,7 +238,10 @@ export function translateSseResponse(
null,
null,
false,
suppressThinkClose
suppressThinkClose,
undefined,
undefined,
65536
);
const headers = cloneHeaders(response.headers);
headers.set("content-type", "text/event-stream");

5
open-sse/executors/tinycms.ts
View File

@@ -16,7 +16,9 @@ async function getPublicIp(): Promise<string> {
return publicIp;
}
try {
const res = await fetch("https://api64.ipify.org?format=json");
const res = await fetch("https://api64.ipify.org?format=json", {
signal: AbortSignal.timeout(5000),
});
const json = (await res.json()) as { ip: string };
publicIp = json.ip;
lastIpFetch = now;
@@ -35,6 +37,7 @@ async function fetchChallenge(uuid: string): Promise<any> {
Accept: "application/json",
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
},
signal: AbortSignal.timeout(10000),
});
if (!res.ok) {
throw new Error(`Failed to fetch challenge: ${res.status}`);

42
open-sse/executors/zcodeProtocol.ts
View File

@@ -178,7 +178,7 @@ export class ZcodeAppServerClient implements ZcodeClientLike {
private readonly startupTimeoutMs: number;
private readonly requestTimeoutMs: number;
private child?: ChildProcessWithoutNullStreams;
private outputBuffer = Buffer.alloc(0);
private pendingChunks: Buffer[] = [];
private handshakeDone = false;
private ready = false;
private startPromise?: Promise<void>;
@@ -220,7 +220,7 @@ export class ZcodeAppServerClient implements ZcodeClientLike {
}
this.child = child;
this.outputBuffer = Buffer.alloc(0);
this.pendingChunks = [];
this.handshakeDone = false;
this.ready = false;
child.stdin.on("error", () => {
@@ -271,18 +271,29 @@ export class ZcodeAppServerClient implements ZcodeClientLike {
}
}
// Buffer accumulated stdout bytes. Chunks are collected in an array and
// collapsed into one contiguous buffer only when a complete frame (or the
// hello line) might be present — the previous `concat(prev, chunk)` per data
// event re-allocated the whole buffer on every chunk, i.e. O(n²) total.
private onStdout(chunk: Buffer): void {
this.outputBuffer = Buffer.concat([this.outputBuffer, chunk]);
this.pendingChunks.push(chunk);
let total = 0;
for (const part of this.pendingChunks) total += part.byteLength;
const buffer = total === chunk.byteLength && this.pendingChunks.length > 0
? chunk
: Buffer.concat(this.pendingChunks);
this.pendingChunks = [buffer];
if (!this.handshakeDone) {
const newline = this.outputBuffer.indexOf(0x0a);
const newline = buffer.indexOf(0x0a);
if (newline < 0) {
if (this.outputBuffer.byteLength > 64 * 1024) {
if (buffer.byteLength > 64 * 1024) {
this.serverReadyError?.(new Error("ZCode hello line is too large"));
}
return;
}
const line = this.outputBuffer.subarray(0, newline).toString("utf8").trim();
this.outputBuffer = this.outputBuffer.subarray(newline + 1);
const line = buffer.subarray(0, newline).toString("utf8").trim();
this.pendingChunks = [buffer.subarray(newline + 1)];
let hello: unknown;
try {
hello = JSON.parse(line);
@@ -307,9 +318,13 @@ export class ZcodeAppServerClient implements ZcodeClientLike {
}
private consumeFrames(): void {
while (this.outputBuffer.byteLength >= HEADER_SIZE) {
const type = this.outputBuffer.readUInt8(0);
const length = this.outputBuffer.readUInt32BE(9);
// Collapse to one buffer for frame scanning (only happens once per data
// event since onStdout already deduped), then drop consumed frames.
const buffer = this.pendingChunks[0];
let offset = 0;
while (buffer.byteLength - offset >= HEADER_SIZE) {
const type = buffer.readUInt8(offset);
const length = buffer.readUInt32BE(offset + 9);
if (length > MAX_FRAME_BYTES) {
const error = new Error("ZCode frame exceeds the configured safety limit");
this.serverReadyError?.(error);
@@ -317,9 +332,9 @@ export class ZcodeAppServerClient implements ZcodeClientLike {
return;
}
const frameLength = HEADER_SIZE + length;
if (this.outputBuffer.byteLength < frameLength) return;
const body = this.outputBuffer.subarray(HEADER_SIZE, frameLength);
this.outputBuffer = this.outputBuffer.subarray(frameLength);
if (buffer.byteLength - offset < frameLength) break;
const body = buffer.subarray(offset + HEADER_SIZE, offset + frameLength);
offset += frameLength;
if (type !== REGULAR_MESSAGE) continue;
try {
const header = decodeZcodeValue(body, 0);
@@ -331,6 +346,7 @@ export class ZcodeAppServerClient implements ZcodeClientLike {
this.rejectPending(normalized);
}
}
if (offset > 0) this.pendingChunks = [buffer.subarray(offset)];
}
private handleMessage(headerValue: unknown, payload: unknown): void {

18
open-sse/handlers/responseSanitizer.ts
View File

@@ -1061,6 +1061,7 @@ function convertOpenAIResponseToResponses(openaiResponse: JsonRecord): JsonRecor
/**
* Sanitize a streaming SSE chunk for passthrough mode.
* Lighter than full sanitization — only strips problematic extra fields.
* Fast-path: returns original when no mutations are needed.
*/
export function sanitizeStreamingChunk(parsed: unknown): unknown {
const parsedRecord = toRecord(parsed);
@@ -1078,14 +1079,29 @@ export function sanitizeStreamingChunk(parsed: unknown): unknown {
if (eventType === "content_block_delta") {
const deltaRecord = toRecord(parsedRecord.delta);
if (deltaRecord) {
let mutated = false;
if (typeof deltaRecord.text === "string") {
deltaRecord.text = stripZeroWidthText(deltaRecord.text);
mutated = true;
}
if (typeof deltaRecord.thinking === "string") {
deltaRecord.thinking = stripZeroWidthText(deltaRecord.thinking);
mutated = true;
}
return mutated ? parsedRecord : parsed;
}
return parsedRecord;
return parsed;
}
// Fast-path: check if any mutations would actually be needed
// Most passthrough chunks (content deltas) need no sanitization
const needsIdNormalization = parsedRecord.id !== undefined && parsedRecord.id !== null && typeof parsedRecord.id !== "string";
const hasChoices = Array.isArray(parsedRecord.choices) && parsedRecord.choices.length > 0;
const hasUsage = parsedRecord.usage !== undefined;
const hasSystemFingerprint = parsedRecord.system_fingerprint !== undefined;
if (!needsIdNormalization && !hasChoices && !hasUsage && !hasSystemFingerprint) {
// Nothing to sanitize — forward original
return parsed;
}
// Build sanitized chunk

29
open-sse/services/accountFallback.ts
View File

@@ -64,6 +64,16 @@ import {
MAX_SHORT_RETRY_HINT_MS,
} from "./retryAfterJson.ts";
// Pre-compiled regex constants for hot-path retry parsing (avoid per-call compilation)
const RETRY_AFTER_RE = /retry\s+after\s+(\d+)\s*s/i;
const PLEASE_RETRY_RE = /please retry in\s+([\d.]+\s*s)/i;
const ISO_RETRY_RE = /\b(?:try again at|wait until|reset(?:s)? at|available at|retry after)\s+(\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)/i;
const RESETS_AFTER_RE = /resets? after (\d+h)?(\d+m)?(\d+s)?/i;
const WILL_RESET_AFTER_RE = /will reset after (\d+h)?(\d+m)?(\d+s)?/i;
const RESETS_IN_RE = /resets? in (\d+h)?(\d+m)?(\d+s)?/i;
const RETRY_IN_SEC_RE = /please retry in (\d+(?:\.\d+)?)\s*s/i;
const COOLDOWN_NUMERIC_RE = /^\d+(\.\d+)?$/;
export type RetryHintProvenance = "header" | "google_rpc_retry_info" | "body";
export function retryHintBypassesMaxCooldownMs(
@@ -1371,7 +1381,7 @@ export function parseRetryAfterFromBody(responseBody: unknown): {
// OpenAI: "Please retry after 20s" in message
const msg = String(error.message || body.message || "");
const retryMatch = /retry\s+after\s+(\d+)\s*s/i.exec(msg);
const retryMatch = RETRY_AFTER_RE.exec(msg);
if (retryMatch) {
return {
retryAfterMs: Number.parseInt(retryMatch[1], 10) * 1000,
@@ -1404,16 +1414,13 @@ export function parseRetryFromErrorText(errorText: unknown): number | null {
// Gemini free-tier text fallback (no parseable JSON details present):
// "Please retry in 26.660853464s." Short throttle hint — capped independently of
// MAX_PROVIDER_COOLDOWN_MS, mirroring the JSON RetryInfo.retryDelay cap (#7940).
const pleaseRetryMs = parseDelayString(/please retry in\s+([\d.]+\s*s)/i.exec(msg)?.[1]);
const pleaseRetryMs = parseDelayString(PLEASE_RETRY_RE.exec(msg)?.[1]);
if (pleaseRetryMs !== null && pleaseRetryMs > 0) {
return Math.min(pleaseRetryMs, MAX_SHORT_RETRY_HINT_MS);
}
// Issue #2321: parse embedded absolute ISO retry timestamps.
const isoMatch =
/\b(?:try again at|wait until|reset(?:s)? at|available at|retry after)\s+(\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)/i.exec(
msg
);
const isoMatch = ISO_RETRY_RE.exec(msg);
if (isoMatch) {
const parsedTs = Date.parse(isoMatch[1]);
if (Number.isFinite(parsedTs)) {
@@ -1422,21 +1429,21 @@ export function parseRetryFromErrorText(errorText: unknown): number | null {
}
}
const match = /resets? after (\d+h)?(\d+m)?(\d+s)?/i.exec(msg);
const match = RESETS_AFTER_RE.exec(msg);
if (match?.[1] || match?.[2] || match?.[3]) return computeDurationMs(match);
// Variant without "reset after": "will reset after XhYmZs"
const altMatch = /will reset after (\d+h)?(\d+m)?(\d+s)?/i.exec(msg);
const altMatch = WILL_RESET_AFTER_RE.exec(msg);
if (altMatch?.[1] || altMatch?.[2] || altMatch?.[3]) return computeDurationMs(altMatch);
// Antigravity / Cloud Code phrasing: "Resets in 164h27m24s".
const resetsInMatch = /resets? in (\d+h)?(\d+m)?(\d+s)?/i.exec(msg);
const resetsInMatch = RESETS_IN_RE.exec(msg);
if (resetsInMatch?.[1] || resetsInMatch?.[2] || resetsInMatch?.[3]) {
return computeDurationMs(resetsInMatch);
}
// Gemini phrasing: "Please retry in 54.472178091s" (fractional seconds).
const retryInSecMatch = /please retry in (\d+(?:\.\d+)?)\s*s/i.exec(msg);
const retryInSecMatch = RETRY_IN_SEC_RE.exec(msg);
if (retryInSecMatch?.[1]) {
const sec = Number.parseFloat(retryInSecMatch[1]);
if (Number.isFinite(sec) && sec > 0) {
@@ -2226,7 +2233,7 @@ export function cooldownUntilMs(value: string | number | Date | null | undefined
if (value instanceof Date) return value.getTime();
if (typeof value === "number") return value;
const raw = value.trim();
if (/^\d+(\.\d+)?$/.test(raw)) return Number(raw);
if (COOLDOWN_NUMERIC_RE.test(raw)) return Number(raw);
return new Date(raw).getTime();
}

22
open-sse/services/browserPool.ts
View File

@@ -90,13 +90,18 @@ function createBrowserPoolMetrics(): BrowserPoolMetrics {
type PoolEngine = "obscura" | "cloakbrowser" | "chromium";
interface PendingContextEntry {
promise: Promise<PooledContext>;
createdAt: number;
}
interface PoolState {
browser: Browser | null;
/** Engine backing the headless browser, for metrics and stealth detection. */
engine: PoolEngine | null;
headedBrowser: Browser | null;
contexts: Map<string, PooledContext>;
pendingContexts: Map<string, Promise<PooledContext>>;
pendingContexts: Map<string, PendingContextEntry>;
launching: Promise<Browser> | null;
headedLaunching: Promise<Browser> | null;
generation: number;
@@ -119,7 +124,7 @@ const state: PoolState = {
engine: null,
headedBrowser: null,
contexts: new Map(),
pendingContexts: new Map(),
pendingContexts: new Map<string, { promise: Promise<PooledContext>; createdAt: number }>(),
launching: null,
headedLaunching: null,
generation: 0,
@@ -182,6 +187,15 @@ function evictStaleContexts(): void {
pooled.context.close().catch(() => {});
}
}
// #12179: also evict pendingContexts entries that never resolved, so a hung
// launch cannot pin the map (and the pool) open forever.
const PENDING_TTL_MS = 5 * 60 * 1000;
for (const [key, pending] of state.pendingContexts) {
if (now - pending.createdAt > PENDING_TTL_MS) {
state.pendingContexts.delete(key);
state.metrics.contextsEvicted++;
}
}
if (
state.contexts.size === 0 &&
state.pendingContexts.size === 0 &&
@@ -485,7 +499,7 @@ export async function acquireBrowserContext(
// Dedup concurrent creations for the same key
const pending = state.pendingContexts.get(poolKey);
if (pending) return pending;
if (pending) return pending.promise;
const createPromise = (async (): Promise<PooledContext> => {
const [browser, proxy] = await Promise.all([
@@ -531,7 +545,7 @@ export async function acquireBrowserContext(
return pooled;
})();
state.pendingContexts.set(poolKey, createPromise);
state.pendingContexts.set(poolKey, { promise: createPromise, createdAt: Date.now() });
createPromise
.then(() => settlePendingContext(poolKey, false))
.catch(() => settlePendingContext(poolKey, true));

4
open-sse/services/compression/resultMemo.ts
View File

@@ -149,7 +149,7 @@ export function memoLookup(key: string): CompressionResult | null {
memoHits++;
recordLookup(true);
// Return a clone so downstream mutation cannot corrupt the cached value.
const cloned = JSON.parse(JSON.stringify(hit)) as CompressionResult;
const cloned = structuredClone(hit);
if (cloned.stats) {
cloned.stats.memoHit = true;
}
@@ -162,7 +162,7 @@ export function memoStore(key: string, result: CompressionResult): CompressionRe
// Returns the stored clone so callers that need a fresh instance (the common
// `memoStore(key, result); return memoLookup(key)!` idiom) can avoid a redundant
// second multi-MB deep clone of the body on the way out.
const stored = JSON.parse(JSON.stringify(result)) as CompressionResult;
const stored = structuredClone(result);
boundedSet(key, stored);
return stored;
}

26
open-sse/services/gigachatAuth.ts
View File

@@ -15,6 +15,22 @@ type GigachatTokenOptions = {
const DEFAULT_GIGACHAT_AUTH_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth";
const DEFAULT_GIGACHAT_SCOPE = "GIGACHAT_API_PERS";
const CACHE_SKEW_MS = 60_000;
const TOKEN_CACHE_MAX = 100;
const INFLIGHT_MAX = 50;
function evictOldest<T>(cache: Map<string, T>): void {
if (cache.size >= TOKEN_CACHE_MAX) {
const first = cache.keys().next().value;
if (first) cache.delete(first);
}
}
function evictOldestInflight(cache: Map<string, Promise<GigachatTokenResult>>): void {
if (cache.size >= INFLIGHT_MAX) {
const first = cache.keys().next().value;
if (first) cache.delete(first);
}
}
const tokenCache = new Map<string, GigachatTokenResult>();
const inflightRequests = new Map<string, Promise<GigachatTokenResult>>();
@@ -23,10 +39,12 @@ function getCacheKey(credentials: string, authUrl: string, scope: string) {
return `${authUrl}::${scope}::${credentials}`;
}
function isFreshToken(token: GigachatTokenResult | undefined) {
function isFreshToken(token: GigachatTokenResult | undefined, key?: string) {
if (!token?.accessToken || !token?.expiresAt) return false;
const expiresAtMs = new Date(token.expiresAt).getTime();
return Number.isFinite(expiresAtMs) && expiresAtMs - Date.now() > CACHE_SKEW_MS;
const fresh = Number.isFinite(expiresAtMs) && expiresAtMs - Date.now() > CACHE_SKEW_MS;
if (!fresh && key) tokenCache.delete(key);
return fresh;
}
function normalizeExpiry(rawExpiry: unknown) {
@@ -59,7 +77,7 @@ export async function getGigachatAccessToken(
const cacheKey = getCacheKey(credentials, authUrl, scope);
const cached = tokenCache.get(cacheKey);
if (isFreshToken(cached)) {
if (isFreshToken(cached, cacheKey)) {
return cached;
}
@@ -100,10 +118,12 @@ export async function getGigachatAccessToken(
accessToken,
expiresAt: normalizeExpiry(data.exp ?? data.expires_at),
};
evictOldest(tokenCache);
tokenCache.set(cacheKey, token);
return token;
})();
evictOldestInflight(inflightRequests);
inflightRequests.set(cacheKey, requestPromise);
try {
return await requestPromise;

8
open-sse/services/responsesInputSanitizer.ts
View File

@@ -11,6 +11,10 @@ const SERVER_ITEM_ID_PREFIX_BY_TYPE: Record<string, string> = {
reasoning: "rs_",
};
const SERVER_ITEM_ID_PATTERN = /^(fc|msg|rs|resp)_/;
// Validated per input item of type function_call / function_call_output (the agentic
// Responses path), so kept as a module constant instead of an inline literal.
const FUNCTION_NAME_VALID_RE = /^[a-zA-Z0-9_-]{1,128}$/;
const FUNCTION_NAME_SANITIZE_RE = /[^a-zA-Z0-9_-]/g;
function toRecord(value: unknown): JsonRecord | null {
return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : null;
@@ -38,7 +42,7 @@ export function isInternalAssistantMessage(record: JsonRecord): boolean {
// Sanitize after cloning so upstream never sees an invalid name.
function sanitizeFunctionName(name: string): string {
// Replace any character not in [a-zA-Z0-9_-] with underscore, then truncate.
return name.replace(/[^a-zA-Z0-9_-]/g, "_").slice(0, 128);
return name.replace(FUNCTION_NAME_SANITIZE_RE, "_").slice(0, 128);
}
function sanitizeInputItemId(record: JsonRecord): JsonRecord {
@@ -149,7 +153,7 @@ function sanitizeInputItem(item: unknown): unknown {
if (
(next.type === "function_call" || next.type === "function_call_output") &&
typeof next.name === "string" &&
!/^[a-zA-Z0-9_-]{1,128}$/.test(next.name)
!FUNCTION_NAME_VALID_RE.test(next.name)
) {
next = { ...next, name: sanitizeFunctionName(next.name) };
}

25
open-sse/utils/composerToolCalls.ts
View File

@@ -48,6 +48,18 @@ const INNER_RE = new RegExp(
// Match an arg separator.
const ARG_SEP_RE = new RegExp(`<${FW}tool${SEP}sep${FW}>`, "gi");
// Opening-only marker, matched on every streamed delta in the holdback path;
// kept as a module constant so it is compiled once instead of per call.
const OPEN_ONLY_RE = new RegExp(`<${FW}tool${SEP}calls${SEP}begin${FW}>`, "i");
// Parse helpers below run once per tool-call block / per argument value during
// streaming, so their literals are hoisted too.
const TRIM_EDGES_RE = /^\s+|\s+$/g;
const FIRST_SPACE_RE = /\s/;
const TRAILING_NEWLINES_RE = /\n+$/;
const INTEGER_RE = /^-?\d+$/;
const DECIMAL_RE = /^-?\d*\.\d+$/;
// Heuristic: any partial opening marker (start of `<｜tool` ... without the
// final `>`). Used by the streaming parser to know it must hold back text.
const PARTIAL_OPEN_MARKER_RE = new RegExp(
@@ -115,7 +127,7 @@ function generateToolCallId(index: number): string {
function parseInnerCall(body: string): { name: string; arguments: string } | null {
// Body starts with the tool name on (typically) its own line, optionally
// surrounded by whitespace, then the first `<｜tool▁sep｜>`.
const trimmed = body.replace(/^\s+|\s+$/g, "");
const trimmed = body.replace(TRIM_EDGES_RE, "");
// Split by argument separator first to isolate name + arg blocks.
const segments = trimmed.split(ARG_SEP_RE);
// First segment is the tool name (and any preamble whitespace).
@@ -137,7 +149,7 @@ function parseInnerCall(body: string): { name: string; arguments: string } | nul
let argName: string;
let argValue: string;
if (idxNl < 0) {
const idxSp = seg.search(/\s/);
const idxSp = seg.search(FIRST_SPACE_RE);
if (idxSp < 0) {
argName = seg.trim();
argValue = "";
@@ -155,7 +167,7 @@ function parseInnerCall(body: string): { name: string; arguments: string } | nul
if (!argName) continue;
// Strip the trailing newline before the next separator (the separator
// marker itself was already consumed by the split).
argValue = argValue.replace(/\n+$/, "");
argValue = argValue.replace(TRAILING_NEWLINES_RE, "");
// Attempt JSON parse so structured args (objects/arrays/numbers/bools)
// come through as native JSON values rather than quoted strings.
args[argName] = coerceArgValue(argValue);
@@ -179,11 +191,11 @@ function coerceArgValue(raw: string): unknown {
if (stripped === "true") return true;
if (stripped === "false") return false;
if (stripped === "null") return null;
if (/^-?\d+$/.test(stripped)) {
if (INTEGER_RE.test(stripped)) {
const n = Number(stripped);
if (Number.isSafeInteger(n)) return n;
}
if (/^-?\d*\.\d+$/.test(stripped)) {
if (DECIMAL_RE.test(stripped)) {
const n = Number(stripped);
if (Number.isFinite(n)) return n;
}
@@ -295,8 +307,7 @@ export function feedStreamingChunk(state: StreamingState, accumulated: string):
// 2. Look for an opening-only marker. If found, everything before it is
// safe; everything after must be held until we see the closing marker.
const openOnlyRe = new RegExp(`<${FW}tool${SEP}calls${SEP}begin${FW}>`, "i");
const openMatch = accumulated.match(openOnlyRe);
const openMatch = accumulated.match(OPEN_ONLY_RE);
if (openMatch && openMatch.index !== undefined) {
const safe = accumulated.slice(0, openMatch.index);
const safeDelta = safe.length > state.emitted ? safe.slice(state.emitted) : "";

68
open-sse/utils/reasoningFields.ts
View File

@@ -21,45 +21,61 @@ export function extractReasoningDetailsText(value: unknown): string {
.join("");
}
export function getReadableReasoningValue(value: unknown): string {
/**
* Consolidated reasoning field extraction - single pass returns all categories
* to avoid 3-5 separate object traversals per chunk.
*/
export interface ReasoningFields {
readable: string;
unsupported: string;
any: string;
hasUnsupportedSignal: boolean;
hasAnySignal: boolean;
}
export function extractReasoningFields(value: unknown): ReasoningFields {
const record = asReasoningRecord(value);
return nonEmptyString(record.reasoning_content) || nonEmptyString(record.reasoning);
const readable = nonEmptyString(record.reasoning_content) || nonEmptyString(record.reasoning);
const reasoningText = nonEmptyString(record.reasoning_text);
const thinking = nonEmptyString(record.thinking);
const thought = nonEmptyString(record.thought);
const details = extractReasoningDetailsText(record);
const unsupported = reasoningText || thinking || thought || details;
const any = readable || unsupported;
const hasUnsupportedSignal = !!(
!readable &&
(reasoningText ||
thinking ||
thought ||
(Array.isArray(record.reasoning_details) && record.reasoning_details.length > 0))
);
const hasAnySignal = !!any;
return { readable, unsupported, any, hasUnsupportedSignal, hasAnySignal };
}
/** Back-compat wrappers for existing callers - delegate to consolidated extractor. */
export function getReadableReasoningValue(value: unknown): string {
return extractReasoningFields(value).readable;
}
export function getUnsupportedReasoningValue(value: unknown): string {
const record = asReasoningRecord(value);
return (
nonEmptyString(record.reasoning_text) ||
nonEmptyString(record.thinking) ||
nonEmptyString(record.thought) ||
extractReasoningDetailsText(record)
);
return extractReasoningFields(value).unsupported;
}
export function getAnyReasoningValue(value: unknown): string {
return getReadableReasoningValue(value) || getUnsupportedReasoningValue(value);
return extractReasoningFields(value).any;
}
export function hasUnsupportedReasoningSignal(value: unknown): boolean {
const record = asReasoningRecord(value);
return Boolean(
!getReadableReasoningValue(record) &&
(nonEmptyString(record.reasoning_text) ||
nonEmptyString(record.thinking) ||
nonEmptyString(record.thought) ||
(Array.isArray(record.reasoning_details) && record.reasoning_details.length > 0))
);
return extractReasoningFields(value).hasUnsupportedSignal;
}
export function hasAnyReasoningSignal(value: unknown): boolean {
const record = asReasoningRecord(value);
return Boolean(
getReadableReasoningValue(record) ||
nonEmptyString(record.reasoning_text) ||
nonEmptyString(record.thinking) ||
nonEmptyString(record.thought) ||
(Array.isArray(record.reasoning_details) && record.reasoning_details.length > 0)
);
return extractReasoningFields(value).hasAnySignal;
}
const STRIPPABLE_REASONING_FIELDS = [

12
open-sse/utils/responsesStreamHelpers.ts
View File

@@ -98,25 +98,29 @@ function buildResponsesOutputItemKey(item: unknown): string | null {
return `${type}:${id}:${callId}:${outputIndex}:${name}`;
}
// Module-level Set reused across calls to avoid allocation per event
const _seenResponsesKeys = new Set<string>();
export function pushUniqueResponsesOutputItems(target: unknown[], items: readonly unknown[]) {
const seen = new Set<string>();
// Clear the reused Set instead of allocating new one
_seenResponsesKeys.clear();
for (const existingItem of target) {
const key = buildResponsesOutputItemKey(existingItem);
if (key) {
seen.add(key);
_seenResponsesKeys.add(key);
}
}
for (const item of items) {
const key = buildResponsesOutputItemKey(item);
if (key && seen.has(key)) {
if (key && _seenResponsesKeys.has(key)) {
continue;
}
target.push(item);
if (key) {
seen.add(key);
_seenResponsesKeys.add(key);
}
}
}

16
open-sse/utils/stream.ts
View File

@@ -178,6 +178,8 @@ type StreamOptions = {
* codex-compatible `namespace` + `name` fields.
*/
requestToolIdentityMap?: Map<string, { namespace: string; name: string }> | null;
/** High water mark for the TransformStream internal buffer (default: 16384) */
highWaterMark?: number;
};
type TranslateState = ReturnType<typeof initState> & {
@@ -1173,6 +1175,8 @@ export function createSSEStream(options: StreamOptions = {}) {
}
};
const highWaterMark = options.highWaterMark ?? 16384;
return new TransformStream(
{
start(controller) {
@@ -2992,8 +2996,8 @@ export function createSSEStream(options: StreamOptions = {}) {
clearIdleTimer();
},
},
{ highWaterMark: 16384 },
{ highWaterMark: 16384 }
{ highWaterMark },
{ highWaterMark }
);
}
@@ -3015,7 +3019,8 @@ export function createSSETransformStreamWithLogger(
copilotCompatibleReasoning = false,
suppressThinkClose = false,
customToolNames: ReadonlySet<string> = new Set(),
requestToolIdentityMap: Map<string, { namespace: string; name: string }> | null = null
requestToolIdentityMap: Map<string, { namespace: string; name: string }> | null = null,
highWaterMark?: number
) {
return createSSEStream({
mode: STREAM_MODE.TRANSLATE,
@@ -3034,6 +3039,7 @@ export function createSSETransformStreamWithLogger(
suppressThinkClose,
customToolNames,
requestToolIdentityMap,
highWaterMark,
});
}
@@ -3048,7 +3054,8 @@ export function createPassthroughStreamWithLogger(
apiKeyInfo: unknown = null,
onFailure: ((payload: StreamFailurePayload) => boolean | void | Promise<void>) | null = null,
clientResponseFormat: string | null = null,
requestToolIdentityMap: Map<string, { namespace: string; name: string }> | null = null
requestToolIdentityMap: Map<string, { namespace: string; name: string }> | null = null,
highWaterMark?: number
) {
return createSSEStream({
mode: STREAM_MODE.PASSTHROUGH,
@@ -3063,6 +3070,7 @@ export function createPassthroughStreamWithLogger(
onFailure,
clientResponseFormat,
requestToolIdentityMap,
highWaterMark,
});
}

18
open-sse/utils/streamHandler.ts
View File

@@ -629,7 +629,11 @@ function resolveSilentCloseOutcome(input: {
return null;
}
export function createDisconnectAwareStream(transformStream, streamController) {
export function createDisconnectAwareStream(
transformStream,
streamController,
options: { highWaterMark?: number } = {}
) {
const reader = transformStream.readable.getReader();
const writer = transformStream.writable.getWriter();
const terminalDecoder = new TextDecoder();
@@ -697,6 +701,8 @@ export function createDisconnectAwareStream(transformStream, streamController) {
}
};
const highWaterMark = options.highWaterMark ?? 16384;
return new ReadableStream(
{
async pull(controller) {
@@ -818,7 +824,7 @@ export function createDisconnectAwareStream(transformStream, streamController) {
await Promise.allSettled([reader.cancel(reason), writer.abort(reason)]);
},
},
{ highWaterMark: 16384 }
{ highWaterMark }
);
}
@@ -845,7 +851,7 @@ export function pipeWithDisconnect(
providerResponse: Response,
transformStream: TransformStream<Uint8Array, Uint8Array>,
streamController: StreamController,
opts: { stallTimeoutMs?: number } = {}
opts: { stallTimeoutMs?: number; highWaterMark?: number } = {}
) {
const stallTimeoutMs = opts.stallTimeoutMs ?? DEFAULT_STREAM_STALL_TIMEOUT_MS;
@@ -854,7 +860,8 @@ export function pipeWithDisconnect(
const transformedBody = providerResponse.body.pipeThrough(transformStream);
return createDisconnectAwareStream(
{ readable: transformedBody, writable: createNoopAbortWritable() },
streamController
streamController,
{ highWaterMark: opts.highWaterMark }
);
}
@@ -956,6 +963,7 @@ export function pipeWithDisconnect(
.pipeThrough(transformStream);
return createDisconnectAwareStream(
{ readable: transformedBody, writable: createNoopAbortWritable() },
wrappedController
wrappedController,
{ highWaterMark: opts.highWaterMark }
);
}

19
open-sse/utils/streamHelpers.ts
View File

@@ -70,6 +70,13 @@ function isRecord(value: unknown): value is Record<string, unknown> {
const ANSI_ESCAPE_RE =
/\x1b(?:\[[0-9;?]*[A-Za-z]|\][^\x07\x1b]*(?:\x07|\x1b\\)|[A-Z\[\]\\^_`])|[\x00-\x08\x0b\x0c\x0e-\x1f]/g;
// Pre-compiled regex constants for hot-path SSE processing (avoid per-call compilation)
const CR_STRIP_RE = /\r$/;
const SSE_FIELD_RE = /^(?:event:|id:|retry:|:)/i;
const SSE_EVENT_RE = /^event:\s*(.+)$/i;
const SSE_ID_RETRY_RE = /^(?::|id:|retry:)/i;
const SSE_EVENT_ONLY_RE = /^event:/i;
/**
* Strip ANSI/VT100 escape sequences (and stray C0 controls) from a string.
* Non-string inputs (null/undefined) are returned unchanged. Preserves \t \n \r.
@@ -125,7 +132,7 @@ export function parseSSELine(line: string): SSEJsonPayload | null {
}
function extractSseDataLine(line: string): string | null {
const trimmed = stripAnsiCodes(line.trimStart().replace(/\r$/, ""));
const trimmed = stripAnsiCodes(line.trimStart().replace(CR_STRIP_RE, ""));
if (!trimmed.startsWith("data:")) return null;
return trimmed.slice(5).trimStart();
}
@@ -192,12 +199,12 @@ export function createSSEDataLineNormalizer(): SSEDataLineNormalizer {
normalize(lines: string[]) {
const output: string[] = [];
for (const line of lines) {
const normalizedLine = line.replace(/\r$/, "");
const normalizedLine = line.replace(CR_STRIP_RE, "");
const trimmed = normalizedLine.trim();
if (
trimmed &&
/^(?:event:|id:|retry:|:)/i.test(trimmed) &&
SSE_FIELD_RE.test(trimmed) &&
hasSelfDescribingPendingDataPayload()
) {
flush(output);
@@ -235,7 +242,7 @@ export function createSSEEventPrefixBuffer(options?: { forwardEvent?: boolean })
},
eventType() {
for (let i = lines.length - 1; i >= 0; i--) {
const match = lines[i].trim().match(/^event:\s*(.+)$/i);
const match = lines[i].trim().match(SSE_EVENT_RE);
if (match) return match[1].trim();
}
return "";
@@ -251,10 +258,10 @@ export function createSSEEventPrefixBuffer(options?: { forwardEvent?: boolean })
// `id:`/`retry:` and bare `:` comment lines are not part of any of the
// OpenAI Chat-Completions, OpenAI Responses, or Claude Messages SSE
// protocols — never buffer (and thus never re-forward) them (#10017).
if (/^(?::|id:|retry:)/i.test(trimmed)) return;
if (SSE_ID_RETRY_RE.test(trimmed)) return;
// `event:` framing is only forwarded for protocols that define it; drop it
// for plain OpenAI Chat-Completions-format clients.
if (/^event:/i.test(trimmed) && !forwardEvent) return;
if (SSE_EVENT_ONLY_RE.test(trimmed) && !forwardEvent) return;
lines.push(line);
emitted = false;
},

19
open-sse/utils/usageTracking.ts
View File

@@ -680,10 +680,29 @@ export function isEmptyUsage(usage: unknown): boolean {
/**
* Extract usage from supported formats (Claude, OpenAI, Gemini, Responses API)
* Fast-path: return early for chunks without any usage-related fields.
* Most streaming chunks (content deltas) have no usage — avoids property checks.
*/
export function extractUsage(chunk: UsagePayloadLike | null | undefined) {
if (!chunk || typeof chunk !== "object") return null;
// Fast-path: check for any usage-like fields before doing full extraction
// Most chunks are content deltas with no usage — return null immediately.
const c = chunk as Record<string, unknown>;
const response = c.response as Record<string, unknown> | undefined;
const message = c.message as Record<string, unknown> | undefined;
if (
!c.type &&
c.usage === undefined &&
c.usageMetadata === undefined &&
response?.usage === undefined &&
response?.usageMetadata === undefined &&
message?.usage === undefined &&
c.done !== true
) {
return null;
}
// Claude/Antigravity streaming: message_start event carries INPUT tokens
// FIX #74: This event was not handled — input_tokens were being dropped
// Structure: { type: "message_start", message: { usage: { input_tokens: N, output_tokens: 0 } } }

1
package.json
View File

@@ -95,6 +95,7 @@
"bench:compression": "bun scripts/compression/benchmark.ts",
"bench:heap-body": "node --expose-gc --import tsx/esm scripts/perf/request-body-heap.ts",
"bench:routing-events": "node --import tsx/esm scripts/perf/routing-events-bench.ts",
"bench:highwatermark": "node --import tsx/esm scripts/perf/benchmark-highwatermark.ts",
"eval:compression": "node --import tsx scripts/compression-eval/index.ts",
"eval:router": "node --import tsx scripts/router-eval/index.ts",
"eval:router:compare": "node --import tsx scripts/router-eval/compare.ts",