pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-02 13:22:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/release/v3.8.23' into fix/oauth-token-resilience

Browse Source
This commit is contained in:
diegosouzapw
2026-06-12 02:50:34 -03:00
parent 643d7a82d7 bb0019845b
commit eb537cfd58
11 changed files with 664 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
open-sse/executors/vertex.ts
View File

@@ -21,6 +21,28 @@ export function parseSAFromApiKey(apiKey: string): ServiceAccount {
}
}
/**
* A Service Account credential is a JSON object (type/client_email/private_key). A Vertex AI
* Express-mode API key is an opaque non-JSON string. Distinguishing them lets the executor
* support BOTH: Service Account JSON (JWT → OAuth → project-scoped endpoint + Bearer auth) and
* Express keys (project-less publisher endpoint + x-goog-api-key auth), instead of failing every
* Express key with "requires a valid Service Account JSON".
*/
export function looksLikeServiceAccountJson(apiKey: string): boolean {
if (!apiKey || typeof apiKey !== "string") return false;
try {
const parsed = JSON.parse(apiKey);
return !!parsed && typeof parsed === "object" && !Array.isArray(parsed);
} catch {
return false;
}
}
/** True for a Vertex AI Express-mode API key (a non-empty, non-JSON, non-OAuth credential). */
export function isExpressApiKey(apiKey?: string | null): boolean {
return typeof apiKey === "string" && apiKey.trim().length > 0 && !looksLikeServiceAccountJson(apiKey);
}
export async function getAccessToken(sa: ServiceAccount): Promise<string> {
if (!sa.client_email || !sa.private_key) {
throw new Error(
@@ -110,7 +132,13 @@ export class VertexExecutor extends BaseExecutor {
async execute(input: ExecuteInput) {
const { credentials, log } = input;
if (credentials.apiKey && !credentials.accessToken) {
// Defensive: trim stray surrounding whitespace from a pasted credential.
if (typeof credentials.apiKey === "string") {
credentials.apiKey = credentials.apiKey.trim();
}
// Service Account JSON → mint a short-lived OAuth token (Bearer). An Express-mode API key is
// sent as-is via x-goog-api-key (see buildHeaders), so no token exchange is needed for it.
if (credentials.apiKey && !credentials.accessToken && looksLikeServiceAccountJson(credentials.apiKey)) {
try {
const sa = parseSAFromApiKey(credentials.apiKey);
credentials.accessToken = await getAccessToken(sa);
@@ -123,6 +151,19 @@ export class VertexExecutor extends BaseExecutor {
}
buildUrl(model: string, stream: boolean, urlIndex = 0, credentials: any = null) {
// Vertex AI Express mode: project-less v1 publisher endpoint with the API key passed as a
// ?key= query parameter (verified working contract — same as the CaptionAI GeminiClient). The
// Express key is NOT accepted as a Bearer/OAuth credential or via x-goog-api-key on this API.
if (isExpressApiKey(credentials?.apiKey) && !credentials?.accessToken) {
const expressKey = encodeURIComponent(String(credentials.apiKey).trim());
if (isPartnerModel(model)) {
// Partner (Anthropic/etc.) models are not available via Express keys; best-effort.
return `https://aiplatform.googleapis.com/v1/publishers/openapi/chat/completions?key=${expressKey}`;
}
const op = stream ? "streamGenerateContent?alt=sse&" : "generateContent?";
return `https://aiplatform.googleapis.com/v1/publishers/google/models/${model}:${op}key=${expressKey}`;
}
const region = credentials?.providerSpecificData?.region || "us-central1";
let project = "unknown-project";
@@ -146,6 +187,7 @@ export class VertexExecutor extends BaseExecutor {
if (credentials.accessToken) {
headers["Authorization"] = `Bearer ${credentials.accessToken}`;
}
// Express-mode keys are carried in the ?key= query parameter (see buildUrl), not a header.
if (stream) {
headers["Accept"] = "text/event-stream";
}

47
open-sse/handlers/chatCore.ts
View File

@@ -1986,12 +1986,13 @@ export async function handleChatCore({
// Use credentials.connectionId as a fallback so that requests without an
// explicit session-level connectionId still register in the pendingRequests map.
const pendingConnId = connectionId || credentials?.connectionId || null;
const pendingRequestId = trackPendingRequest(model, provider, pendingConnId, true, {
clientEndpoint: clientRawRequest?.endpoint || "/v1/chat/completions",
clientRequest: clientRawRequest?.body ?? body,
providerRequest: initialProviderRequest,
stage: "registered",
}) || generateRequestId();
const pendingRequestId =
trackPendingRequest(model, provider, pendingConnId, true, {
clientEndpoint: clientRawRequest?.endpoint || "/v1/chat/completions",
clientRequest: clientRawRequest?.body ?? body,
providerRequest: initialProviderRequest,
stage: "registered",
}) || generateRequestId();
// Initialize rate limit settings from persisted DB (once, lazy)
await initializeRateLimits();
@@ -2769,7 +2770,10 @@ export async function handleChatCore({
comboTargetLimits,
});
contextLimit = resolved.limit;
log?.info?.("CONTEXT", `Combo context limit: ${resolved.limit} (source=${resolved.source})`);
log?.info?.(
"CONTEXT",
`Combo context limit: ${resolved.limit} (source=${resolved.source})`
);
} catch (err) {
log?.warn?.("CONTEXT", "Failed to resolve combo limits for compression: " + err);
}
@@ -3104,6 +3108,12 @@ export async function handleChatCore({
translatedBody.messages,
DEFAULT_THINKING_CLAUDE_SIGNATURE
) as typeof translatedBody.messages;
// Anthropic API rejects requests with both temperature and top_p.
// VS Code Claude extension and similar clients send both; strip top_p.
if (translatedBody.temperature !== undefined && translatedBody.top_p !== undefined) {
delete translatedBody.top_p;
}
}
// Fix #2468: always extract role:"system" → top-level system.
@@ -5442,17 +5452,14 @@ export async function handleChatCore({
}
const responseHeaders: Record<string, string> = {
...buildStreamingResponseHeaders(
providerResponse.headers,
{
provider,
model,
cacheHit: false,
latencyMs: 0,
usage: null,
costUsd: 0,
}
),
...buildStreamingResponseHeaders(providerResponse.headers, {
provider,
model,
cacheHit: false,
latencyMs: 0,
usage: null,
costUsd: 0,
}),
"x-omniroute-request-id": pendingRequestId,
};
@@ -5560,7 +5567,9 @@ export async function handleChatCore({
});
} catch (e) {
// Best-effort — don't break the stream completion path if this fails
try { console.warn("finalizeMostRecentPendingRequest failed:", e && (e.message || e)); } catch {}
try {
console.warn("finalizeMostRecentPendingRequest failed:", e && (e.message || e));
} catch {}
}
if (apiKeyInfo?.id && streamUsage) {

2
open-sse/translator/request/openai-to-claude.ts
View File

@@ -212,7 +212,7 @@ export function openaiToClaudeRequest(model, body, stream) {
if (body.temperature !== undefined) {
result.temperature = body.temperature;
}
if (body.top_p !== undefined) {
if (body.temperature === undefined && body.top_p !== undefined) {
result.top_p = body.top_p;
}
if (body.stop !== undefined) {

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

@@ -109,7 +109,11 @@ function normalizeResponsesSseIds(payload: JsonRecord): boolean {
}
}
if (payload.response && typeof payload.response === "object" && !Array.isArray(payload.response)) {
if (
payload.response &&
typeof payload.response === "object" &&
!Array.isArray(payload.response)
) {
const response = payload.response as JsonRecord;
let responseChanged = false;
const normalizedResponse = { ...response };
@@ -1016,6 +1020,32 @@ export function createSSEStream(options: StreamOptions = {}) {
}
};
const emitClaudeEmptyStreamErrorAndAbort = (
controller: TransformStreamDefaultController,
decrementPendingRequest = true
) => {
clearIdleTimer();
const msg = "Claude returned an empty response (no content block)";
console.warn(
`[STREAM] Empty Claude stream at flush - emitting error (${provider || "provider"}:${model || "unknown"})`
);
const errorBody = buildErrorBody(502, msg);
const errorEvent: Record<string, unknown> = { type: "error", error: errorBody.error };
const errOutput = formatSSE(errorEvent, FORMATS.CLAUDE);
reqLogger?.appendConvertedChunk?.(errOutput);
clientPayloadCollector.push(errorEvent);
controller.enqueue(encoder.encode(errOutput));
if (onFailure) {
try {
void onFailure({ status: 502, message: msg, code: "empty_response" });
} catch {}
}
if (decrementPendingRequest) {
trackPendingRequest(model, provider, connectionId, false);
}
controller.error(markPendingRequestCleared(new Error(msg)));
};
const emitTranslatedClientItem = (
controller: TransformStreamDefaultController,
item: Record<string, unknown>
@@ -1059,13 +1089,8 @@ export function createSSEStream(options: StreamOptions = {}) {
sourceFormat === FORMATS.CLAUDE &&
shouldInjectClaudeEmptyResponseBeforeCurrentEvent(claudeEmptyResponseLifecycle, itemSanitized)
) {
const eventType = getClaudeEventType(itemSanitized);
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: true,
includeMessageDelta:
eventType === "message_stop" && !claudeEmptyResponseLifecycle.hasMessageDelta,
includeMessageStop: false,
});
emitClaudeEmptyStreamErrorAndAbort(controller);
return;
}
if (sourceFormat === FORMATS.CLAUDE && isClaudeEventPayload(itemSanitized)) {
@@ -1300,12 +1325,8 @@ export function createSSEStream(options: StreamOptions = {}) {
type: eventType,
})
) {
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: true,
includeMessageDelta:
eventType === "message_stop" && !claudeEmptyResponseLifecycle.hasMessageDelta,
includeMessageStop: false,
});
emitClaudeEmptyStreamErrorAndAbort(controller);
return;
}
pendingPassthroughEventLine = line;
@@ -1337,7 +1358,8 @@ export function createSSEStream(options: StreamOptions = {}) {
// clients like OpenCode, so drop it only for Responses-native consumers.
const hasActiveDeltaValue = (value: unknown): boolean => {
if (typeof value === "string") return value.length > 0;
if (Array.isArray(value)) return value.some((entry) => hasActiveDeltaValue(entry));
if (Array.isArray(value))
return value.some((entry) => hasActiveDeltaValue(entry));
if (value && typeof value === "object") {
return Object.values(value).some((entry) => hasActiveDeltaValue(entry));
}
@@ -1605,7 +1627,12 @@ export function createSSEStream(options: StreamOptions = {}) {
parsed,
passthroughResponsesOutputItems
);
if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) {
if (
stripped ||
backfilled ||
textualToolCallBackfilled ||
responsesIdsNormalized
) {
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
}
@@ -1632,13 +1659,8 @@ export function createSSEStream(options: StreamOptions = {}) {
parsed
)
) {
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: true,
includeMessageDelta:
parsed.type === "message_stop" &&
!claudeEmptyResponseLifecycle.hasMessageDelta,
includeMessageStop: false,
});
emitClaudeEmptyStreamErrorAndAbort(controller);
return;
}
updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, parsed);
const restoredToolName = restoreClaudePassthroughToolUseName(parsed, toolNameMap);
@@ -1708,14 +1730,16 @@ export function createSSEStream(options: StreamOptions = {}) {
!parsed.choices[0].delta.reasoning_content
);
const hadNonStringToolCallId = Array.isArray(parsed.choices)
? parsed.choices.some((choice) =>
Array.isArray(choice?.delta?.tool_calls) &&
choice.delta.tool_calls.some(
(tc) => tc?.id != null && typeof tc.id !== "string"
)
? parsed.choices.some(
(choice) =>
Array.isArray(choice?.delta?.tool_calls) &&
choice.delta.tool_calls.some(
(tc) => tc?.id != null && typeof tc.id !== "string"
)
)
: false;
const hadNonStringTopLevelId = parsed?.id != null && typeof parsed.id !== "string";
const hadNonStringTopLevelId =
parsed?.id != null && typeof parsed.id !== "string";
parsed = sanitizeStreamingChunk(parsed);
if (
@@ -2148,13 +2172,8 @@ export function createSSEStream(options: StreamOptions = {}) {
bufferedPayload
)
) {
const eventType = getClaudeEventType(bufferedPayload);
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: true,
includeMessageDelta:
eventType === "message_stop" && !claudeEmptyResponseLifecycle.hasMessageDelta,
includeMessageStop: false,
});
emitClaudeEmptyStreamErrorAndAbort(controller, false);
return;
}
if (isClaudeEventPayload(bufferedPayload)) {
updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, bufferedPayload);
@@ -2164,7 +2183,8 @@ export function createSSEStream(options: StreamOptions = {}) {
// Normalize numeric IDs for final buffered data: chunk (same as transform path)
if (typeof bufferedPayload === "object" && !Array.isArray(bufferedPayload)) {
const flushedParsed = bufferedPayload as JsonRecord;
const flushedType = typeof flushedParsed.type === "string" ? flushedParsed.type : "";
const flushedType =
typeof flushedParsed.type === "string" ? flushedParsed.type : "";
const isResponses = flushedType.startsWith("response.");
const isClaude = isClaudeEventPayload(flushedParsed);
if (isResponses) {
@@ -2181,7 +2201,9 @@ export function createSSEStream(options: StreamOptions = {}) {
}
if (Array.isArray(flushedParsed.choices)) {
for (const choice of flushedParsed.choices as JsonRecord[]) {
const tcs = (choice as JsonRecord | undefined)?.delta as JsonRecord | undefined;
const tcs = (choice as JsonRecord | undefined)?.delta as
| JsonRecord
| undefined;
if (Array.isArray(tcs?.tool_calls)) {
for (const tc of tcs.tool_calls as JsonRecord[]) {
if (tc?.id != null && typeof tc.id !== "string") {
@@ -2208,11 +2230,8 @@ export function createSSEStream(options: StreamOptions = {}) {
}
if (shouldInjectClaudeEmptyResponseOnFlush(claudeEmptyResponseLifecycle)) {
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: true,
includeMessageDelta: !claudeEmptyResponseLifecycle.hasMessageDelta,
includeMessageStop: !claudeEmptyResponseLifecycle.hasMessageStop,
});
emitClaudeEmptyStreamErrorAndAbort(controller, false);
return;
} else if (shouldInjectClaudeMissingFinalizersOnFlush(claudeEmptyResponseLifecycle)) {
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: false,
@@ -2489,11 +2508,8 @@ export function createSSEStream(options: StreamOptions = {}) {
if (sourceFormat === FORMATS.CLAUDE) {
if (shouldInjectClaudeEmptyResponseOnFlush(claudeEmptyResponseLifecycle)) {
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: true,
includeMessageDelta: !claudeEmptyResponseLifecycle.hasMessageDelta,
includeMessageStop: !claudeEmptyResponseLifecycle.hasMessageStop,
});
emitClaudeEmptyStreamErrorAndAbort(controller, false);
return;
} else if (shouldInjectClaudeMissingFinalizersOnFlush(claudeEmptyResponseLifecycle)) {
emitSyntheticClaudeEmptyResponse(controller, {
includeContentBlock: false,
@@ -2631,7 +2647,7 @@ export function createSSEStream(options: StreamOptions = {}) {
);
}
export default createSSEStream
export default createSSEStream;
// Convenience functions for backward compatibility
export function createSSETransformStreamWithLogger(

12
src/lib/providers/validation.ts
View File

@@ -3954,8 +3954,13 @@ export async function validateProviderApiKey({ provider, apiKey, providerSpecifi
},
vertex: async ({ apiKey }: any) => {
try {
const { parseSAFromApiKey, getAccessToken } =
const { parseSAFromApiKey, getAccessToken, isExpressApiKey } =
await import("@omniroute/open-sse/executors/vertex.ts");
// Express-mode API keys are opaque strings sent directly as the ?key= query param — there is
// no JWT to mint, so accept any non-empty Express key (the live chat/media call validates it).
if (isExpressApiKey(apiKey)) {
return { valid: true, error: null };
}
const sa = parseSAFromApiKey(apiKey);
// Validates credentials by successfully successfully exchanging them for a JWT from Google Identity
await getAccessToken(sa);
@@ -3966,8 +3971,11 @@ export async function validateProviderApiKey({ provider, apiKey, providerSpecifi
},
"vertex-partner": async ({ apiKey }: any) => {
try {
const { parseSAFromApiKey, getAccessToken } =
const { parseSAFromApiKey, getAccessToken, isExpressApiKey } =
await import("@omniroute/open-sse/executors/vertex.ts");
if (isExpressApiKey(apiKey)) {
return { valid: true, error: null };
}
const sa = parseSAFromApiKey(apiKey);
await getAccessToken(sa);
return { valid: true, error: null };

278
tests/unit/claude-empty-stream-error-3685.test.ts Normal file
View File

@@ -0,0 +1,278 @@
import test from "node:test";
import assert from "node:assert/strict";
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
// #3685 — When a Claude stream completes with lifecycle events (message_start /
// message_delta / message_stop) but zero content_block events, the router was
// injecting a synthetic success message instead of failing over. Fix: emit a
// real SSE error event and call controller.error() so the combo layer can retry.
const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-3685-"));
process.env.DATA_DIR = TEST_DATA_DIR;
const core = await import("../../src/lib/db/core.ts");
const { createSSEStream } = await import("../../open-sse/utils/stream.ts");
const { FORMATS } = await import("../../open-sse/translator/formats.ts");
const { getPendingRequests, clearPendingRequests } =
await import("../../src/lib/usage/usageHistory.ts");
const enc = new TextEncoder();
async function readTransformed(chunks: string[], options: Record<string, unknown>) {
const source = new ReadableStream<Uint8Array>({
start(c) {
for (const chunk of chunks) c.enqueue(enc.encode(chunk));
c.close();
},
});
return new Response(source.pipeThrough(createSSEStream(options as any))).text();
}
test.after(() => {
core.resetDbInstance();
if (fs.existsSync(TEST_DATA_DIR)) {
for (const entry of fs.readdirSync(TEST_DATA_DIR)) {
fs.rmSync(path.join(TEST_DATA_DIR, entry), { recursive: true, force: true });
}
}
});
// --- Golden path: empty content-block stream (the bug case) should now error ---
test("#3685 passthrough: empty Claude SSE (no content_block) rejects the stream", async () => {
let failurePayload: Record<string, unknown> | null = null;
await assert.rejects(
readTransformed(
[
`event: message_start\ndata: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_3685",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
content: [],
stop_reason: null,
stop_sequence: null,
usage: { input_tokens: 5, output_tokens: 0 },
},
})}\n\n`,
`event: message_delta\ndata: ${JSON.stringify({
type: "message_delta",
delta: { stop_reason: "content_filter", stop_sequence: null },
usage: { output_tokens: 1 },
})}\n\n`,
`event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.CLAUDE,
provider: "anthropic",
model: "claude-sonnet-4-6",
body: { messages: [{ role: "user", content: "hello" }] },
onFailure(p: Record<string, unknown>) {
failurePayload = p;
},
}
),
/empty response/i,
"stream should reject with empty-response error"
);
assert.ok(failurePayload, "onFailure callback must be invoked");
assert.equal((failurePayload as any).status, 502);
assert.match((failurePayload as any).message as string, /empty response/i);
});
test("#3685 passthrough: empty Claude SSE emits event: error SSE line before aborting", async () => {
const collected: string[] = [];
const source = new ReadableStream<Uint8Array>({
start(c) {
const chunks = [
`event: message_start\ndata: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_3685b",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
content: [],
stop_reason: null,
usage: { input_tokens: 5, output_tokens: 0 },
},
})}\n\n`,
`event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`,
];
for (const chunk of chunks) c.enqueue(enc.encode(chunk));
c.close();
},
});
const transformed = source.pipeThrough(
createSSEStream({
mode: "passthrough",
sourceFormat: FORMATS.CLAUDE,
provider: "anthropic",
model: "claude-sonnet-4-6",
body: { messages: [{ role: "user", content: "hello" }] },
} as any)
);
const reader = transformed.getReader();
const dec = new TextDecoder();
let gotError = false;
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
collected.push(dec.decode(value));
}
} catch {
gotError = true;
}
assert.ok(gotError, "stream reader should throw on error");
const full = collected.join("");
assert.match(full, /event: error/, "SSE error event must be emitted before abort");
assert.doesNotMatch(
full,
/event: content_block_start/,
"no synthetic content_block must be emitted"
);
});
// --- Regression guards: excluded cases must NOT be turned into errors ---
test("#3685 regression: stream with content_block events is NOT turned into an error", async () => {
// A max_tokens:1 ping returns exactly 1 token → content_block events exist.
// hasContentBlock = true → shouldInjectClaudeEmptyResponseOnFlush = false → no error.
const text = await readTransformed(
[
`event: message_start\ndata: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_ping",
type: "message",
role: "assistant",
model: "claude-haiku-4-5",
content: [],
stop_reason: null,
usage: { input_tokens: 3, output_tokens: 0 },
},
})}\n\n`,
`event: content_block_start\ndata: ${JSON.stringify({
type: "content_block_start",
index: 0,
content_block: { type: "text", text: "" },
})}\n\n`,
`event: content_block_delta\ndata: ${JSON.stringify({
type: "content_block_delta",
index: 0,
delta: { type: "text_delta", text: "Hi" },
})}\n\n`,
`event: content_block_stop\ndata: ${JSON.stringify({
type: "content_block_stop",
index: 0,
})}\n\n`,
`event: message_delta\ndata: ${JSON.stringify({
type: "message_delta",
delta: { stop_reason: "max_tokens", stop_sequence: null },
usage: { output_tokens: 1 },
})}\n\n`,
`event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.CLAUDE,
provider: "anthropic",
model: "claude-haiku-4-5",
body: { messages: [{ role: "user", content: "ping" }] },
}
);
assert.match(text, /Hi/, "content must pass through untouched");
assert.doesNotMatch(text, /event: error/, "must NOT emit an error event");
});
test("#3685 pending request counter is decremented when empty-stream error fires", async () => {
// Regression guard for the bug caught by Cursor/Codex: emitClaudeEmptyStreamErrorAndAbort
// was marking the error with PENDING_REQUEST_CLEARED_MARKER but never calling
// trackPendingRequest(..., false). streamHandler.clearPendingRequest() trusts the marker
// and skips its own decrement, leaving the counter permanently inflated.
clearPendingRequests();
const { trackPendingRequest } = await import("../../src/lib/usage/usageHistory.ts");
// Simulate the stream engine incrementing the counter at request start.
trackPendingRequest("claude-sonnet-4-6", "anthropic", "conn-test", true);
assert.equal(
getPendingRequests().byModel["claude-sonnet-4-6 (anthropic)"],
1,
"pending count should start at 1 after request begins"
);
await assert.rejects(
readTransformed(
[
`event: message_start\ndata: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_pending_test",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
content: [],
stop_reason: null,
usage: { input_tokens: 3, output_tokens: 0 },
},
})}\n\n`,
`event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.CLAUDE,
provider: "anthropic",
model: "claude-sonnet-4-6",
connectionId: "conn-test",
body: { messages: [{ role: "user", content: "hello" }] },
}
),
/empty response/i
);
// emitClaudeEmptyStreamErrorAndAbort must call trackPendingRequest(..., false) so the
// counter is back to 0 after the stream terminates.
assert.equal(
getPendingRequests().byModel["claude-sonnet-4-6 (anthropic)"],
0,
"pending count must be 0 after empty-stream error — not left inflated"
);
});
test("#3685 regression: upstream error event sets hasError=true and does NOT trigger empty-stream path", () => {
// If Claude itself emits type:error, lifecycle.hasError=true.
// shouldInjectClaudeEmptyResponseBeforeCurrentEvent / shouldInjectClaudeEmptyResponseOnFlush
// both check !lifecycle.hasError first — so neither our new error path nor the old synthetic
// path is triggered. Verified by inspecting the guard functions directly.
const lifecycle = {
hasMessageStart: true,
hasContentBlock: false,
hasMessageDelta: false,
hasMessageStop: false,
hasError: false,
syntheticContentInjected: false,
warningLogged: false,
};
// Simulate receiving an error event: sets hasError = true.
const lifecycleWithError = { ...lifecycle, hasError: true };
// shouldInjectClaudeEmptyResponseOnFlush equivalent: hasError blocks it
const wouldInjectOnFlush =
!lifecycleWithError.hasError &&
!lifecycleWithError.hasContentBlock &&
(lifecycleWithError.hasMessageStart ||
lifecycleWithError.hasMessageDelta ||
lifecycleWithError.hasMessageStop);
assert.equal(
wouldInjectOnFlush,
false,
"hasError=true must prevent the empty-stream error path from firing"
);
});

31
tests/unit/executor-vertex-extended.test.ts
View File

@@ -51,17 +51,22 @@ test("VertexExecutor.buildUrl defaults to us-central1 and unknown-project when p
apiKey: createServiceAccountJson({ includeProjectId: false }),
providerSpecificData: {},
});
const invalidJson = executor.buildUrl("gemini-2.5-flash", false, 0, {
apiKey: "not-json",
});
assert.equal(
missingProject,
"https://aiplatform.googleapis.com/v1/projects/unknown-project/locations/us-central1/publishers/google/models/gemini-2.5-flash:generateContent"
);
});
test("VertexExecutor.buildUrl routes a non-JSON Express API key to the project-less publisher endpoint", () => {
const executor = new VertexExecutor();
const expressUrl = executor.buildUrl("gemini-2.5-flash", false, 0, {
apiKey: "express-key-abc",
});
assert.equal(
invalidJson,
"https://aiplatform.googleapis.com/v1/projects/unknown-project/locations/us-central1/publishers/google/models/gemini-2.5-flash:generateContent"
expressUrl,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-2.5-flash:generateContent?key=express-key-abc"
);
});
@@ -191,20 +196,12 @@ test("VertexExecutor.execute skips Service Account parsing when accessToken is a
}
});
test("VertexExecutor.execute rejects invalid or incomplete Service Account JSON clearly", async () => {
test("VertexExecutor.execute rejects incomplete Service Account JSON clearly", async () => {
const executor = new VertexExecutor();
await assert.rejects(
() =>
executor.execute({
model: "gemini-2.5-flash",
body: { contents: [] },
stream: false,
credentials: { apiKey: "not-json" },
}),
/Service Account JSON/
);
// A JSON object missing client_email/private_key is treated as a Service Account (not an Express
// key) and must fail clearly when minting a JWT. A non-JSON string is an Express key (covered
// elsewhere) and is intentionally NOT rejected here.
await assert.rejects(
() =>
executor.execute({

174
tests/unit/stream-utils.test.ts
View File

@@ -725,7 +725,11 @@ test("createSSEStream passthrough drops leaked empty chat bootstrap chunks for R
created: 1,
model: "gpt-5.4",
choices: [
{ index: 0, delta: { role: "assistant", content: null, refusal: null }, finish_reason: null },
{
index: 0,
delta: { role: "assistant", content: null, refusal: null },
finish_reason: null,
},
],
})}\n\n`,
`event: response.created\ndata: ${JSON.stringify({
@@ -1047,50 +1051,46 @@ test("createSSEStream passthrough merges Claude usage chunks and restores mapped
assert.equal(onCompletePayload.responseBody.usage.total_tokens, 10);
});
test("createSSEStream passthrough injects a synthetic Claude text block for empty assistant SSE", async () => {
let onCompletePayload = null;
const text = await readTransformed(
[
`event: message_start\ndata: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_empty_passthrough",
type: "message",
role: "assistant",
model: "claude-sonnet-4",
content: [],
stop_reason: null,
stop_sequence: null,
usage: { input_tokens: 7, output_tokens: 0 },
test("#3685 createSSEStream passthrough emits SSE error (not synthetic text) for empty Claude assistant SSE", async () => {
let failurePayload = null;
await assert.rejects(
readTransformed(
[
`event: message_start\ndata: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_empty_passthrough",
type: "message",
role: "assistant",
model: "claude-sonnet-4",
content: [],
stop_reason: null,
stop_sequence: null,
usage: { input_tokens: 7, output_tokens: 0 },
},
})}\n\n`,
`event: message_stop\ndata: ${JSON.stringify({
type: "message_stop",
})}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.CLAUDE,
provider: "claude",
model: "claude-sonnet-4",
body: {
messages: [{ role: "user", content: "hello" }],
},
})}\n\n`,
`event: message_stop\ndata: ${JSON.stringify({
type: "message_stop",
})}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.CLAUDE,
provider: "claude",
model: "claude-sonnet-4",
body: {
messages: [{ role: "user", content: "hello" }],
},
onComplete(payload) {
onCompletePayload = payload;
},
}
onFailure(payload) {
failurePayload = payload;
},
}
),
/empty response/i
);
assert.equal((text.match(/event: message_start/g) || []).length, 1);
assert.equal((text.match(/event: message_delta/g) || []).length, 1);
assert.match(text, /event: content_block_start/);
assert.match(text, /event: content_block_delta/);
assert.match(text, /event: message_stop/);
assert.ok(text.indexOf("event: content_block_start") > text.indexOf("event: message_start"));
assert.ok(text.indexOf("event: message_stop") > text.indexOf("event: content_block_stop"));
// SYNTHETIC_CLAUDE_EMPTY_RESPONSE_TEXT is "" so the accumulator produces null content (empty delta is falsy).
assert.equal(onCompletePayload.responseBody.choices[0].message.content, null);
assert.ok(failurePayload, "onFailure should be called");
assert.equal(failurePayload.status, 502);
assert.match(failurePayload.message, /empty response/i);
});
test("createSSEStream passthrough does not emit [DONE] for Claude SSE clients", async () => {
@@ -1149,51 +1149,46 @@ test("createSSEStream passthrough does not emit [DONE] for Claude SSE clients",
assert.doesNotMatch(text, /\[DONE\]/);
});
test("createSSEStream translate mode injects a synthetic Claude text block when OpenAI finishes empty", async () => {
let onCompletePayload = null;
const text = await readTransformed(
[
`data: ${JSON.stringify({
id: "chatcmpl_empty_1",
object: "chat.completion.chunk",
created: 1,
test("#3685 createSSEStream translate mode emits SSE error (not synthetic text) when OpenAI upstream finishes empty for Claude client", async () => {
let failurePayload = null;
await assert.rejects(
readTransformed(
[
`data: ${JSON.stringify({
id: "chatcmpl_empty_1",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: { role: "assistant" } }],
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl_empty_1",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
usage: { prompt_tokens: 3, completion_tokens: 0, total_tokens: 3 },
})}\n\n`,
],
{
mode: "translate",
targetFormat: FORMATS.OPENAI,
sourceFormat: FORMATS.CLAUDE,
provider: "openai",
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: { role: "assistant" } }],
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl_empty_1",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
usage: { prompt_tokens: 3, completion_tokens: 0, total_tokens: 3 },
})}\n\n`,
],
{
mode: "translate",
targetFormat: FORMATS.OPENAI,
sourceFormat: FORMATS.CLAUDE,
provider: "openai",
model: "gpt-4.1-mini",
body: {
messages: [{ role: "user", content: "hello" }],
},
onComplete(payload) {
onCompletePayload = payload;
},
}
body: {
messages: [{ role: "user", content: "hello" }],
},
onFailure(payload) {
failurePayload = payload;
},
}
),
/empty response/i
);
assert.equal((text.match(/event: message_start/g) || []).length, 1);
assert.match(text, /event: content_block_start/);
assert.match(text, /event: content_block_delta/);
assert.match(text, /event: message_delta/);
assert.match(text, /event: message_stop/);
assert.ok(text.indexOf("event: content_block_start") > text.indexOf("event: message_start"));
assert.ok(text.indexOf("event: message_delta") > text.indexOf("event: content_block_stop"));
// SYNTHETIC_CLAUDE_EMPTY_RESPONSE_TEXT is "" so the accumulator produces null content (empty delta is falsy).
assert.equal(onCompletePayload.responseBody.choices[0].message.content, null);
assert.equal(onCompletePayload.responseBody.usage.total_tokens, 3);
assert.ok(failurePayload, "onFailure should be called");
assert.equal(failurePayload.status, 502);
assert.match(failurePayload.message, /empty response/i);
});
test("createSSETransformStreamWithLogger flushes a trailing Claude usage event without a newline", async () => {
@@ -1741,7 +1736,7 @@ test("createSSEStream passthrough logs empty response after tool_calls completio
index: 0,
id: "call_tc",
type: "function",
function: { name: "task_complete", arguments: '{}' },
function: { name: "task_complete", arguments: "{}" },
},
],
},
@@ -1771,7 +1766,10 @@ test("createSSEStream passthrough logs empty response after tool_calls completio
assert.match(text, /"finish_reason":"tool_calls"/);
assert.equal(onCompletePayload.status, 200);
assert.equal(onCompletePayload.responseBody.choices[0].finish_reason, "tool_calls");
assert.equal(onCompletePayload.responseBody.choices[0].message.tool_calls[0].function.name, "task_complete");
assert.equal(
onCompletePayload.responseBody.choices[0].message.tool_calls[0].function.name,
"task_complete"
);
// Content should be null (empty) since no text was generated
assert.equal(onCompletePayload.responseBody.choices[0].message.content, null);
});

23
tests/unit/t29-vertex-sa-json-executor.test.ts
View File

@@ -55,17 +55,34 @@ test("T29: Vertex executor headers include Bearer token and SSE Accept when stre
assert.equal(headers.Accept, "text/event-stream");
});
test("T29: Vertex executor rejects invalid Service Account JSON clearly", async () => {
test("T29: Vertex executor rejects incomplete Service Account JSON clearly", async () => {
const executor = new VertexExecutor();
// A JSON object (not an opaque Express key) that is missing client_email/private_key must still
// fail clearly when the executor tries to mint a JWT from it.
await assert.rejects(
() =>
executor.execute({
model: "gemini-2.5-flash",
body: { contents: [] },
stream: false,
credentials: { apiKey: "not-json" },
credentials: { apiKey: JSON.stringify({ project_id: "p" }) },
}),
/Service Account JSON/i
/missing required fields/i
);
});
test("T29: Vertex executor routes a non-JSON Express API key to the project-less publisher endpoint", () => {
const executor = new VertexExecutor();
const stream = executor.buildUrl("gemini-2.5-flash", true, 0, { apiKey: "express-key-123" });
const nonStream = executor.buildUrl("gemini-2.5-flash", false, 0, { apiKey: "express-key-123" });
assert.equal(
stream,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-2.5-flash:streamGenerateContent?alt=sse&key=express-key-123"
);
assert.equal(
nonStream,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-2.5-flash:generateContent?key=express-key-123"
);
});

18
tests/unit/translator-openai-to-claude.test.ts
View File

@@ -84,7 +84,8 @@ test("OpenAI -> Claude maps system messages, parameters and assistant cache mark
assert.equal(result.stream, true);
assert.equal(result.max_tokens, 33);
assert.equal(result.temperature, 0.25);
assert.equal(result.top_p, 0.8);
// top_p is stripped when temperature is also present (Anthropic rejects both).
assert.equal(result.top_p, undefined);
assert.deepEqual(result.stop_sequences, ["DONE"]);
assert.equal(result.system[0].text, "Rule A\nRule B\nRule C");
assert.equal(result.messages[0].role, "user");
@@ -94,6 +95,21 @@ test("OpenAI -> Claude maps system messages, parameters and assistant cache mark
assert.deepEqual(result.messages[1].content[0].cache_control, { type: "ephemeral" });
});
test("OpenAI -> Claude strips top_p when temperature is also present", () => {
const result = openaiToClaudeRequest(
"claude-4-sonnet",
{
messages: [{ role: "user", content: "Hello" }],
temperature: 0.25,
top_p: 0.8,
},
false
);
assert.equal(result.temperature, 0.25);
assert.equal(result.top_p, undefined);
});
test("OpenAI -> Claude converts multimodal content, tool declarations, tool calls and tool results", () => {
const result = openaiToClaudeRequest(
"claude-4-sonnet",

102
tests/unit/vertex-express-apikey.test.ts Normal file
View File

@@ -0,0 +1,102 @@
import test from "node:test";
import assert from "node:assert/strict";
const { VertexExecutor, isExpressApiKey, looksLikeServiceAccountJson } = await import(
"../../open-sse/executors/vertex.ts"
);
test("looksLikeServiceAccountJson is true only for a JSON object credential", () => {
assert.equal(looksLikeServiceAccountJson(JSON.stringify({ project_id: "p" })), true);
assert.equal(looksLikeServiceAccountJson("express-opaque-key"), false);
assert.equal(looksLikeServiceAccountJson(JSON.stringify([1, 2, 3])), false);
assert.equal(looksLikeServiceAccountJson(""), false);
});
test("isExpressApiKey is true for a non-empty, non-JSON credential", () => {
assert.equal(isExpressApiKey("AIzaSyExpressKey"), true);
assert.equal(isExpressApiKey(" "), false);
assert.equal(isExpressApiKey(""), false);
assert.equal(isExpressApiKey(null), false);
assert.equal(isExpressApiKey(undefined), false);
assert.equal(isExpressApiKey(JSON.stringify({ project_id: "p" })), false);
});
test("buildUrl Express: streaming google model uses streamGenerateContent + ?alt=sse&key=", () => {
const executor = new VertexExecutor();
const url = executor.buildUrl("gemini-3-flash-preview", true, 0, { apiKey: "k-express" });
assert.equal(
url,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-3-flash-preview:streamGenerateContent?alt=sse&key=k-express"
);
});
test("buildUrl Express: non-streaming google model uses generateContent?key=", () => {
const executor = new VertexExecutor();
const url = executor.buildUrl("gemini-3-flash-preview", false, 0, { apiKey: "k-express" });
assert.equal(
url,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-3-flash-preview:generateContent?key=k-express"
);
});
test("buildUrl Express: the API key is URL-encoded and trimmed", () => {
const executor = new VertexExecutor();
const url = executor.buildUrl("gemini-2.5-flash", false, 0, { apiKey: " a/b+c=d " });
assert.equal(
url,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-2.5-flash:generateContent?key=a%2Fb%2Bc%3Dd"
);
});
test("buildUrl Express: a present accessToken takes the Service Account path, not Express", () => {
const executor = new VertexExecutor();
const url = executor.buildUrl("gemini-2.5-flash", false, 0, {
apiKey: "k-express",
accessToken: "ya29.token",
providerSpecificData: { region: "us-central1" },
});
assert.equal(
url,
"https://aiplatform.googleapis.com/v1/projects/unknown-project/locations/us-central1/publishers/google/models/gemini-2.5-flash:generateContent"
);
});
test("buildHeaders for an Express key (no accessToken) omits the Authorization header", () => {
const executor = new VertexExecutor();
const headers = executor.buildHeaders({ apiKey: "k-express" }, false);
assert.equal(headers["Content-Type"], "application/json");
assert.equal(headers.Authorization, undefined);
});
test("execute with an Express key calls the publisher endpoint directly (no OAuth token exchange)", async () => {
const executor = new VertexExecutor();
const originalFetch = globalThis.fetch;
const calls: string[] = [];
globalThis.fetch = async (url: any) => {
calls.push(String(url));
return new Response(JSON.stringify({ ok: true }), {
status: 200,
headers: { "Content-Type": "application/json" },
});
};
try {
const result = await executor.execute({
model: "gemini-3-flash-preview",
body: { contents: [{ role: "user", parts: [{ text: "hi" }] }] },
stream: false,
credentials: { apiKey: "AIzaSyExpressKey" },
} as any);
assert.equal(result.response.status, 200);
// Exactly one call — straight to Vertex, no oauth2 token mint.
assert.equal(calls.length, 1);
assert.equal(
calls[0],
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-3-flash-preview:generateContent?key=AIzaSyExpressKey"
);
} finally {
globalThis.fetch = originalFetch;
}
});