mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-21 14:42:20 +03:00
Compare commits
1 Commits
fix/10727-
...
fix/8450-r
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a30c28abb6 |
16
.env.example
16
.env.example
@@ -1361,14 +1361,6 @@ CURSOR_USER_AGENT="Cursor/3.4"
|
||||
# FETCH_BODY_TIMEOUT_MS=600000 # Time to receive full response body
|
||||
# FETCH_CONNECT_TIMEOUT_MS=30000 # TCP connection establishment (default: 30s)
|
||||
# FETCH_KEEPALIVE_TIMEOUT_MS=4000 # Keep-alive socket idle timeout (default: 4s)
|
||||
# OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS=30000 # Bounded response-start window per direct
|
||||
# # (no-proxy) attempt (#10214). A silently-dropped
|
||||
# # pooled keep-alive socket surfaces no transport
|
||||
# # error, so without this bound a direct request can
|
||||
# # stall until undici's headersTimeout (600s) or the
|
||||
# # caller's deadline; on expiry the request retries
|
||||
# # once on a fresh no-keep-alive socket. 0 disables
|
||||
# # the bound (default: 30000 = 30s).
|
||||
|
||||
# Default timeout (ms) for src/shared/utils/fetchTimeout.ts. Acts as the
|
||||
# fallback when FETCH_TIMEOUT_MS is unset. Default: 120000 (2 min).
|
||||
@@ -1423,14 +1415,6 @@ CURSOR_USER_AGENT="Cursor/3.4"
|
||||
# OMNIROUTE_PPLX_TLS_TIMEOUT_MS=30000
|
||||
# OMNIROUTE_PPLX_TLS_GRACE_MS=10000
|
||||
|
||||
# ── Perplexity web: built-in-search hint ──
|
||||
# Used by: open-sse/executors/perplexity-web/protocol.ts — appends "You have
|
||||
# built-in web search. Answer questions directly using search results." to the
|
||||
# caller's system message. Off by default: Perplexity's answer engine searches
|
||||
# anyway, and for coding clients the sentence leaks into replies as
|
||||
# meta-commentary. Set to 1/true/yes/on to restore the old behavior.
|
||||
# OMNIROUTE_PPLX_SEARCH_HINT=0
|
||||
|
||||
# ── Grok web TLS sidecar (Chrome-fingerprinted client) ──
|
||||
# Used by: open-sse/services/grokTlsClient.ts — wire-level timeout for the
|
||||
# bogdanfinn/tls-client koffi binding and the JS-side grace window layered on
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
- **fix(network):** direct (no-proxy) egress now bounds each attempt's response-start window (default 30s, `OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`) and retries once on a fresh no-keep-alive socket, so a silently-dropped pooled keep-alive connection can no longer stall direct providers (opencode-go, command-code) until a service restart ([#10214](https://github.com/diegosouzapw/OmniRoute/issues/10214))
|
||||
@@ -1 +0,0 @@
|
||||
- **fix(executors):** the Meta AI (muse-spark-web) WebSocket send-message timeout now reports the socket's `readyState` at the moment it fires, so a "Meta AI WS timed out" failure can be told apart as either the connection never opening (`readyState=0`) or opening successfully and then going silent (`readyState=1`) — the exact ambiguity that made #10727 undiagnosable from logs alone (#10727).
|
||||
@@ -1 +0,0 @@
|
||||
- **fix(perplexity-web):** make the built-in-search hint appended to every system message opt-in via `OMNIROUTE_PPLX_SEARCH_HINT` (off by default) — Perplexity's answer engine searches anyway, and the hint leaked into replies as meta-commentary for coding clients ([#10902](https://github.com/diegosouzapw/OmniRoute/pull/10902), extracted from [#8634](https://github.com/diegosouzapw/OmniRoute/pull/8634)) — thanks @danscMax
|
||||
@@ -1 +0,0 @@
|
||||
- **fix(translator):** merge consecutive same-role contents in direct Claude to Gemini request translation to prevent upstream HTTP 400 errors
|
||||
@@ -732,7 +732,6 @@ REQUEST_TIMEOUT_MS (global override)
|
||||
| `OMNIROUTE_AGENT_GOAL_STREAM_RECOVERY` | `true` | Enable early stream recovery automatically for detected `/goal` agent runs. Set `false`/`0`/`off` to disable the goal-specific opt-in. This can only ADD recovery on top of the operator default — it never overrides an explicit `STREAM_RECOVERY_ENABLED`/DB settings opt-out. |
|
||||
| `OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS` | _(off)_ | Strip non-standard `codex.*` SSE events (e.g. `codex.rate_limits`) that break the OpenAI SDK's `responses.stream()` with a 502. Set `true`/`1`/`yes` to enable. |
|
||||
| `FETCH_HEADERS_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive response headers. |
|
||||
| `OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS` | `30000` (30s) | Maximum response-start wait (ms) for each direct no-proxy attempt. A timeout retries once on a fresh socket; set `0` to disable the bound and retain the previous behavior. |
|
||||
| `FETCH_BODY_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive the full response body. |
|
||||
| `FETCH_CONNECT_TIMEOUT_MS` | `30000` | TCP connection establishment timeout. |
|
||||
| `FETCH_KEEPALIVE_TIMEOUT_MS` | `4000` | Keep-alive socket idle timeout. |
|
||||
@@ -756,7 +755,6 @@ REQUEST_TIMEOUT_MS (global override)
|
||||
| `OMNIROUTE_CLAUDE_TLS_GRACE_MS` | `10000` | JS-side grace added on top of the wire timeout when the native binding is wedged. |
|
||||
| `OMNIROUTE_PPLX_TLS_TIMEOUT_MS` | `30000` | Wire-level timeout for the bogdanfinn/tls-client koffi binding (`perplexityTlsClient.ts`). |
|
||||
| `OMNIROUTE_PPLX_TLS_GRACE_MS` | `10000` | JS-side grace added on top of the wire timeout when the native binding is wedged. |
|
||||
| `OMNIROUTE_PPLX_SEARCH_HINT` | `0` (off) | Appends "You have built-in web search. Answer questions directly using search results." to the caller's system message (`perplexity-web/protocol.ts`). Off by default — Perplexity searches anyway, and the sentence leaks into replies as meta-commentary for coding clients. Set `1`/`true`/`yes`/`on` to restore. |
|
||||
| `OMNIROUTE_GROK_TLS_TIMEOUT_MS` | `60000` | Wire-level timeout for the bogdanfinn/tls-client koffi binding (`grokTlsClient.ts`). |
|
||||
| `OMNIROUTE_GROK_TLS_GRACE_MS` | `10000` | JS-side grace added on top of the wire timeout when the native binding is wedged. |
|
||||
| `OMNIROUTE_NOTION_TLS_TIMEOUT_MS` | `30000` | Wire-level timeout for the bogdanfinn/tls-client koffi binding (`notionTlsClient.ts`); the `notion-web` executor raises it per-request to `180000` for long generations. |
|
||||
|
||||
@@ -1070,7 +1070,7 @@ async function wsChat(
|
||||
|
||||
const fail = (error: string) => finish({ content: "", deltas: [], error });
|
||||
|
||||
timeout = setTimeout(() => fail(`Meta AI WS timed out (readyState=${ws.readyState})`), 30000);
|
||||
timeout = setTimeout(() => fail("Meta AI WebSocket timed out"), 30000);
|
||||
abortHandler = () => fail("Request aborted");
|
||||
signal?.addEventListener("abort", abortHandler, { once: true });
|
||||
|
||||
|
||||
@@ -370,29 +370,15 @@ export function buildPplxRequestBody(
|
||||
};
|
||||
}
|
||||
|
||||
const SEARCH_HINT = "You have built-in web search. Answer questions directly using search results.";
|
||||
|
||||
/**
|
||||
* Whether to append {@link SEARCH_HINT} to the caller's system message.
|
||||
*
|
||||
* It used to be unconditional. Perplexity's answer engine is search-first anyway, and
|
||||
* for coding clients the sentence leaks into replies as meta-commentary ("I need to
|
||||
* search before responding per my instructions"), so it is now opt-in via
|
||||
* `OMNIROUTE_PPLX_SEARCH_HINT`. Read per call rather than at module load so the flag
|
||||
* can be flipped without restarting the server (and so tests can toggle it).
|
||||
*/
|
||||
function searchHintEnabled(): boolean {
|
||||
return /^(1|true|yes|on)$/i.test(process.env.OMNIROUTE_PPLX_SEARCH_HINT ?? "");
|
||||
}
|
||||
|
||||
export function buildQuery(parsed: ParsedMessages, followUpUuid: string | null): string {
|
||||
if (followUpUuid) return parsed.currentMsg;
|
||||
|
||||
const obj: Record<string, unknown> = {};
|
||||
if (parsed.systemMsg.trim()) {
|
||||
obj.instructions = searchHintEnabled()
|
||||
? [parsed.systemMsg.trim(), SEARCH_HINT]
|
||||
: [parsed.systemMsg.trim()];
|
||||
obj.instructions = [
|
||||
parsed.systemMsg.trim(),
|
||||
"You have built-in web search. Answer questions directly using search results.",
|
||||
];
|
||||
}
|
||||
if (parsed.history.length > 0) {
|
||||
obj.history = parsed.history;
|
||||
|
||||
@@ -15,8 +15,6 @@ import { getModelSpec } from "../../../src/shared/constants/modelSpecs.ts";
|
||||
import {
|
||||
buildChangedToolNameMap,
|
||||
buildHistoricalToolResultContext,
|
||||
mergeConsecutiveSameRoleContents,
|
||||
type GeminiContent,
|
||||
} from "./openai-to-gemini/helpers.ts";
|
||||
|
||||
/**
|
||||
@@ -47,7 +45,7 @@ export function claudeToGeminiRequest(model, body, stream, credentials = null) {
|
||||
: null;
|
||||
const result: {
|
||||
model: string;
|
||||
contents: GeminiContent[];
|
||||
contents: Array<Record<string, unknown>>;
|
||||
generationConfig: Record<string, unknown>;
|
||||
safetySettings: unknown;
|
||||
systemInstruction?: { role: string; parts: Array<{ text: string }> };
|
||||
@@ -316,11 +314,6 @@ export function claudeToGeminiRequest(model, body, stream, credentials = null) {
|
||||
result._toolNameMap = changedToolNameMap;
|
||||
}
|
||||
|
||||
// Gemini strictly rejects requests containing consecutive messages with the same role
|
||||
// (400 INVALID_ARGUMENT: "Request contains consecutive messages with the same role").
|
||||
// Normalize adjacent same-role messages by concatenating their parts.
|
||||
result.contents = mergeConsecutiveSameRoleContents(result.contents);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@@ -39,13 +39,8 @@ import {
|
||||
escapeHistoricalContextAttribute,
|
||||
escapeHistoricalContextContent,
|
||||
buildHistoricalToolResultContext,
|
||||
type GeminiPart,
|
||||
type GeminiContent,
|
||||
mergeConsecutiveSameRoleContents,
|
||||
} from "./openai-to-gemini/helpers.ts";
|
||||
|
||||
export { mergeConsecutiveSameRoleContents, type GeminiContent, type GeminiPart };
|
||||
|
||||
// Observed Antigravity wrapper output cap, not an underlying model capability.
|
||||
// Keep this bridge-local: Antigravity currently caps visible output around 16K.
|
||||
// See: https://github.com/keisksw/antigravity-output-analysis
|
||||
@@ -61,6 +56,9 @@ const GEMINI_BUILTIN_TOOL_NAMES = new Set<string>([
|
||||
"googleSearch",
|
||||
]);
|
||||
|
||||
type GeminiPart = Record<string, unknown>;
|
||||
type GeminiContent = { role: string; parts: GeminiPart[] };
|
||||
|
||||
type GeminiFunctionDeclaration = {
|
||||
name: string;
|
||||
description: string;
|
||||
@@ -160,6 +158,29 @@ type GeminiToolNameOptions = {
|
||||
supportsSignatureBypass?: boolean;
|
||||
};
|
||||
|
||||
// Gemini-family APIs (incl. Antigravity / Vertex) reject a `contents[]` array that
|
||||
// has two adjacent entries with the same role:
|
||||
// 400 INVALID_ARGUMENT "Request contains consecutive messages with the same role".
|
||||
// Client history that carries consecutive user turns — or a tool-result turn (mapped
|
||||
// to role:"user") immediately followed by a plain user turn — would otherwise leak
|
||||
// that invalid alternation through. Merge adjacent same-role entries by concatenating
|
||||
// their parts, the same normalization the Kiro and Claude request paths already apply
|
||||
// (9router#2191).
|
||||
export function mergeConsecutiveSameRoleContents(contents: GeminiContent[]): GeminiContent[] {
|
||||
const merged: GeminiContent[] = [];
|
||||
for (const entry of contents) {
|
||||
const last = merged[merged.length - 1];
|
||||
if (last && last.role === entry.role) {
|
||||
last.parts.push(...entry.parts);
|
||||
} else {
|
||||
// Shallow-copy the entry and its `parts` array so a later same-role merge
|
||||
// (`last.parts.push(...)`) never mutates the caller's input objects.
|
||||
merged.push({ ...entry, parts: [...entry.parts] });
|
||||
}
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
|
||||
// Core: Convert OpenAI request to Gemini format (base for all variants)
|
||||
function openaiToGeminiBase(
|
||||
model: string,
|
||||
|
||||
@@ -152,29 +152,3 @@ export function buildHistoricalToolResultContext(name: string, response: unknown
|
||||
"</previous_tool_result_context>",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
export type GeminiPart = Record<string, unknown>;
|
||||
export type GeminiContent = { role: string; parts: GeminiPart[] };
|
||||
|
||||
// Gemini-family APIs (incl. Antigravity / Vertex) reject a `contents[]` array that
|
||||
// has two adjacent entries with the same role:
|
||||
// 400 INVALID_ARGUMENT "Request contains consecutive messages with the same role".
|
||||
// Client history that carries consecutive user turns — or a tool-result turn (mapped
|
||||
// to role:"user") immediately followed by a plain user turn — would otherwise leak
|
||||
// that invalid alternation through. Merge adjacent same-role entries by concatenating
|
||||
// their parts, the same normalization the Kiro and Claude request paths already apply
|
||||
// (9router#2191).
|
||||
export function mergeConsecutiveSameRoleContents(contents: GeminiContent[]): GeminiContent[] {
|
||||
const merged: GeminiContent[] = [];
|
||||
for (const entry of contents) {
|
||||
const last = merged[merged.length - 1];
|
||||
if (last && last.role === entry.role) {
|
||||
last.parts.push(...entry.parts);
|
||||
} else {
|
||||
// Shallow-copy the entry and its `parts` array so a later same-role merge
|
||||
// (`last.parts.push(...)`) never mutates the caller's input objects.
|
||||
merged.push({ ...entry, parts: [...entry.parts] });
|
||||
}
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
|
||||
@@ -1,77 +0,0 @@
|
||||
type DirectFetchOptions = RequestInit & { dispatcher?: unknown };
|
||||
type DirectFetch = (
|
||||
input: RequestInfo | URL,
|
||||
options: DirectFetchOptions
|
||||
) => Promise<Response>;
|
||||
|
||||
const DEFAULT_DIRECT_HEADERS_TIMEOUT_MS = 30_000;
|
||||
const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT";
|
||||
|
||||
export function resolveDirectHeadersTimeoutMs(
|
||||
env: Record<string, string | undefined> = process.env
|
||||
): number {
|
||||
const raw = env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS;
|
||||
if (raw == null || raw.trim() === "") return DEFAULT_DIRECT_HEADERS_TIMEOUT_MS;
|
||||
const parsed = Number(raw);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0;
|
||||
}
|
||||
|
||||
function createDirectResponseStartTimeout(timeoutMs: number): Error & { code: string } {
|
||||
const err = new Error(
|
||||
`Direct response did not start within ${timeoutMs}ms — retrying on a fresh socket`
|
||||
) as Error & { code: string };
|
||||
err.name = "TimeoutError";
|
||||
err.code = DIRECT_RESPONSE_START_TIMEOUT_CODE;
|
||||
return err;
|
||||
}
|
||||
|
||||
export function isDirectResponseStartTimeout(err: unknown): boolean {
|
||||
return (
|
||||
!!err &&
|
||||
typeof err === "object" &&
|
||||
"code" in err &&
|
||||
err.code === DIRECT_RESPONSE_START_TIMEOUT_CODE
|
||||
);
|
||||
}
|
||||
|
||||
function mergeAbortSignals(
|
||||
primary: AbortSignal | null | undefined,
|
||||
secondary: AbortSignal
|
||||
): AbortSignal {
|
||||
if (!primary) return secondary;
|
||||
if (primary.aborted) return primary;
|
||||
const controller = new AbortController();
|
||||
const onPrimaryAbort = () => controller.abort(primary.reason);
|
||||
const onSecondaryAbort = () => controller.abort(secondary.reason);
|
||||
const cleanup = () => {
|
||||
primary.removeEventListener("abort", onPrimaryAbort);
|
||||
secondary.removeEventListener("abort", onSecondaryAbort);
|
||||
};
|
||||
primary.addEventListener("abort", onPrimaryAbort, { once: true });
|
||||
secondary.addEventListener("abort", onSecondaryAbort, { once: true });
|
||||
controller.signal.addEventListener("abort", cleanup, { once: true });
|
||||
return controller.signal;
|
||||
}
|
||||
|
||||
export async function directFetchWithBoundedResponseStart(
|
||||
input: RequestInfo | URL,
|
||||
options: DirectFetchOptions,
|
||||
fetchImpl: DirectFetch,
|
||||
timeoutMs: number
|
||||
): Promise<Response> {
|
||||
if (!timeoutMs || timeoutMs <= 0) return fetchImpl(input, options);
|
||||
const attemptController = new AbortController();
|
||||
const timer = setTimeout(
|
||||
() => attemptController.abort(createDirectResponseStartTimeout(timeoutMs)),
|
||||
timeoutMs
|
||||
);
|
||||
timer.unref?.();
|
||||
try {
|
||||
return await fetchImpl(input, {
|
||||
...options,
|
||||
signal: mergeAbortSignals(options.signal, attemptController.signal),
|
||||
});
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
@@ -19,11 +19,6 @@ import {
|
||||
isControlPlaneProxyDirectFallbackEnabled,
|
||||
isFeatureFlagEnabled,
|
||||
} from "@/shared/utils/featureFlags";
|
||||
import {
|
||||
directFetchWithBoundedResponseStart,
|
||||
isDirectResponseStartTimeout,
|
||||
resolveDirectHeadersTimeoutMs,
|
||||
} from "./directResponseStartTimeout.ts";
|
||||
|
||||
// #9100: relay egress (Vercel / Deno / Cloudflare edge functions) used to go
|
||||
// through bare `originalFetch` — NO connection pooling, NO timeout, NO retry.
|
||||
@@ -159,6 +154,7 @@ type TlsFingerprintStore = {
|
||||
provider?: string | null;
|
||||
sessionScope?: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* #5217 (Gap-secondary): a mutable sink that records the proxy actually applied
|
||||
* by `runWithProxyContext` for the in-flight request. Executors that pin their
|
||||
@@ -806,7 +802,15 @@ async function patchedFetch(
|
||||
(deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher;
|
||||
return _nativeFetch(input, options);
|
||||
}
|
||||
// Direct undici path: bound response-start, fresh-socket retry, and body guard.
|
||||
// Direct connection (no proxy) — use undici with custom dispatcher for timeout control.
|
||||
// Falls back to original native fetch if dispatcher initialization fails (#1054).
|
||||
// Retries once on transient dispatcher errors before falling back (fix: proxyfetch-undici-retry).
|
||||
//
|
||||
// Non-replayable body guard: if the body is stream-like (ReadableStream/Blob)
|
||||
// or the input is a Request that carries a body, the first dispatcher attempt
|
||||
// owns that body. Retrying or falling back to native fetch would replay a
|
||||
// consumed/locked body and can mask the original transport error with
|
||||
// "Response body object should not be disturbed or locked".
|
||||
const hasNonReplayableBody = requestHasNonReplayableBody(input, options);
|
||||
const maxAttempts = hasNonReplayableBody ? 1 : 2;
|
||||
const _undiciDirect =
|
||||
@@ -814,44 +818,32 @@ async function patchedFetch(
|
||||
const _nativeFallback =
|
||||
(deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher;
|
||||
let lastDispatcherError: unknown = null;
|
||||
const directHeadersTimeoutMs = resolveDirectHeadersTimeoutMs();
|
||||
let targetHostForLogs = "";
|
||||
try {
|
||||
targetHostForLogs = new URL(targetUrl).host;
|
||||
} catch {
|
||||
// ignore — logging is best-effort
|
||||
}
|
||||
for (let attempt = 0; attempt < maxAttempts; attempt++) {
|
||||
try {
|
||||
return await directFetchWithBoundedResponseStart(
|
||||
input,
|
||||
{
|
||||
...options,
|
||||
dispatcher: attempt === 0 ? getDefaultDispatcher() : getRetryDispatcher(),
|
||||
},
|
||||
_undiciDirect,
|
||||
directHeadersTimeoutMs
|
||||
);
|
||||
return await _undiciDirect(input, {
|
||||
...options,
|
||||
// #4252: first attempt uses the pooled keep-alive dispatcher; a retry
|
||||
// (after a transient socket error) uses the no-keep-alive dispatcher so
|
||||
// it opens a FRESH socket instead of grabbing another stale pooled one
|
||||
// — the burst pattern was the retry re-hitting a dead pooled socket and
|
||||
// then falling through to native fetch (which also pools) → 502.
|
||||
dispatcher: attempt === 0 ? getDefaultDispatcher() : getRetryDispatcher(),
|
||||
});
|
||||
} catch (dispatcherError) {
|
||||
if (isDirectResponseStartTimeout(dispatcherError)) {
|
||||
if (attempt === 0 && maxAttempts > 1) {
|
||||
console.warn(
|
||||
`[ProxyFetch] Direct response-start timeout (${directHeadersTimeoutMs}ms) on pooled dispatcher — retrying on fresh no-keep-alive dispatcher: ${targetHostForLogs}`
|
||||
);
|
||||
lastDispatcherError = dispatcherError;
|
||||
continue;
|
||||
}
|
||||
throw dispatcherError;
|
||||
}
|
||||
const msg =
|
||||
dispatcherError instanceof Error ? dispatcherError.message : String(dispatcherError);
|
||||
// CAUTION: Do NOT fallback to native fetch if the error is a version mismatch (invalid onRequestStart)
|
||||
// because the native fetch will definitely fail with the undici v8 dispatcher.
|
||||
if (msg.includes("onRequestStart")) {
|
||||
console.error(
|
||||
`[ProxyFetch] Fatal version mismatch: Dispatcher (v8) vs Fetch (v6/native). Hardware upgrade or SOCKS5 config isolation required. Error: ${msg}`
|
||||
);
|
||||
throw dispatcherError;
|
||||
}
|
||||
// Retry/fallback only for connection errors, never HTTP errors.
|
||||
// Only retry/fallback for connection/dispatcher errors, not HTTP errors.
|
||||
// Prefer the .code property when available (more stable across undici
|
||||
// versions than message-string matching); fall back to substring match
|
||||
// for errors that lack a structured code.
|
||||
tagProxyUnreachable(dispatcherError);
|
||||
const errCode = (dispatcherError as { code?: unknown })?.code;
|
||||
if (
|
||||
@@ -862,7 +854,10 @@ async function patchedFetch(
|
||||
msg.includes("UND_ERR")
|
||||
) {
|
||||
if (attempt === 0 && maxAttempts > 1) {
|
||||
// Retry after a short fixed backoff on a fresh socket.
|
||||
// First failure — retry once after a short backoff before giving up.
|
||||
// Delay is OMNIROUTE_RETRY_BACKOFF_MS (default 10ms): a fixed backoff
|
||||
// beats random jitter here because the retry opens a fresh socket, so
|
||||
// jitter was pure added latency with no herd benefit.
|
||||
lastDispatcherError = dispatcherError;
|
||||
await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS));
|
||||
continue;
|
||||
@@ -878,7 +873,7 @@ async function patchedFetch(
|
||||
throw tagProxyUnreachable(dispatcherError);
|
||||
}
|
||||
|
||||
// Exhausted attempts: try proxy fallback before native fetch.
|
||||
// All attempts exhausted — try proxy fallback before native fetch
|
||||
if (
|
||||
!tlsDirectFallback &&
|
||||
source === "direct" &&
|
||||
@@ -904,14 +899,20 @@ async function patchedFetch(
|
||||
}
|
||||
}
|
||||
}
|
||||
// Preserve the original monitoring phrase and append the transport cause.
|
||||
// Preserve original phrase intact for monitoring: "Undici dispatcher failed, falling back to native fetch"
|
||||
// #4252: append the flattened err.cause (code/syscall/errno/address) — the bare
|
||||
// "fetch failed" message hides what actually broke, making bursts undiagnosable.
|
||||
console.warn(
|
||||
`[ProxyFetch] Undici dispatcher failed, falling back to native fetch (after retry): ${describeFetchCause(dispatcherError)}`
|
||||
);
|
||||
try {
|
||||
return await _nativeFallback(input, options);
|
||||
} catch (nativeError) {
|
||||
// Surface both dispatcher and native causes immediately.
|
||||
// #4252: both the undici dispatcher AND native fetch failed. Surface BOTH
|
||||
// causes (server log) and tag the propagated error so the combo executor sees
|
||||
// a diagnosable failure IMMEDIATELY instead of a bare "fetch failed" — the
|
||||
// latter left jobs sitting until the 30s semaphore queue timeout, which then
|
||||
// tripped the circuit breaker.
|
||||
const detail = `dispatcher=[${describeFetchCause(dispatcherError)}] native=[${describeFetchCause(nativeError)}]`;
|
||||
console.warn(`[ProxyFetch] native fetch fallback ALSO failed: ${detail}`);
|
||||
if (nativeError instanceof Error) {
|
||||
|
||||
@@ -7,61 +7,6 @@ export async function OPTIONS() {
|
||||
return handleCorsOptions();
|
||||
}
|
||||
|
||||
const DEFAULT_LIST_LIMIT = 20;
|
||||
const MAX_LIST_LIMIT = 10000;
|
||||
|
||||
export function parseFilesListQuery(searchParams: URLSearchParams):
|
||||
| {
|
||||
ok: true;
|
||||
limit: number;
|
||||
after: string | undefined;
|
||||
order: "asc" | "desc";
|
||||
purpose: string | undefined;
|
||||
}
|
||||
| { ok: false; response: Response } {
|
||||
const rawLimit = searchParams.get("limit");
|
||||
let limit = DEFAULT_LIST_LIMIT;
|
||||
|
||||
if (rawLimit !== null) {
|
||||
if (!/^\d+$/.test(rawLimit)) {
|
||||
return {
|
||||
ok: false,
|
||||
response: NextResponse.json(
|
||||
{ error: { message: "limit must be a positive integer", type: "invalid_request_error" } },
|
||||
{ status: 400, headers: CORS_HEADERS }
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
limit = Number.parseInt(rawLimit, 10);
|
||||
if (limit < 1 || limit > MAX_LIST_LIMIT) {
|
||||
return {
|
||||
ok: false,
|
||||
response: NextResponse.json(
|
||||
{
|
||||
error: {
|
||||
message: `limit must be between 1 and ${MAX_LIST_LIMIT}`,
|
||||
type: "invalid_request_error",
|
||||
},
|
||||
},
|
||||
{ status: 400, headers: CORS_HEADERS }
|
||||
),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const orderParam = searchParams.get("order");
|
||||
const order = orderParam === "asc" ? "asc" : "desc";
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
limit,
|
||||
after: searchParams.get("after") || undefined,
|
||||
order,
|
||||
purpose: searchParams.get("purpose") || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
export async function POST(request: Request) {
|
||||
const scope = await getApiKeyRequestScope(request);
|
||||
if (scope.rejection) return scope.rejection;
|
||||
@@ -133,9 +78,10 @@ export async function GET(request: Request) {
|
||||
const apiKeyId = scope.apiKeyId;
|
||||
|
||||
const { searchParams } = new URL(request.url);
|
||||
const parsed = parseFilesListQuery(searchParams);
|
||||
if (!parsed.ok) return parsed.response;
|
||||
const { limit, after, order, purpose } = parsed;
|
||||
const limit = Math.min(Number.parseInt(searchParams.get("limit") || "20") || 20, 10000);
|
||||
const after = searchParams.get("after") || undefined;
|
||||
const order = (searchParams.get("order") as "asc" | "desc") || "desc";
|
||||
const purpose = searchParams.get("purpose") || undefined;
|
||||
|
||||
// We fetch limit + 1 to check if there are more items
|
||||
const files = listFiles({
|
||||
|
||||
@@ -1,77 +0,0 @@
|
||||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert";
|
||||
import { createFile, deleteFile } from "@/lib/db/files";
|
||||
import { GET, parseFilesListQuery } from "@/app/api/v1/files/route";
|
||||
|
||||
describe("GET /v1/files limit validation", () => {
|
||||
it("defaults to 20 when limit is absent", () => {
|
||||
const parsed = parseFilesListQuery(new URLSearchParams("order=asc"));
|
||||
|
||||
assert.equal(parsed.ok, true);
|
||||
if (!parsed.ok) return;
|
||||
assert.equal(parsed.limit, 20);
|
||||
});
|
||||
|
||||
it("parses an explicit positive integer limit", () => {
|
||||
const parsed = parseFilesListQuery(new URLSearchParams("limit=2&order=asc&purpose=batch"));
|
||||
|
||||
assert.equal(parsed.ok, true);
|
||||
if (!parsed.ok) return;
|
||||
assert.equal(parsed.limit, 2);
|
||||
assert.equal(parsed.order, "asc");
|
||||
assert.equal(parsed.purpose, "batch");
|
||||
});
|
||||
|
||||
it("rejects non-integer, zero, and oversized limits", async () => {
|
||||
for (const rawLimit of ["abc", "1.5", "-1", "0", "10001"]) {
|
||||
const parsed = parseFilesListQuery(
|
||||
new URLSearchParams(`limit=${encodeURIComponent(rawLimit)}`)
|
||||
);
|
||||
assert.equal(parsed.ok, false, `limit=${rawLimit} should be rejected`);
|
||||
if (parsed.ok) continue;
|
||||
assert.equal(parsed.response.status, 400);
|
||||
const body = await parsed.response.json();
|
||||
assert.equal(body.error.type, "invalid_request_error");
|
||||
}
|
||||
});
|
||||
|
||||
it("returns only the requested number of files over HTTP", async () => {
|
||||
const created = [
|
||||
createFile({
|
||||
bytes: 1,
|
||||
filename: "test-files-limit-http-a.txt",
|
||||
purpose: "assistants",
|
||||
content: Buffer.from("a"),
|
||||
mimeType: "text/plain",
|
||||
}),
|
||||
createFile({
|
||||
bytes: 1,
|
||||
filename: "test-files-limit-http-b.txt",
|
||||
purpose: "assistants",
|
||||
content: Buffer.from("b"),
|
||||
mimeType: "text/plain",
|
||||
}),
|
||||
];
|
||||
|
||||
try {
|
||||
const response = await GET(
|
||||
new Request("http://localhost/v1/files?limit=1&purpose=assistants")
|
||||
);
|
||||
assert.equal(response.status, 200);
|
||||
const body = await response.json();
|
||||
assert.equal(body.object, "list");
|
||||
assert.equal(body.data.length, 1);
|
||||
assert.equal(body.has_more, true);
|
||||
} finally {
|
||||
for (const file of created) deleteFile(file.id);
|
||||
}
|
||||
});
|
||||
|
||||
it("returns 400 over HTTP for an invalid limit instead of listing files", async () => {
|
||||
const response = await GET(new Request("http://localhost/v1/files?limit=-1"));
|
||||
|
||||
assert.equal(response.status, 400);
|
||||
const body = await response.json();
|
||||
assert.equal(body.error.type, "invalid_request_error");
|
||||
});
|
||||
});
|
||||
@@ -1,150 +0,0 @@
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
const { claudeToGeminiRequest } =
|
||||
await import("../../open-sse/translator/request/claude-to-gemini.ts");
|
||||
|
||||
test("Claude -> Gemini merges consecutive user text turns into a single user turn", () => {
|
||||
const result = claudeToGeminiRequest(
|
||||
"gemini-2.5-flash",
|
||||
{
|
||||
messages: [
|
||||
{ role: "user", content: "hello" },
|
||||
{ role: "user", content: [{ type: "text", text: "world" }] },
|
||||
],
|
||||
},
|
||||
false
|
||||
);
|
||||
|
||||
assert.equal(result.contents.length, 1);
|
||||
assert.equal(result.contents[0].role, "user");
|
||||
assert.deepEqual(result.contents[0].parts, [{ text: "hello" }, { text: "world" }]);
|
||||
});
|
||||
|
||||
test("Claude -> Gemini merges tool_result and subsequent user instruction into single user turn", () => {
|
||||
const result = claudeToGeminiRequest(
|
||||
"gemini-2.5-flash",
|
||||
{
|
||||
messages: [
|
||||
{ role: "user", content: "Calculate 2+2" },
|
||||
{
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "I will calculate that." }],
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
type: "tool_result",
|
||||
tool_use_id: "tool_call_1",
|
||||
content: "4",
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: "Now add 10 to that result",
|
||||
},
|
||||
],
|
||||
},
|
||||
false
|
||||
);
|
||||
|
||||
// Contents must alternate properly and not have consecutive same-role turns
|
||||
for (let i = 1; i < result.contents.length; i++) {
|
||||
assert.notEqual(
|
||||
result.contents[i].role,
|
||||
result.contents[i - 1].role,
|
||||
`Consecutive same-role detected at index ${i - 1} and ${i}: ${result.contents[i].role}`
|
||||
);
|
||||
}
|
||||
|
||||
// The last turn should be a merged user turn containing both the tool context and the text
|
||||
const lastTurn = result.contents[result.contents.length - 1];
|
||||
assert.equal(lastTurn.role, "user");
|
||||
assert.equal(lastTurn.parts.length, 2);
|
||||
assert.ok(
|
||||
typeof (lastTurn.parts[0] as { text: string }).text === "string" &&
|
||||
(lastTurn.parts[0] as { text: string }).text.includes("previous_tool_result_context")
|
||||
);
|
||||
assert.deepEqual(lastTurn.parts[1], { text: "Now add 10 to that result" });
|
||||
});
|
||||
|
||||
test("Claude -> Gemini preserves alternating conversation turns without spurious merging", () => {
|
||||
const result = claudeToGeminiRequest(
|
||||
"gemini-2.5-flash",
|
||||
{
|
||||
messages: [
|
||||
{ role: "user", content: "Hello" },
|
||||
{ role: "assistant", content: "Hi! How can I help?" },
|
||||
{ role: "user", content: "What is the capital of France?" },
|
||||
],
|
||||
},
|
||||
false
|
||||
);
|
||||
|
||||
assert.equal(result.contents.length, 3);
|
||||
assert.equal(result.contents[0].role, "user");
|
||||
assert.deepEqual(result.contents[0].parts, [{ text: "Hello" }]);
|
||||
assert.equal(result.contents[1].role, "model");
|
||||
assert.deepEqual(result.contents[1].parts, [{ text: "Hi! How can I help?" }]);
|
||||
assert.equal(result.contents[2].role, "user");
|
||||
assert.deepEqual(result.contents[2].parts, [{ text: "What is the capital of France?" }]);
|
||||
});
|
||||
|
||||
test("Claude -> Gemini merges three or more consecutive user turns into a single user turn", () => {
|
||||
const result = claudeToGeminiRequest(
|
||||
"gemini-2.5-flash",
|
||||
{
|
||||
messages: [
|
||||
{ role: "user", content: "part 1" },
|
||||
{ role: "user", content: "part 2" },
|
||||
{ role: "user", content: "part 3" },
|
||||
],
|
||||
},
|
||||
false
|
||||
);
|
||||
|
||||
assert.equal(result.contents.length, 1);
|
||||
assert.equal(result.contents[0].role, "user");
|
||||
assert.deepEqual(result.contents[0].parts, [
|
||||
{ text: "part 1" },
|
||||
{ text: "part 2" },
|
||||
{ text: "part 3" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("Claude -> Gemini handles empty messages array without error", () => {
|
||||
const result = claudeToGeminiRequest(
|
||||
"gemini-2.5-flash",
|
||||
{
|
||||
messages: [],
|
||||
},
|
||||
false
|
||||
);
|
||||
|
||||
assert.deepEqual(result.contents, []);
|
||||
});
|
||||
|
||||
test("Claude -> Gemini merges consecutive assistant turns into a single model turn", () => {
|
||||
const result = claudeToGeminiRequest(
|
||||
"gemini-2.5-flash",
|
||||
{
|
||||
messages: [
|
||||
{ role: "user", content: "hello" },
|
||||
{ role: "assistant", content: "response part 1" },
|
||||
{ role: "assistant", content: [{ type: "text", text: "response part 2" }] },
|
||||
],
|
||||
},
|
||||
false
|
||||
);
|
||||
|
||||
assert.equal(result.contents.length, 2);
|
||||
assert.equal(result.contents[0].role, "user");
|
||||
assert.deepEqual(result.contents[0].parts, [{ text: "hello" }]);
|
||||
assert.equal(result.contents[1].role, "model");
|
||||
assert.deepEqual(result.contents[1].parts, [
|
||||
{ text: "response part 1" },
|
||||
{ text: "response part 2" },
|
||||
]);
|
||||
});
|
||||
@@ -1,164 +0,0 @@
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import {
|
||||
MuseSparkWebExecutor,
|
||||
__resetMuseSparkConversationCacheForTesting,
|
||||
__setMuseSparkWebSocketForTesting,
|
||||
} from "../../open-sse/executors/muse-spark-web.ts";
|
||||
import { WebSocket } from "ws";
|
||||
|
||||
// #10727: Meta AI (muse-spark-web) times out on the WS send-message step at
|
||||
// exactly the executor's hardcoded 30s timeout, with no onerror/onclose
|
||||
// firing first. The reporter's log shows the flow reaching wsChat and
|
||||
// hanging until the timeout fires, meaning Meta's gateway either never
|
||||
// truly opens the socket or silently drops frames after opening — but the
|
||||
// old flat "Meta AI WebSocket timed out" message could not distinguish
|
||||
// those two failure modes for whoever debugs the next occurrence.
|
||||
//
|
||||
// Root-causing (and fixing) the reverse-engineered private WS protocol
|
||||
// itself requires a live meta.ai session + a fresh DevTools capture (see
|
||||
// the plan-file's `needs-vps` verdict) — not achievable in this sandbox.
|
||||
// This regression test locks in the diagnosability improvement that *is*
|
||||
// verifiable here: the timeout error now reports the socket's readyState
|
||||
// at the moment it fires, so a future report can tell "never opened"
|
||||
// (readyState 0) apart from "opened but Meta went silent" (readyState 1).
|
||||
|
||||
/**
|
||||
* Intercepts only the wsChat 30000ms timeout registration and lets every
|
||||
* other setTimeout (including the mock WebSocket's own onopen scheduling)
|
||||
* run for real. Firing the captured callback directly — instead of
|
||||
* advancing a fake clock — keeps the test fast and avoids interleaving
|
||||
* bugs between fake timers and the executor's real async/await chain.
|
||||
*/
|
||||
function interceptWsTimeout(): { fire: () => void; restore: () => void } {
|
||||
const original = globalThis.setTimeout;
|
||||
let captured: (() => void) | null = null;
|
||||
globalThis.setTimeout = ((cb: (...a: unknown[]) => void, ms?: number, ...args: unknown[]) => {
|
||||
if (ms === 30000 && captured === null) {
|
||||
captured = cb as () => void;
|
||||
return 0 as unknown as ReturnType<typeof setTimeout>;
|
||||
}
|
||||
return original(cb as () => void, ms, ...args);
|
||||
}) as typeof setTimeout;
|
||||
return {
|
||||
fire: () => {
|
||||
assert.ok(captured, "the 30000ms wsChat timeout was never registered");
|
||||
captured?.();
|
||||
},
|
||||
restore: () => {
|
||||
globalThis.setTimeout = original;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
class NeverOpensWebSocket {
|
||||
onopen: (() => void) | null = null;
|
||||
onmessage: ((evt: { data: string }) => void) | null = null;
|
||||
onclose: (() => void) | null = null;
|
||||
onerror: ((evt: Error) => void) | null = null;
|
||||
readyState = WebSocket.CONNECTING;
|
||||
url: string;
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
// Never calls onopen, onmessage, onerror, or onclose — mirrors the
|
||||
// reported symptom exactly: the socket just hangs until the timeout.
|
||||
}
|
||||
send(_data: Uint8Array | string) {}
|
||||
close() {}
|
||||
}
|
||||
|
||||
class OpensThenSilentWebSocket {
|
||||
onopen: (() => void) | null = null;
|
||||
onmessage: ((evt: { data: string }) => void) | null = null;
|
||||
onclose: (() => void) | null = null;
|
||||
onerror: ((evt: Error) => void) | null = null;
|
||||
readyState = WebSocket.CONNECTING;
|
||||
url: string;
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
setTimeout(() => {
|
||||
this.readyState = WebSocket.OPEN;
|
||||
this.onopen?.();
|
||||
}, 0);
|
||||
}
|
||||
send(_data: Uint8Array | string) {}
|
||||
close() {}
|
||||
}
|
||||
|
||||
function baseInput(connectionId: string): Parameters<MuseSparkWebExecutor["execute"]>[0] {
|
||||
return {
|
||||
model: "muse-spark",
|
||||
body: { messages: [{ role: "user", content: "ping" }] },
|
||||
stream: false,
|
||||
credentials: {
|
||||
apiKey: "ecto_1_sess=test123",
|
||||
connectionId,
|
||||
providerSpecificData: { authorization: "ecto1:test-auth-token" },
|
||||
},
|
||||
signal: null,
|
||||
log: null,
|
||||
upstreamExtraHeaders: undefined,
|
||||
} as Parameters<MuseSparkWebExecutor["execute"]>[0];
|
||||
}
|
||||
|
||||
test("#10727: WS timeout while still CONNECTING reports readyState=0 (never opened)", async () => {
|
||||
__resetMuseSparkConversationCacheForTesting();
|
||||
const executor = new MuseSparkWebExecutor();
|
||||
const originalFetch = globalThis.fetch;
|
||||
globalThis.fetch = async () => new Response("{}", { status: 200 });
|
||||
const restore = __setMuseSparkWebSocketForTesting(
|
||||
NeverOpensWebSocket as unknown as typeof WebSocket
|
||||
);
|
||||
const timeoutHook = interceptWsTimeout();
|
||||
try {
|
||||
const resultPromise = executor.execute(baseInput("conn-10727-never-opens"));
|
||||
// Let the GraphQL warmup/mode-switch awaits and the WS constructor run
|
||||
// before the 30s timeout is registered.
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
timeoutHook.fire();
|
||||
|
||||
const result = await resultPromise;
|
||||
assert.equal(result.response.status, 502);
|
||||
const body = await result.response.json();
|
||||
assert.match(
|
||||
body.error.message,
|
||||
/readyState=0/,
|
||||
"timeout while the socket never left CONNECTING must report readyState=0"
|
||||
);
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
restore();
|
||||
timeoutHook.restore();
|
||||
}
|
||||
});
|
||||
|
||||
test("#10727: WS timeout after a successful open reports readyState=1 (opened, then silent)", async () => {
|
||||
__resetMuseSparkConversationCacheForTesting();
|
||||
const executor = new MuseSparkWebExecutor();
|
||||
const originalFetch = globalThis.fetch;
|
||||
globalThis.fetch = async () => new Response("{}", { status: 200 });
|
||||
const restore = __setMuseSparkWebSocketForTesting(
|
||||
OpensThenSilentWebSocket as unknown as typeof WebSocket
|
||||
);
|
||||
const timeoutHook = interceptWsTimeout();
|
||||
try {
|
||||
const resultPromise = executor.execute(baseInput("conn-10727-opens-silent"));
|
||||
// Let the GraphQL awaits run, the WS open (its own real setTimeout(...,0)),
|
||||
// and the intro/prompt frames send before the 30s timeout is registered.
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
timeoutHook.fire();
|
||||
|
||||
const result = await resultPromise;
|
||||
assert.equal(result.response.status, 502);
|
||||
const body = await result.response.json();
|
||||
assert.match(
|
||||
body.error.message,
|
||||
/readyState=1/,
|
||||
"timeout after the socket reached OPEN must report readyState=1, not the never-opened case"
|
||||
);
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
restore();
|
||||
timeoutHook.restore();
|
||||
}
|
||||
});
|
||||
@@ -912,35 +912,6 @@ test("Model mapping: thinking mode uses thinking variant", async () => {
|
||||
}
|
||||
});
|
||||
|
||||
// ─── The search hint is opt-in ──────────────────────────────────────────────
|
||||
// It used to be appended to every system message and leaked into answers as
|
||||
// meta-commentary, which is noise for coding clients.
|
||||
|
||||
test("buildQuery: search hint is off by default and opt-in via env", async () => {
|
||||
const { buildQuery } = await import("../../open-sse/executors/perplexity-web/protocol.ts");
|
||||
const parsed = { systemMsg: "You are terse.", history: [], currentMsg: "hi" };
|
||||
const HINT = "built-in web search";
|
||||
const prev = process.env.OMNIROUTE_PPLX_SEARCH_HINT;
|
||||
|
||||
try {
|
||||
delete process.env.OMNIROUTE_PPLX_SEARCH_HINT;
|
||||
const off = JSON.parse(buildQuery(parsed, null));
|
||||
assert.deepEqual(off.instructions, ["You are terse."]);
|
||||
assert.equal(off.query, "hi");
|
||||
|
||||
process.env.OMNIROUTE_PPLX_SEARCH_HINT = "1";
|
||||
const on = JSON.parse(buildQuery(parsed, null));
|
||||
assert.equal(on.instructions.length, 2);
|
||||
assert.ok(on.instructions[1].includes(HINT));
|
||||
|
||||
process.env.OMNIROUTE_PPLX_SEARCH_HINT = "0";
|
||||
assert.equal(JSON.parse(buildQuery(parsed, null)).instructions.length, 1);
|
||||
} finally {
|
||||
if (prev === undefined) delete process.env.OMNIROUTE_PPLX_SEARCH_HINT;
|
||||
else process.env.OMNIROUTE_PPLX_SEARCH_HINT = prev;
|
||||
}
|
||||
});
|
||||
|
||||
// ─── Test: Live multi-step stream (no COMPLETED; text_completed + diffs) ────
|
||||
|
||||
test("Live multi-step: reconstructs answer without status COMPLETED", async () => {
|
||||
|
||||
@@ -1,143 +0,0 @@
|
||||
/**
|
||||
* #10214 — Direct (no-proxy) requests stall on a silently-dropped pooled
|
||||
* keep-alive socket until the caller's deadline or a service restart.
|
||||
*
|
||||
* The default direct dispatcher pools keep-alive sockets for up to
|
||||
* `fetchKeepAliveTimeoutMs` (4 s). A socket that silently drops (half-open, no
|
||||
* RST) surfaces NO transport error — undici's headersTimeout (600 s default) is
|
||||
* the only guard, so the existing fresh-socket retry (which fires on
|
||||
* UND_ERR/ECONNRESET/fetch-failed) never triggers. Observed live: opencode-go
|
||||
* and command-code stall 100% of routed requests until `systemctl restart`.
|
||||
*
|
||||
* The fix bounds the response-start window per direct attempt
|
||||
* (`OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`, default 30 s) and retries once on the
|
||||
* fresh no-keep-alive dispatcher (a brand-new socket) when the pooled attempt
|
||||
* times out — converting the zombie-socket stall into a clean failover.
|
||||
*/
|
||||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { proxyFetch } from "../../open-sse/utils/proxyFetch.ts";
|
||||
import { getDefaultDispatcher, getRetryDispatcher } from "../../open-sse/utils/proxyDispatcher.ts";
|
||||
|
||||
const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT";
|
||||
|
||||
/** Simulates a silent half-open pooled socket: the request never resolves, but
|
||||
* observes the abort signal like real undici does (rejects with the reason). */
|
||||
function hangingFetch(capture: {
|
||||
calls: number;
|
||||
dispatchers: unknown[];
|
||||
}): (input: RequestInfo | URL, init?: RequestInit) => Promise<Response> {
|
||||
return (input, init) => {
|
||||
capture.calls++;
|
||||
capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
|
||||
return new Promise<Response>((_, reject) => {
|
||||
const signal = init?.signal;
|
||||
signal?.addEventListener(
|
||||
"abort",
|
||||
() =>
|
||||
reject(signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason))),
|
||||
{ once: true }
|
||||
);
|
||||
// never resolve — the upstream accepted the connection but sends nothing
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
function withFastTimeout<T>(fn: () => Promise<T>): Promise<T> {
|
||||
process.env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS = "50";
|
||||
return fn().finally(() => {
|
||||
delete process.env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS;
|
||||
});
|
||||
}
|
||||
|
||||
test("#10214 a response-start timeout on the pooled attempt retries on the FRESH no-keep-alive dispatcher", async () => {
|
||||
const capture = { calls: 0, dispatchers: [] as unknown[] };
|
||||
|
||||
const mockUndici = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
|
||||
capture.calls++;
|
||||
capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
|
||||
if (capture.calls === 1) {
|
||||
// First attempt hits a silently-dead pooled socket — hang, no error.
|
||||
return new Promise<Response>((_, reject) => {
|
||||
init?.signal?.addEventListener("abort", () => reject(init.signal!.reason), { once: true });
|
||||
});
|
||||
}
|
||||
return new Response("ok", { status: 200 });
|
||||
};
|
||||
const mockNative = async (): Promise<Response> =>
|
||||
new Response("native-should-not-fire", { status: 200 });
|
||||
|
||||
const res = await withFastTimeout(() =>
|
||||
proxyFetch(
|
||||
"https://opencode.ai/zen/go/v1/chat/completions",
|
||||
{ method: "POST" },
|
||||
{ undiciFetch: mockUndici, nativeFetch: mockNative }
|
||||
)
|
||||
);
|
||||
|
||||
assert.equal(capture.calls, 2, "pooled attempt times out and must retry once");
|
||||
assert.equal(await res.text(), "ok");
|
||||
// The regression guard: attempt 0 used the pooled keep-alive dispatcher; the
|
||||
// retry used the fresh no-keep-alive dispatcher — a DIFFERENT instance, so the
|
||||
// retry opens a brand-new socket that cannot be the zombie.
|
||||
assert.equal(
|
||||
capture.dispatchers[0],
|
||||
getDefaultDispatcher(),
|
||||
"first attempt must use the pooled default dispatcher"
|
||||
);
|
||||
assert.equal(
|
||||
capture.dispatchers[1],
|
||||
getRetryDispatcher(),
|
||||
"timeout retry must use the fresh no-keep-alive dispatcher"
|
||||
);
|
||||
assert.notEqual(capture.dispatchers[0], capture.dispatchers[1]);
|
||||
});
|
||||
|
||||
test("#10214 when the fresh-dispatcher retry also stalls, the timeout surfaces (no native fallback)", async () => {
|
||||
const capture = { calls: 0, dispatchers: [] as unknown[] };
|
||||
const mockUndici = hangingFetch(capture);
|
||||
const mockNative = async (): Promise<Response> =>
|
||||
new Response("native-should-not-fire", { status: 200 });
|
||||
|
||||
await assert.rejects(
|
||||
withFastTimeout(() =>
|
||||
proxyFetch(
|
||||
"https://opencode.ai/zen/go/v1/chat/completions",
|
||||
{ method: "POST" },
|
||||
{ undiciFetch: mockUndici, nativeFetch: mockNative }
|
||||
)
|
||||
),
|
||||
(err: unknown) => {
|
||||
assert.equal(
|
||||
(err as { code?: unknown }).code,
|
||||
DIRECT_RESPONSE_START_TIMEOUT_CODE,
|
||||
"final failure must be the classified direct response-start timeout"
|
||||
);
|
||||
return true;
|
||||
}
|
||||
);
|
||||
assert.equal(capture.calls, 2, "both attempts must have been made");
|
||||
assert.equal(capture.dispatchers[0], getDefaultDispatcher());
|
||||
assert.equal(capture.dispatchers[1], getRetryDispatcher());
|
||||
});
|
||||
|
||||
test("#10214 a healthy fast response is untouched by the bound (single attempt, no retry)", async () => {
|
||||
const capture = { calls: 0, dispatchers: [] as unknown[] };
|
||||
const mockUndici = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
|
||||
capture.calls++;
|
||||
capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
|
||||
return new Response("ok", { status: 200 });
|
||||
};
|
||||
|
||||
const res = await withFastTimeout(() =>
|
||||
proxyFetch(
|
||||
"https://opencode.ai/zen/go/v1/chat/completions",
|
||||
{ method: "POST" },
|
||||
{ undiciFetch: mockUndici }
|
||||
)
|
||||
);
|
||||
|
||||
assert.equal(capture.calls, 1, "healthy request must not retry");
|
||||
assert.equal(capture.dispatchers[0], getDefaultDispatcher());
|
||||
assert.equal(await res.text(), "ok");
|
||||
});
|
||||
Reference in New Issue
Block a user