Files
OmniRoute/tests/unit/chatcore-translation-paths.test.ts
Markus Hartung cc17b304ab fix(sse): Gemini TPM/RPD quota classification + combo cooldown-wait resilience (#8213)
* fix(sse): Gemini TPM classification, combo-cooldown-wait for auto/quota-share, and target-timeout floor

Gemini TPM/RPM 429s were misclassified as QUOTA_EXHAUSTED because
sanitizeErrorMessage() truncates to the first line, hiding Google's
metric name and retry hint on lines 2-3. Added a rawMessage field
(internal-only, never reaches the client) and classifyGeminiQuotaMetricFromText()
to classify from the untruncated text, reordered ahead of the generic
credits/daily-quota checks.

Widened comboCooldownWaitEnabled (wait out a short transient cooldown
instead of crystallizing a 429/503) from quota-share-only to also cover
auto-strategy combos, and raised the wait ceiling to 65s/130s-budget/90s-cap
to match Gemini's ~60s TPM/RPM windows.

The per-target timeout (DEFAULT_COMBO_TARGET_TIMEOUT_MS, 120s) was shorter
than the new 130s cooldown-wait budget, so a target could get cut off
mid-wait with a synthetic 524 instead of completing the retry. Added
resolveComboTargetTimeoutMsForCombo()/isComboCooldownWaitEligible() in
comboConfig.ts to raise the per-target floor to budgetMs+buffer only for
wait-eligible strategies (auto/quota-share), verified live: a 12-request
concurrent burst against a TPM-exhausted combo went from 2/12 succeeding
(10 x 524) to 12/12 succeeding with zero 503/524.

Also: liveGeminiShared.ts's sendAndValidate now fails fast on a 503
instead of retrying past it, and the health dashboard + request logger
surface TPM stats alongside RPM/RPD.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* fix(sse): combo-exhausted rejection logs now capture request body + attempted models

recordRejectedRequestUsage() (the fast path for combo requests that never
reach handleChatCore, e.g. all targets locked by resilience cooldown)
hardcoded provider: "-" and never passed a request body to saveCallLog(),
so /dashboard/logs entries for these failures were nearly useless for
debugging: no way to see the client's request or which models were tried.

- recordRejectedRequestUsage() now accepts requestBody and persists it
  through the existing saveCallLog() artifact mechanism (same path
  handleChatCore's own logging uses).
- Added summarizeComboAttemptedModels(), which reads the combo's own model
  list (always available, unlike the response's combo-diagnostics headers —
  a model-level resilience-lockout skip never touches the
  exhaustedProviders/exhaustedConnections sets those headers are built
  from) to populate a real "provider" value instead of "-".
- Wired both into the call site in src/sse/handlers/chat.ts.

NOTE: unrelated to the Gemini TPM/combo-cooldown-wait fix on this branch —
landed here per operator request, to be split into its own branch/PR.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* feat(sse): synthetic streaming keep-alive event + 5-minute Gemini cooldown-wait ceiling

Many clients enforce a first-SSE-byte timeout, which made it unsafe to wait
out a longer upstream rate-limit cooldown on a streaming request — the
client would abandon the connection before any bytes arrived. This landed
in two parts:

1. Synthetic startup "thinking" event (OpenAI chat/completions format):
   the already-existing withEarlyStreamKeepalive wrapper (open-sse/utils/
   earlyStreamKeepalive.ts, wired into /v1/chat/completions, /v1/messages,
   /v1/responses since #2544) opens the SSE stream immediately once a
   request runs past its threshold, but only ever sent empty/no-op
   keepalive frames. Added a `startupFrame` option (defaults to
   `keepaliveFrame` — zero behavior change unless a route opts in) so the
   very first frame can carry real content instead. Wired
   OPENAI_STARTUP_THINKING_FRAME (a reasoning_content delta: "OmniRoute:
   got request, sending to provider") into /v1/chat/completions only —
   Claude Messages and Responses API formats both require a preceding
   envelope event (message_start / response.created) that a synthetic
   pre-dispatch frame can't safely fabricate without risking a duplicate
   envelope once the real stream arrives, so those two routes keep their
   existing (safe, proven) keepalive frames unchanged.

2. Raised the "wait out a known cooldown, then retry" ceiling to 5 minutes
   for both retry mechanisms, now that a client-side first-byte timeout is
   no longer a risk on the (opted-in) route:
   - comboCooldownWait (auto/quota-share combos, open-sse/services/combo.ts):
     maxWaitMs hard clamp raised 90s -> 300s (src/lib/resilience/settings/
     normalize.ts); defaults raised to maxWaitMs:90s/maxAttempts:5/
     budgetMs:300s. comboConfig.ts's resolveComboTargetTimeoutMsForCombo
     already derives the per-target timeout floor from budgetMs, so it
     tracks the new ceiling with no further changes.
   - waitForCooldown (direct, non-combo model requests, src/sse/handlers/
     chat.ts): this mechanism had NO cumulative cap before — only a
     per-wait cap (maxRetryWaitMs) and a retry count (maxRetries), so
     maxRetries x maxRetryWaitMs could exceed 5 minutes with no ceiling.
     Added a budgetMs field (mirrors comboCooldownWait) to
     WaitForCooldownSettings/CooldownAwareRetrySettings, threaded a
     requestRetryBudgetLeftMs tracker through chat.ts's requestAttemptLoop
     (mirrors combo.ts's comboCooldownBudgetLeftMs), and made
     getCooldownAwareRetryDecision refuse to wait once the cumulative
     budget is exhausted even if the single wait is under maxRetryWaitMs.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* fix(sse): extend the synthetic keep-alive thinking event to /v1/responses

Live incident (OpenClaw, log id 1784407081908-cbc24f): a /v1/responses
request to gemini/gemma-4-31b-it took 56s to produce a first byte and the
client disconnected (499 request_signal_aborted) — the same client-first-byte-
timeout problem the previous commit fixed for /v1/chat/completions, but
/v1/responses only had the generic bare-comment keepalive (no content), so it
wasn't covered.

Added RESPONSES_STARTUP_THINKING_FRAME: a self-contained synthetic reasoning
item (response.output_item.added -> reasoning_summary_part.added ->
reasoning_summary_text.delta -> reasoning_summary_part.done), opened AND
closed within this one frame rather than left dangling — it never carries a
response_id, so it can't collide with the real upstream response's own
independent response.created lifecycle that follows. Mirrors the abbreviated
delta+part.done close pattern open-sse/utils/stream.ts's own
emitSyntheticResponsesReasoningSummary already uses for real mid-stream
reasoning content.

Wired into src/app/api/v1/responses/route.ts via the startupFrame option
added in the previous commit.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* fix(sse): combo cooldown-wait vars reset every setTry, crystallizing a bogus 503 instead of waiting

Live incident (log id 1784416706646-51): a request to the "default" combo
(strategy=auto, maxSetRetries=3) hit a real Gemini TPM 429 on both gemma-4
targets, correctly classified as a short 40s rate_limit lockout — then
crystallized a 503 "all upstream accounts are inactive" in 6.9s instead of
ever reaching the cooldown-aware wait.

Root cause: `lastError`/`earliestRetryAfter`/`lastStatus` were declared with
`let` INSIDE the `for (setTry...)` loop body, so they reset to null at the
start of every set-try. When both targets lock out on setTry 0, every
subsequent setTry (1..maxSetRetries) pre-skips both targets via the
isModelLocked check with no real dispatch — so on the FINAL setTry (the only
one whose values the post-loop decision reads, since it's gated behind
`if (setTry < maxSetRetries) continue`), lastStatus was null, hitting the
"!lastStatus" branch (ALL_ACCOUNTS_INACTIVE 503) and completely bypassing the
comboCooldownWaitEnabled / earliestRetryAfter wait logic — even though a
real 429 with a known ~40s retry-after WAS observed on setTry 0.

This bug predates today's Gemini TPM work (any combo with maxSetRetries > 0
whose targets all lock out on the first pass was affected) but was masked in
existing tests: the "auto strategy (2 models...)" regression test uses
maxSetRetries: 0, so it only ever runs ONE setTry iteration and never
exercises the reset-on-retry path. It also explains why the dedicated
12-concurrent-request burst test passed cleanly — with concurrent requests,
timing variance meant some request's FINAL setTry iteration still had a live
target to dispatch to, giving lastStatus/earliestRetryAfter fresh data. A
single isolated request has no such luck.

Fix: hoist lastError/earliestRetryAfter/lastStatus to just inside
dispatchWithCooldownRetry, before the setTry loop, so they persist across
set-tries (still reset fresh on each recursive dispatchWithCooldownRetry()
call after a wait, which is correct). recordedAttempts/fallbackCount/
exhaustedProviders etc. are intentionally left per-iteration (unrelated to
this bug).

New regression test in tests/unit/combo-quota-share-cooldown-wait.test.ts
reproduces the exact live scenario (2 targets, both lock out on setTry 0,
maxSetRetries: 3) — confirmed red (503) against the pre-fix code, green
(200, waits and retries) against the fix.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* test(sse): extend live Gemini workload to Responses API + add large-context TPM test

Two additions to the live Gemini test suite, both live-verified against the
dev instance:

1. sendAndValidate() (tests/integration/liveGeminiShared.ts) now accepts an
   apiFormat: "chat" | "responses" parameter, building the Responses-API
   request shape (input array, max_output_tokens) and parsing its SSE events
   (response.output_text.delta / response.reasoning_summary_text.delta /
   response.completed) via the new readResponsesSSEStream(). Wired into two
   new tests in live-gemini-workload.test.ts ([30]/[31]), mirroring the
   existing Chat Completions streaming coverage. Verified live: 24/25 + 5/5
   payloads succeeded end-to-end through the new code path (the one failure
   was a ~300s test-client fetch timeout unrelated to the Responses API code
   itself — a separate, not-yet-addressed test-harness limitation).

2. genHugeContextMessage() builds a single message large enough (~4
   chars/token estimate) to approach or exceed Gemini's free-tier TPM ceiling
   (16000 input tokens/min for gemma-4) by itself. Every other prompt
   generator in this file tops out around 1-2k tokens — nowhere near that
   ceiling — so none of the existing workload tests ever exercised a REAL TPM
   429, only RPM-style rate limiting. tests/integration/gemini-large-context-tpm.test.ts
   sends two ~12-13k-token requests back-to-back (comfortably exceeding
   16000/min together) to exercise the full path against production Gemini:
   TPM classification, the comboCooldownWait retry, and the synthetic
   keep-alive frame on a genuinely slow request.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* fix(sse): abandoned combo target dispatch now observes its own per-target timeout, fixing a permanent "pending" dashboard leak

Live incident (dashboard log id 1784418258231-14961a, reported as "an ongoing
request even though there's already a 200"): a combo target dispatch
abandoned by comboTargetTimeoutMs (open-sse/services/combo/targetTimeoutRunner.ts)
left a permanent phantom "pending" entry in the dashboard, even after the
overall combo request had already succeeded via a different retry.

Root cause: chatCore.ts's createStreamController — and everything downstream
that depends on it (withRateLimit's Promise.race against Bottleneck,
acquireAccountSemaphore) — only ever watches clientRawRequest.signal, which
is the ORIGINAL client's request signal (set once via buildClientRawRequest
and reused unchanged across every target dispatch in a combo). It has no
connection to targetTimeoutRunner.ts's OWN AbortController
(target.modelAbortSignal), which is what actually fires when
comboTargetTimeoutMs (300s) elapses. src/sse/handlers/chat.ts's
handleSingleModel bridge between combo.ts and handleSingleModelChat received
`target.modelAbortSignal` but silently dropped it — never forwarded it
anywhere. So when a target got abandoned (e.g. stuck inside a wedged
Bottleneck rate-limiter queue, see the WEDGED force-reset log line from the
same incident), its per-target timeout fired and let the COMBO move on and
retry successfully elsewhere — but the abandoned dispatch's own promise
chain never learned it had been superseded, so it hung forever waiting on a
signal that was never going to fire, and trackPendingRequest(false) (the
finalize call) never ran.

Fix: thread target.modelAbortSignal through as a new modelAbortSignal
runtimeOption, and merge it into clientRawRequest.signal (via the existing
mergeAbortSignals helper from open-sse/executors/base.ts) right before
dispatch, so an abandoned target's own promise chain now observes its abort
and can reach its cleanup path — new resolveDispatchClientRawRequest() makes
this mechanically testable in isolation.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* fix(sse): combo cooldown-wait state recording, rate-limit wedge recovery, OpenAI-format SSE error frames

Five related fixes surfaced by live incidents (dashboard log ids 1784457764961-73,
1784465227489-a2cbc0, 1784504040241-6f8b9a) while validating the Gemini TPM/cooldown-wait
work on this branch against real OpenClaw traffic:

- combo.ts: the model-lockout bail-out branches in dispatchWithCooldownRetry never
  recorded lastStatus, so once every target in a set hit an existing lockout the final
  check crystallized a bogus ALL_ACCOUNTS_INACTIVE 503 instead of reaching the
  cooldown-wait decision, even with a real 429 + short retry-after observed.
- combo.ts/combo/types.ts: the "all credentials cooling down" pre-dispatch rejection
  (buildModelCooldownBody) nests its retry hint as error.retry_after/reset_seconds, not
  the top-level retryAfter every other 429 shape uses — combo's extraction only read the
  latter, so earliestRetryAfter stayed null for this shape even after lastStatus was fixed.
- rateLimitManager.ts: the wedge-recovery watchdog used disconnect(), which releases the
  heartbeat timer but never rejects jobs already QUEUED on that instance — orphaned
  dispatches hung until the outer ~300s per-target timeout, well past real clients'
  patience. Switched to stop({ dropWaitingJobs: true }), safe because the wedge condition
  already requires RUNNING===0 && EXECUTING===0.
- earlyStreamKeepalive.ts: the in-band error frame emitted after committing to a 200 SSE
  stream was hardcoded to Anthropic's `event: error` convention for every route, including
  the OpenAI-format ones (/v1/chat/completions, /v1/responses) where that framing is
  either invisible or malformed to a plain data-line parser. Added per-route
  OPENAI_CHAT_ERROR_FRAME / OPENAI_RESPONSES_ERROR_FRAME and wired them in.
- chatCore.ts: persisted a synthetic clientResponse error body even when the client had
  already disconnected (AbortError) before that body was ever computed — misleading the
  dashboard into showing "what the client received" for a response that was never sent.

Also: RequestLoggerDetail.tsx — Provider/Client Event Stream panes lost their collapse
toggle when StreamSection replaced the collapsible PayloadSection (692d6be80, unifying
active/finished request views) without carrying the toggle over.

Each fix has a TDD regression test with a confirmed red-before-green cycle.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* test(sse): free-tier model + gemma-4 TPM-ceiling benchmark harness

Adds a live benchmark comparing free models OmniRoute exposes across
configured providers plus previously-unexercised no-auth providers
(felo-web, aihorde, opencode, duckduckgo-web — none need a connection
row, they were just never tried). Reuses liveGeminiShared.ts's SSE
parsers and CASE_BUILDERS instead of duplicating them.

Also adds a targeted TPM-stress test firing back-to-back large-context
prompts at the gemma-4-31b model across its 3 free hosts (Gemini,
NVIDIA, AI Horde) to isolate whether the documented 16k-tokens/minute
free-tier ceiling is Gemini-specific enforcement or an inherent
model property.

FORCE_TOOL_CHOICE_REQUIRED is a test-only, default-off env flag added
to liveGeminiShared.ts and live-gemini-agentic-loop.test.ts for an
earlier live A/B comparison of tool_choice: required vs unset — kept
as a reusable knob for future runs.

Co-Authored-By: Markus Hartung <markus.hartung@gmail.com>

* test(sse): benchmark for the 2026-07-22 newly-enabled provider batch

Adds NEWLY_ENABLED_MODELS to freeModelBenchmarkShared.ts (Mistral
Leanstral, OpenRouter's live "free"-tagged roster, OpenCode Zen's
current free models — refetched live from
https://opencode.ai/zen/v1/models since the static catalog had
drifted) and a dedicated workload benchmark test for them.

Co-Authored-By: Markus Hartung <markus.hartung@gmail.com>

* test(sse): sync geminiRateLimitTracker tests with e74a1722b's corrected Gemma 4 limits

e74a1722b updated geminiRateLimits.json's gemma-4-* entries from the stale
15/1500/-1 (rpm/rpd/tpm) to the real published free-tier values
16000/14400/16000, but never updated the tests asserting the old numbers.
Surfaced by running the full test:unit suite as a post-rebase sanity check.

Co-Authored-By: Markus Hartung <markus.hartream@gmail.com>

* chore(quality): file-size baseline for own-growth (#8213)

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>
Co-authored-by: Markus Hartung <markus.hartream@gmail.com>
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
2026-07-23 05:17:35 -03:00

2769 lines
88 KiB
TypeScript
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// @ts-nocheck
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";
const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-chatcore-translation-"));
process.env.DATA_DIR = TEST_DATA_DIR;
const core = await import("../../src/lib/db/core.ts");
const providersDb = await import("../../src/lib/db/providers.ts");
const settingsDb = await import("../../src/lib/db/settings.ts");
const auth = await import("../../src/sse/services/auth.ts");
const upstreamProxyDb = await import("../../src/lib/db/upstreamProxy.ts");
const { invalidateCacheControlSettingsCache } =
await import("../../src/lib/cacheControlSettings.ts");
const { clearCache, getCachedResponse, generateSignature } =
await import("../../src/lib/semanticCache.ts");
const { clearIdempotency } = await import("../../src/lib/idempotencyLayer.ts");
const { getPendingRequests, clearPendingRequests } =
await import("../../src/lib/usage/usageHistory.ts");
const { clearInflight } = await import("../../open-sse/services/requestDedup.ts");
const {
buildAccountSemaphoreKey,
getStats: getAccountSemaphoreStats,
resetAll: resetAccountSemaphores,
} = await import("../../open-sse/services/accountSemaphore.ts");
const { getExecutor } = await import("../../open-sse/executors/index.ts");
const { clearModelLock, isModelLocked } =
await import("../../open-sse/services/accountFallback.ts");
const { saveModelsDevCapabilities, clearModelsDevCapabilities } =
await import("../../src/lib/modelsDevSync.ts");
const {
getBackgroundDegradationConfig,
setBackgroundDegradationConfig,
resetStats: resetBackgroundStats,
} = await import("../../open-sse/services/backgroundTaskDetector.ts");
const { getCallLogs, getCallLogById } = await import("../../src/lib/usage/callLogs.ts");
const {
handleChatCore,
shouldUseNativeCodexPassthrough,
isClaudeCodeSemanticPassthroughRequest,
isTokenExpiringSoon,
clearUpstreamProxyConfigCache,
buildStreamingResponseHeaders,
} = await import("../../open-sse/handlers/chatCore.ts");
const { resetPayloadRulesConfigForTests, setPayloadRulesConfig } =
await import("../../open-sse/services/payloadRules.ts");
const { FORMATS } = await import("../../open-sse/translator/formats.ts");
const { register, getRequestTranslator } = await import("../../open-sse/translator/registry.ts");
const originalFetch = globalThis.fetch;
const originalResponsesToOpenAI = getRequestTranslator(FORMATS.OPENAI_RESPONSES, FORMATS.OPENAI);
const originalSetTimeout = globalThis.setTimeout;
const originalBackgroundConfig = getBackgroundDegradationConfig();
const originalCallLogPipelineCaptureStreamChunks =
process.env.CALL_LOG_PIPELINE_CAPTURE_STREAM_CHUNKS;
function noopLog() {
return {
debug() {},
info() {},
warn() {},
error() {},
};
}
function restorePipelineCaptureEnv() {
if (originalCallLogPipelineCaptureStreamChunks === undefined) {
delete process.env.CALL_LOG_PIPELINE_CAPTURE_STREAM_CHUNKS;
} else {
process.env.CALL_LOG_PIPELINE_CAPTURE_STREAM_CHUNKS =
originalCallLogPipelineCaptureStreamChunks;
}
}
function toPlainHeaders(headers) {
if (!headers) return {};
if (headers instanceof Headers) return Object.fromEntries(headers.entries());
return Object.fromEntries(
Object.entries(headers).map(([key, value]) => [key, value == null ? "" : String(value)])
);
}
function buildOpenAIResponse(stream, text = "ok") {
if (stream) {
return new Response(
`data: ${JSON.stringify({
id: "chatcmpl-stream",
object: "chat.completion.chunk",
choices: [{ index: 0, delta: { role: "assistant", content: text } }],
})}\n\ndata: [DONE]\n\n`,
{
status: 200,
headers: { "Content-Type": "text/event-stream" },
}
);
}
return new Response(
JSON.stringify({
id: "chatcmpl-json",
object: "chat.completion",
model: "gpt-4o-mini",
choices: [
{
index: 0,
message: { role: "assistant", content: text },
finish_reason: "stop",
},
],
usage: {
prompt_tokens: 4,
completion_tokens: 2,
total_tokens: 6,
},
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
}
function buildClaudeResponse(stream, text = "ok") {
if (stream) {
return new Response(
[
"event: message_start",
`data: ${JSON.stringify({
type: "message_start",
message: {
id: "msg_stream",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
usage: { input_tokens: 12, output_tokens: 0 },
},
})}`,
"",
"event: content_block_start",
`data: ${JSON.stringify({
type: "content_block_start",
index: 0,
content_block: { type: "text", text: "" },
})}`,
"",
"event: content_block_delta",
`data: ${JSON.stringify({
type: "content_block_delta",
index: 0,
delta: { type: "text_delta", text },
})}`,
"",
"event: message_delta",
`data: ${JSON.stringify({
type: "message_delta",
delta: { stop_reason: "end_turn" },
usage: { output_tokens: 3 },
})}`,
"",
"event: message_stop",
`data: ${JSON.stringify({ type: "message_stop" })}`,
"",
].join("\n"),
{
status: 200,
headers: { "Content-Type": "text/event-stream" },
}
);
}
return new Response(
JSON.stringify({
id: "msg_json",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
content: [{ type: "text", text }],
stop_reason: "end_turn",
usage: {
input_tokens: 12,
output_tokens: 3,
},
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
}
function buildResponsesResponse(text = "ok") {
return new Response(
JSON.stringify({
id: "resp_123",
object: "response",
status: "completed",
model: "gpt-5.1-codex",
output: [
{
id: "msg_123",
type: "message",
role: "assistant",
content: [{ type: "output_text", text, annotations: [] }],
},
],
usage: {
input_tokens: 4,
output_tokens: 2,
total_tokens: 6,
},
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
}
function capabilityEntry(limitContext) {
return {
tool_call: true,
reasoning: false,
attachment: false,
structured_output: true,
temperature: true,
modalities_input: JSON.stringify(["text"]),
modalities_output: JSON.stringify(["text"]),
knowledge_cutoff: null,
release_date: null,
last_updated: null,
status: null,
family: null,
open_weights: false,
limit_context: limitContext,
limit_input: limitContext,
limit_output: 4096,
interleaved_field: null,
};
}
function hasCacheControl(value) {
if (!value || typeof value !== "object") return false;
if (Array.isArray(value)) {
return value.some((item) => hasCacheControl(item));
}
if (Object.hasOwn(value, "cache_control")) return true;
return Object.values(value).some((item) => hasCacheControl(item));
}
function collectTextBlocks(messages) {
if (!Array.isArray(messages)) return [];
return messages.flatMap((message) =>
Array.isArray(message.content) ? message.content.filter((block) => block?.type === "text") : []
);
}
async function resetStorage() {
clearUpstreamProxyConfigCache();
resetPayloadRulesConfigForTests();
register(FORMATS.OPENAI_RESPONSES, FORMATS.OPENAI, originalResponsesToOpenAI, null);
invalidateCacheControlSettingsCache();
clearCache();
clearIdempotency();
clearInflight();
clearModelsDevCapabilities();
setBackgroundDegradationConfig(originalBackgroundConfig);
resetBackgroundStats();
globalThis.setTimeout = originalSetTimeout;
core.resetDbInstance();
fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
fs.mkdirSync(TEST_DATA_DIR, { recursive: true });
}
// 30s ceiling: c8 instrumentation plus --test-concurrency=8 can stall CI workers
// well past the upstream timeout budget. Green runs return as soon as the condition
// holds, so the ceiling only bounds the failure case.
async function waitFor(fn, timeoutMs = 30000) {
const startedAt = Date.now();
while (Date.now() - startedAt < timeoutMs) {
const result = await fn();
if (result) return result;
await new Promise((resolve) => setTimeout(resolve, 25));
}
return null;
}
async function waitForAsyncSideEffects() {
await new Promise((resolve) => setImmediate(resolve));
await new Promise((resolve) => setTimeout(resolve, 10));
}
async function getLatestCallLog() {
const rows = await getCallLogs({ limit: 5 });
if (!Array.isArray(rows) || rows.length === 0) return null;
return getCallLogById(rows[0].id);
}
async function invokeChatCore({
body,
provider = "openai",
model = "gpt-4o-mini",
endpoint = "/v1/chat/completions",
accept = "application/json",
userAgent = "unit-test",
credentials,
apiKeyInfo = null,
responseFormat = "openai",
responseFactory,
isCombo = false,
comboStrategy = null,
requestHeaders = {},
connectionId = null,
onCredentialsRefreshed = null,
onRequestSuccess = null,
}: any = {}) {
const calls: any[] = [];
globalThis.fetch = async (url, init = {}) => {
const headers = toPlainHeaders(init.headers);
const captured = {
url: String(url),
method: init.method || "GET",
headers,
body: init.body ? JSON.parse(String(init.body)) : null,
};
calls.push(captured);
if (responseFactory) {
return responseFactory(captured, calls);
}
const upstreamStream = String(headers.Accept || headers.accept || "")
.toLowerCase()
.includes("text/event-stream");
if (responseFormat === "claude") return buildClaudeResponse(upstreamStream);
if (responseFormat === "openai-responses") return buildResponsesResponse();
return buildOpenAIResponse(upstreamStream);
};
try {
const requestBody = structuredClone(body);
const result = await handleChatCore({
body: requestBody,
modelInfo: { provider, model, extendedContext: false },
credentials: credentials || {
apiKey: "sk-test",
providerSpecificData: {},
},
log: noopLog(),
clientRawRequest: {
endpoint,
body: structuredClone(body),
headers: new Headers({ accept, ...requestHeaders }),
},
connectionId,
apiKeyInfo,
userAgent,
isCombo,
comboStrategy,
onCredentialsRefreshed,
onRequestSuccess,
} as any);
await waitForAsyncSideEffects();
return { result, calls, call: calls.at(-1) };
} finally {
globalThis.fetch = originalFetch;
}
}
test.afterEach(async () => {
globalThis.fetch = originalFetch;
restorePipelineCaptureEnv();
clearPendingRequests();
resetAccountSemaphores();
await waitForAsyncSideEffects();
await resetStorage();
});
test.after(async () => {
globalThis.fetch = originalFetch;
restorePipelineCaptureEnv();
clearPendingRequests();
resetAccountSemaphores();
await waitForAsyncSideEffects();
await resetStorage();
fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
});
test("chatCore times out upstream execution before provider response headers", async () => {
// This test asserts pendingDetail.providerRequest — only attached when the
// call-log pipeline capture is enabled. Declare the dependency explicitly
// (fresh-DB default leaves it off → the waitFor below would never resolve;
// failed deterministically on CI and on an isolated run, incl. at v3.8.18).
await settingsDb.updateSettings({ call_log_pipeline_enabled: true });
const executor = getExecutor("openai");
const originalGetTimeoutMs = executor.getTimeoutMs?.bind(executor);
executor.getTimeoutMs = () => 200;
const connectionId = "upstream-start-timeout";
const body = {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "never returns" }],
};
const fetchSignals: AbortSignal[] = [];
const upstreamBodies: any[] = [];
globalThis.fetch = async (_url, init = {}) => {
if (init.signal instanceof AbortSignal) fetchSignals.push(init.signal);
if (init.body) upstreamBodies.push(JSON.parse(String(init.body)));
return new Promise(() => {});
};
try {
const invocation = handleChatCore({
body: structuredClone(body),
modelInfo: { provider: "openai", model: "gpt-4o-mini", extendedContext: false },
credentials: {
apiKey: "sk-test",
providerSpecificData: {},
},
log: noopLog(),
clientRawRequest: {
endpoint: "/v1/chat/completions",
body: structuredClone(body),
headers: new Headers({ accept: "application/json" }),
},
connectionId,
userAgent: "unit-test",
} as any);
const pendingDetail = (await waitFor(() =>
// details[connectionId] is Record<modelKey, PendingRequestDetail[]> —
// the original predicate tested each ARRAY's .providerRequest (always
// undefined), so the waitFor could never resolve. Flatten to the details.
Object.values(getPendingRequests().details[connectionId] || {})
.flat()
.find((detail: any) => detail?.providerRequest?.model === "gpt-4o-mini")
)) as any;
assert.equal(pendingDetail?.providerRequest?.model, "gpt-4o-mini");
assert.deepEqual(pendingDetail?.providerRequest?.messages, body.messages);
const result = await invocation;
await waitForAsyncSideEffects();
assert.equal(upstreamBodies[0]?.model, "gpt-4o-mini");
assert.deepEqual(upstreamBodies[0]?.messages, body.messages);
assert.equal(result.success, false);
assert.equal(result.status, 504);
assert.equal(fetchSignals[0]?.aborted, true);
assert.equal(getPendingRequests().details[connectionId], undefined);
} finally {
if (originalGetTimeoutMs) executor.getTimeoutMs = originalGetTimeoutMs;
globalThis.fetch = originalFetch;
}
});
test("chatCore can disable pipeline stream chunk capture through environment", async () => {
process.env.CALL_LOG_PIPELINE_CAPTURE_STREAM_CHUNKS = "false";
await settingsDb.updateSettings({ call_log_pipeline_enabled: true });
const { result } = await invokeChatCore({
accept: "text/event-stream",
body: {
model: "gpt-4o-mini",
stream: true,
messages: [{ role: "user", content: "stream without chunk logging" }],
},
});
assert.equal(result.success, true);
await result.response.text();
await waitForAsyncSideEffects();
const detail = await waitFor(getLatestCallLog);
assert.ok(detail, "expected call log detail to be persisted");
assert.ok(detail.pipelinePayloads, "expected pipeline payloads when capture is enabled");
assert.equal((detail.pipelinePayloads as any).streamChunks, undefined);
});
test("chatCore keeps Responses-native Codex payloads in native passthrough mode", async () => {
const { call, result } = await invokeChatCore({
provider: "codex",
model: "gpt-5.1-codex",
endpoint: "/v1/responses",
credentials: { accessToken: "codex-token", providerSpecificData: {} },
body: {
model: "gpt-5.1-codex",
input: "ship it",
instructions: "custom system prompt",
store: true,
metadata: { source: "codex-client" },
stream: false,
},
responseFormat: "openai-responses",
});
assert.equal(result.success, true);
assert.match(call.url, /\/responses$/);
assert.deepEqual(call.body.input, [
{ type: "message", role: "user", content: [{ type: "input_text", text: "ship it" }] },
]);
assert.equal(call.body.instructions, "custom system prompt");
assert.equal(call.body.store, false);
assert.deepEqual(call.body.metadata, { source: "codex-client" });
assert.equal("messages" in call.body, false);
});
test("chatCore honors providerSpecificData.apiType for legacy openai-compatible providers", async () => {
const { call, result } = await invokeChatCore({
provider: "openai-compatible-sp-openai",
model: "gpt-5.4",
endpoint: "/v1/chat/completions",
credentials: {
apiKey: "sk-test",
providerSpecificData: {
apiType: "responses",
baseUrl: "https://proxy.example.com/v1",
prefix: "sp-openai",
},
},
body: {
model: "gpt-5.4",
stream: false,
messages: [{ role: "user", content: "Reply with OK only." }],
max_tokens: 64,
},
responseFormat: "openai-responses",
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.match(call.url, /\/responses$/);
assert.ok(call.body.input);
assert.equal("messages" in call.body, false);
assert.equal(payload.choices[0].message.content, "ok");
});
test("chatCore helper exports detect responses passthrough paths and token expiry windows", () => {
assert.equal(
shouldUseNativeCodexPassthrough({
provider: "codex",
sourceFormat: FORMATS.OPENAI_RESPONSES,
endpointPath: "/v1/responses///",
}),
true
);
assert.equal(
shouldUseNativeCodexPassthrough({
provider: "codex",
sourceFormat: FORMATS.OPENAI_RESPONSES,
endpointPath: "/v1/chat/completions",
}),
false
);
assert.equal(
isTokenExpiringSoon(new Date(Date.now() + 60_000).toISOString(), 5 * 60 * 1000),
true
);
assert.equal(
isTokenExpiringSoon(new Date(Date.now() + 10 * 60 * 1000).toISOString(), 5 * 60 * 1000),
false
);
assert.equal(isTokenExpiringSoon(null), false);
});
test("chatCore helper detects Claude Code semantic passthrough only for direct Claude-Code routes", () => {
assert.equal(
isClaudeCodeSemanticPassthroughRequest({
provider: "claude",
sourceFormat: FORMATS.CLAUDE,
targetFormat: FORMATS.CLAUDE,
userAgent: "claude-cli/2.1.137",
}),
true
);
assert.equal(
isClaudeCodeSemanticPassthroughRequest({
provider: "anthropic-compatible-cc-test",
sourceFormat: FORMATS.CLAUDE,
targetFormat: FORMATS.CLAUDE,
headers: new Headers({ "x-app": "cli" }),
userAgent: "unit-test",
}),
true
);
assert.equal(
isClaudeCodeSemanticPassthroughRequest({
provider: "anthropic-compatible-test",
sourceFormat: FORMATS.CLAUDE,
targetFormat: FORMATS.CLAUDE,
userAgent: "claude-cli/2.1.137",
}),
false
);
assert.equal(
isClaudeCodeSemanticPassthroughRequest({
provider: "claude",
sourceFormat: FORMATS.CLAUDE,
targetFormat: FORMATS.CLAUDE,
userAgent: "generic-client",
}),
false
);
});
test("chatCore applies payload rules after translating Responses input into Chat payloads", async () => {
setPayloadRulesConfig({
default: [
{
models: [{ name: "gpt-*", protocol: "openai" }],
params: {
"messages.0.metadata.routeTag": "feature-110",
},
},
],
override: [
{
models: [{ name: "gpt-*", protocol: "openai" }],
params: {
temperature: 0.25,
},
},
],
filter: [],
defaultRaw: [],
});
const { call, result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
endpoint: "/v1/responses",
body: {
model: "gpt-4o-mini",
stream: false,
input: [
{
type: "message",
role: "user",
content: [{ type: "input_text", text: "hello" }],
},
],
},
responseFormat: "openai",
});
assert.equal(result.success, true);
assert.equal(call.body.temperature, 0.25);
assert.equal(call.body.messages[0].metadata.routeTag, "feature-110");
assert.equal(call.body.messages[0].role, "user");
});
test("chatCore builds Claude Code-compatible upstream requests for CC providers", async () => {
const { call, result } = await invokeChatCore({
provider: "anthropic-compatible-cc-test",
model: "claude-sonnet-4-6",
endpoint: "/v1/chat/completions",
credentials: {
apiKey: "sk-test",
providerSpecificData: {
baseUrl: "https://proxy.example.com/v1/messages?beta=true",
chatPath: "/v1/messages?beta=true",
},
},
body: {
model: "claude-sonnet-4-6",
stream: false,
messages: [{ role: "user", content: "Ping" }],
},
responseFormat: "claude",
});
assert.equal(result.success, true);
assert.equal(call.headers.Accept ?? call.headers.accept, "text/event-stream");
assert.deepEqual([call.body.stream, call.body.context_management], [true, undefined]);
assert.equal(call.body.system.length, 1);
assert.match(call.body.system[0].text, /Claude Agent SDK/);
assert.equal(typeof call.body.metadata.user_id, "string");
assert.equal(call.body.messages[0].role, "user");
assert.equal(call.body.messages[0].content[0].text, "Ping");
});
// Fix #2468: normalizeClaudeUpstreamMessages() now runs on the pure Claude passthrough
// path too. It extracts role:"system" messages into the top-level system parameter,
// strips empty text blocks, converts inline document blocks (no url/data) to text, and
// drops unknown block types (e.g. future_block). tool_result blocks are preserved via
// preserveToolResultBlocks:true.
test("chatCore normalizes native Claude Code messages for native Claude OAuth passthrough", async () => {
const clientMessages = [
{
role: "system",
content: [{ type: "text", text: "system-message-that-should-stay-in-messages" }],
},
{
role: "user",
content: [
{ type: "text", text: "" },
{ type: "text", text: "Run pwd", cache_control: { type: "ephemeral" } },
{ type: "document", name: "README.md", content: "Do not flatten me" },
{ type: "future_block", payload: { keep: true } },
],
},
{
role: "assistant",
content: [{ type: "tool_use", id: "toolu_pwd", name: "Bash", input: { command: "pwd" } }],
},
{
role: "user",
content: [{ type: "tool_result", tool_use_id: "toolu_pwd", content: "ok" }],
},
];
const { call, result } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "omniroute/alias-that-should-resolve",
max_tokens: 64,
system: [{ type: "text", text: "top-level-system" }],
messages: clientMessages,
tools: [{ name: "Bash", input_schema: { type: "object", properties: {} } }],
},
userAgent: "claude-cli/2.1.137",
requestHeaders: { "x-app": "cli", "x-claude-code-session-id": "session-123" },
responseFormat: "claude",
});
assert.equal(result.success, true);
assert.equal(call.body.model, "claude-sonnet-4-6");
// After normalization: role:"system" msg extracted → top-level system (3 msgs remain, not 4)
assert.equal(call.body.messages.length, 3);
// system-role block appended to top-level system array
assert.equal(
call.body.system.some(
(block: { text?: string }) => block.text === "system-message-that-should-stay-in-messages"
),
true
);
// user msg[0] (was clientMessages[1]): empty text, document and future_block are preserved
// since it is a semantic passthrough request
assert.equal(call.body.messages[0].content.length, 4);
assert.equal(call.body.messages[0].content[0].type, "text");
assert.equal(call.body.messages[0].content[0].text, "");
assert.equal(call.body.messages[0].content[1].text, "Run pwd");
assert.equal(call.body.messages[0].content[2].type, "document");
assert.equal(call.body.messages[0].content[3].type, "future_block");
// assistant msg[1] (was clientMessages[2]): tool_use unchanged
assert.equal(call.body.messages[1].content[0].type, "tool_use");
// user msg[2] (was clientMessages[3]): tool_result preserved (preserveToolResultBlocks:true)
assert.equal(call.body.messages[2].content[0].type, "tool_result");
});
test("chatCore keeps Claude normalization for non-Claude-Code Claude passthrough", async () => {
const { call, result } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
messages: [
{ role: "system", content: "system role should move" },
{
role: "user",
content: [
{ type: "text", text: "" },
{ type: "text", text: "hello" },
{ type: "document", name: "README.md", content: "Read me" },
{ type: "future_block", payload: { drop: true } },
],
},
],
},
userAgent: "generic-client/1.0",
responseFormat: "claude",
});
assert.equal(result.success, true);
assert.equal(
call.body.messages.some((message) => message.role === "system"),
false
);
assert.equal(call.body.system.at(-1).text, "system role should move");
assert.deepEqual(call.body.messages[0].content, [
{ type: "text", text: "hello" },
{ type: "text", text: "[README.md]\nRead me" },
]);
});
// Fix #2468: normalizeClaudeUpstreamMessages() runs on the CC-compatible bridge path too
// (preserveClaudeMessages=true). Same normalization: system-role → top-level system,
// empty text stripped, document→text, future_block dropped, tool_result preserved.
test("chatCore normalizes native Claude Code messages before CC-compatible relay transforms", async () => {
const clientMessages = [
{
role: "system",
content: [{ type: "text", text: "system-message-remains-in-source-history" }],
},
{
role: "user",
content: [
{ type: "text", text: "" },
{ type: "text", text: "Inspect project", cache_control: { type: "ephemeral" } },
{ type: "document", name: "design.md", content: "Keep as document block" },
{ type: "future_block", payload: { keep: true } },
],
},
{
role: "assistant",
content: [{ type: "tool_use", id: "toolu_read", name: "Read", input: { file_path: "a.ts" } }],
},
{
role: "user",
content: [{ type: "tool_result", tool_use_id: "toolu_read", content: "file contents" }],
},
];
const { call, result } = await invokeChatCore({
provider: "anthropic-compatible-cc-test",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: {
apiKey: "sk-test",
providerSpecificData: {
baseUrl: "https://proxy.example.com/v1/messages?beta=true",
chatPath: "/v1/messages?beta=true",
},
},
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "top-level-system" }],
messages: clientMessages,
tools: [{ name: "Read", input_schema: { type: "object", properties: {} } }],
},
userAgent: "Claude-Code/2.1.137",
requestHeaders: { "x-app": "cli", "x-claude-code-session-id": "cc-session-123" },
responseFormat: "claude",
});
assert.equal(result.success, true);
assert.match(call.url, /\/v1\/messages\?beta=true$/);
assert.equal(call.body.stream, true);
// After normalization: role:"system" msg extracted → top-level system (3 msgs remain, not 4)
assert.equal(call.body.messages.length, 3);
// CC bridge prepends its own system block; extracted system block is appended after it
assert.equal(
call.body.system[0].text,
"You are a Claude agent, built on Anthropic's Claude Agent SDK."
);
assert.equal(
call.body.system.some(
(block: { text?: string }) => block.text === "system-message-remains-in-source-history"
),
true
);
// user msg[0] (was clientMessages[1]): empty text, document and future_block are preserved
// since it is a semantic passthrough request
assert.equal(call.body.messages[0].content.length, 4);
assert.equal(call.body.messages[0].content[0].type, "text");
assert.equal(call.body.messages[0].content[0].text, "");
assert.equal(call.body.messages[0].content[1].text, "Inspect project");
assert.equal(call.body.messages[0].content[2].type, "document");
assert.equal(call.body.messages[0].content[3].type, "future_block");
// assistant msg[1] (was clientMessages[2]): tool_use unchanged
assert.equal(call.body.messages[1].content[0].type, "tool_use");
// user msg[2] (was clientMessages[3]): tool_result preserved (preserveToolResultBlocks:true)
assert.equal(call.body.messages[2].content[0].type, "tool_result");
});
test("chatCore preserves cache_control automatically for Claude Code single-model requests", async () => {
await settingsDb.updateSettings({ alwaysPreserveClientCache: "auto" });
invalidateCacheControlSettingsCache();
const claudeBody = {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "system", cache_control: { type: "ephemeral", ttl: "5m" } }],
messages: [
{
role: "user",
content: [{ type: "text", text: "u1", cache_control: { type: "ephemeral" } }],
},
{
role: "assistant",
content: [{ type: "text", text: "a1", cache_control: { type: "ephemeral", ttl: "10m" } }],
},
{ role: "user", content: [{ type: "text", text: "u2" }] },
],
tools: [
{
name: "lookup_weather",
description: "Fetch weather",
input_schema: { type: "object" },
cache_control: { type: "ephemeral", ttl: "30m" },
},
],
};
const { call } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: claudeBody,
userAgent: "Claude-Code/1.0.0",
responseFormat: "claude",
});
assert.equal(hasCacheControl(call.body), true);
// system[0] and system[1] are now the billing line and sentinel injected by base.ts for Claude Code
assert.deepEqual(call.body.system[2].cache_control, { type: "ephemeral", ttl: "5m" });
assert.deepEqual(call.body.messages[0].content[0].cache_control, { type: "ephemeral" });
// base.ts executor explicitly strips cache_control from tools for Claude Code clients
assert.equal(call.body.tools[0].cache_control, undefined);
});
test("chatCore auto cache policy becomes false for nondeterministic combos", async () => {
await settingsDb.updateSettings({ alwaysPreserveClientCache: "auto" });
invalidateCacheControlSettingsCache();
const { call } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "system", cache_control: { type: "ephemeral", ttl: "5m" } }],
messages: [{ role: "user", content: [{ type: "text", text: "u1" }] }],
},
userAgent: "Claude-Code/1.0.0",
isCombo: true,
comboStrategy: "latency-optimized",
responseFormat: "claude",
});
assert.equal(
call.body.system.some(
(block: { type?: string; text?: string }) => block?.type === "text" && block.text === "system"
),
true
);
// Cache markers are kept natively due to the latest Claude strict proxy passthrough implementation
assert.equal(
call.body.system.some((block) => !!block.cache_control),
true
);
});
test("chatCore always-preserve mode keeps cache_control even without Claude Code user-agent", async () => {
await settingsDb.updateSettings({ alwaysPreserveClientCache: "always" });
invalidateCacheControlSettingsCache();
const { call } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "system", cache_control: { type: "ephemeral", ttl: "5m" } }],
messages: [{ role: "user", content: [{ type: "text", text: "u1" }] }],
},
responseFormat: "claude",
});
assert.equal(hasCacheControl(call.body), true);
assert.deepEqual(call.body.system[0].cache_control, { type: "ephemeral", ttl: "5m" });
});
test("chatCore disables raw Claude passthrough when cache preservation is off and normalizes through OpenAI", async () => {
await settingsDb.updateSettings({ alwaysPreserveClientCache: "never" });
invalidateCacheControlSettingsCache();
const { call } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "system", cache_control: { type: "ephemeral", ttl: "5m" } }],
messages: [
{
role: "user",
content: [{ type: "text", text: "u1", cache_control: { type: "ephemeral" } }],
},
],
},
userAgent: "Claude-Code/1.0.0",
responseFormat: "claude",
});
assert.equal(
call.body.system.some(
(block: { type?: string; text?: string }) => block?.type === "text" && block.text === "system"
),
true
);
// Cache preservation is on for native Claude, so cache markers are intact
assert.deepEqual(call.body.messages[0].content[0].cache_control, { type: "ephemeral" });
// Tools disable flag is applied
assert.equal("_disableToolPrefix" in call.body, false);
});
test("chatCore default translation converts Claude requests to OpenAI and strips cache markers for non-Claude providers", async () => {
const { call } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
endpoint: "/v1/messages",
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "system", cache_control: { type: "ephemeral", ttl: "5m" } }],
messages: [
{
role: "user",
content: [{ type: "text", text: "u1", cache_control: { type: "ephemeral" } }],
},
],
},
userAgent: "Claude-Code/1.0.0",
responseFormat: "openai",
});
assert.equal(call.body.model, "gpt-4o-mini");
assert.equal(Array.isArray(call.body.messages), true);
assert.equal(call.body.messages[0].role, "system");
assert.equal(JSON.stringify(call.body).includes("cache_control"), false);
});
test("chatCore sets Claude tool prefix disabling, strips empty Anthropic text blocks, and cleans helper flags", async () => {
const { call } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/chat/completions",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "ignored-client-model",
_toolNameMap: new Map([["proxy_Bash", "Bash"]]),
messages: [
{
role: "user",
content: [
{ type: "text", text: "" },
{ type: "text", text: "hello" },
],
},
],
tools: [
{
type: "function",
function: {
name: "Bash",
description: "Execute bash",
parameters: { type: "object" },
},
},
],
},
responseFormat: "claude",
});
assert.equal(call.body.model, "claude-sonnet-4-6");
assert.equal(call.body.tools[0].name, "Bash");
assert.equal(call.body.tools[0].name.startsWith("proxy_"), false);
assert.equal(call.body._toolNameMap, undefined);
assert.equal(call.body._disableToolPrefix, undefined);
assert.deepEqual(
collectTextBlocks(call.body.messages).map((block) => block.text),
["hello"]
);
});
test("chatCore restores prefixed Claude passthrough tool names in upstream responses", async () => {
const { result } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
body: {
model: "claude-sonnet-4-6",
messages: [{ role: "user", content: [{ type: "text", text: "run bash" }] }],
tools: [
{
name: "Bash",
description: "Execute bash",
input_schema: { type: "object" },
},
],
},
responseFormat: "claude",
responseFactory() {
return new Response(
JSON.stringify({
id: "msg_tool_use",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
content: [
{
type: "tool_use",
id: "toolu_1",
name: "proxy_Bash",
input: { command: "ls" },
},
],
stop_reason: "tool_use",
usage: {
input_tokens: 4,
output_tokens: 2,
},
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(payload.content[0].name, "Bash");
});
test("chatCore strips unsupported reasoning params and caps provider token fields", async () => {
const { call } = await invokeChatCore({
provider: "openai",
model: "o3",
endpoint: "/v1/chat/completions",
body: {
model: "o3",
messages: [{ role: "user", content: "hello" }],
temperature: 0.7,
presence_penalty: 1,
max_tokens: 99999,
max_completion_tokens: 77777,
},
responseFormat: "openai",
});
assert.equal(call.body.temperature, undefined);
assert.equal(call.body.presence_penalty, undefined);
assert.equal(call.body.max_tokens, undefined);
assert.equal(call.body.max_completion_tokens, 16384);
});
test("chatCore preserves reasoning_effort for assistant-prefill OpenAI-compatible requests", async () => {
const { call, result } = await invokeChatCore({
provider: "openai-compatible-aio",
model: "glm-5.1",
endpoint: "/v1/chat/completions",
body: {
model: "aio/glm-5.1",
messages: [
{ role: "user", content: "draft the answer" },
{ role: "assistant", content: "<thinking>" },
],
reasoning_effort: "xhigh",
stream: true,
},
responseFormat: "openai",
});
assert.equal(result.success, true);
assert.equal(call.body.model, "glm-5.1");
assert.equal(call.body.reasoning_effort, "xhigh");
});
test("chatCore logs chat completions endpoint as OpenAI protocol", async () => {
const { call, result } = await invokeChatCore({
provider: "openrouter",
model: "deepseek/deepseek-v4-pro",
endpoint: "/v1/chat/completions",
body: {
model: "openrouter/deepseek/deepseek-v4-pro",
messages: [{ role: "user", content: "Human: Hi" }],
temperature: 1,
max_tokens: 64000,
stream: false,
presence_penalty: 0,
frequency_penalty: 0,
top_p: 0.9,
},
responseFormat: "openai",
});
assert.equal(result.success, true);
assert.equal(call.body.model, "deepseek/deepseek-v4-pro");
const logEntry = await waitFor(getLatestCallLog);
assert.ok(logEntry, "expected call log to be persisted");
assert.equal(logEntry.path, "/v1/chat/completions");
assert.equal(logEntry.sourceFormat, FORMATS.OPENAI);
});
test("chatCore surfaces translation errors with explicit status codes", async () => {
register(
FORMATS.OPENAI_RESPONSES,
FORMATS.OPENAI,
() => {
const error = new Error("responses translator rejected the payload");
error.statusCode = 409;
throw error;
},
null
);
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
endpoint: "/v1/responses",
body: {
model: "gpt-4o-mini",
input: "hello",
},
});
assert.equal(result.success, false);
assert.equal(result.status, 409);
assert.equal(result.error, "responses translator rejected the payload");
});
test("chatCore surfaces typed translation errors with the declared error type", async () => {
register(
FORMATS.OPENAI_RESPONSES,
FORMATS.OPENAI,
() => {
const error = new Error("typed translator failure");
error.statusCode = 422;
error.errorType = "unsupported_feature";
throw error;
},
null
);
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
endpoint: "/v1/responses",
body: {
model: "gpt-4o-mini",
input: "hello",
},
});
assert.equal(result.success, false);
assert.equal(result.status, 422);
const payload = (await result.response.json()) as any;
assert.equal(payload.error.type, "unsupported_feature");
assert.equal(payload.error.code, "unsupported_feature");
});
test("chatCore returns 500 when translation throws a generic error", async () => {
register(
FORMATS.OPENAI_RESPONSES,
FORMATS.OPENAI,
() => {
throw new Error("unexpected translator crash");
},
null
);
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
endpoint: "/v1/responses",
body: {
model: "gpt-4o-mini",
input: "hello",
},
});
assert.equal(result.success, false);
assert.equal(result.status, 500);
assert.equal(result.error, "unexpected translator crash");
});
test("chatCore refreshes GitHub credentials after 401 and retries with the refreshed Copilot token", async () => {
let refreshedCredentials = null;
const { calls, result } = await invokeChatCore({
provider: "github",
model: "gpt-4o-mini",
credentials: {
accessToken: "gh-access-token",
refreshToken: "gh-refresh-token",
providerSpecificData: {},
},
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "retry after auth refresh" }],
},
onCredentialsRefreshed(updated) {
refreshedCredentials = updated;
},
responseFactory(captured, seenCalls) {
if (captured.url.startsWith("https://api.github.com/copilot_internal/v2/token")) {
return new Response(
JSON.stringify({
token: "copilot-refreshed-token",
expires_at: Math.floor(Date.now() / 1000) + 3600,
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
}
const providerCalls = seenCalls.filter((entry) =>
entry.url.startsWith("https://api.githubcopilot.com/")
);
if (providerCalls.length === 1) {
return new Response(
JSON.stringify({
error: { message: "token expired" },
}),
{
status: 401,
headers: { "Content-Type": "application/json" },
}
);
}
return buildOpenAIResponse(false, "retry succeeded after refresh");
},
});
const payload = (await result.response.json()) as any;
const providerCalls = calls.filter((entry) =>
entry.url.startsWith("https://api.githubcopilot.com/")
);
assert.equal(result.success, true);
assert.equal(providerCalls.length, 2);
assert.equal(
providerCalls[1].headers.authorization ?? providerCalls[1].headers.Authorization,
"Bearer copilot-refreshed-token"
);
assert.equal(refreshedCredentials?.providerSpecificData?.copilotToken, "copilot-refreshed-token");
assert.equal(payload.choices[0].message.content, "retry succeeded after refresh");
});
test("chatCore uses the native executor when no upstream proxy mode is enabled", async () => {
const { call } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
messages: [{ role: "user", content: "hello" }],
},
responseFormat: "openai",
});
assert.match(call.url, /^https:\/\/api\.openai\.com\/v1\/chat\/completions$/);
});
test("chatCore routes providers through CLIProxyAPI in passthrough mode", async () => {
await upstreamProxyDb.upsertUpstreamProxyConfig({
providerId: "qoder",
mode: "cliproxyapi",
enabled: true,
});
const { call } = await invokeChatCore({
provider: "qoder",
model: "qoder-rome-30ba3b",
credentials: { apiKey: "qoder-token", providerSpecificData: {} },
body: {
model: "qoder-rome-30ba3b",
messages: [{ role: "user", content: "hello" }],
},
responseFormat: "openai",
});
assert.match(call.url, /^http:\/\/127\.0\.0\.1:8317\/v1\/chat\/completions$/);
assert.equal(call.headers.Authorization ?? call.headers.authorization, "Bearer qoder-token");
});
test("chatCore fallback proxy mode retries through CLIProxyAPI after retryable native failures", async () => {
await upstreamProxyDb.upsertUpstreamProxyConfig({
providerId: "github",
mode: "fallback",
enabled: true,
});
clearUpstreamProxyConfigCache("github");
const { calls, result } = await invokeChatCore({
provider: "github",
model: "gpt-4o",
credentials: {
accessToken: "gh-token",
providerSpecificData: {
copilotToken: "mock-token",
copilotTokenExpiresAt: Date.now() + 3600000,
},
},
body: {
model: "gpt-4o",
messages: [{ role: "user", content: "hello" }],
},
responseFormat: "openai",
responseFactory(captured, seenCalls) {
if (seenCalls.length === 1) {
return new Response(JSON.stringify({ error: { message: "native failed" } }), {
status: 500,
headers: { "Content-Type": "application/json" },
});
}
assert.match(captured.url, /^http:\/\/127\.0\.0\.1:8317\/v1\/chat\/completions$/);
return buildOpenAIResponse(false, "retried");
},
});
assert.equal(result.success, true);
assert.equal(calls.length, 2);
assert.match(calls[0].url, /^https:\/\/api\.githubcopilot\.com\/chat\/completions$/);
assert.match(calls[1].url, /^http:\/\/127\.0\.0\.1:8317\/v1\/chat\/completions$/);
});
test("chatCore fallback proxy mode surfaces CLIProxyAPI errors after a retryable native status", async () => {
await upstreamProxyDb.upsertUpstreamProxyConfig({
providerId: "github",
mode: "fallback",
enabled: true,
});
clearUpstreamProxyConfigCache("github");
const { calls, result } = await invokeChatCore({
provider: "github",
model: "gpt-4o",
credentials: {
accessToken: "gh-token",
providerSpecificData: {
copilotToken: "mock-token",
copilotTokenExpiresAt: Date.now() + 3600000,
},
},
body: {
model: "gpt-4o",
messages: [{ role: "user", content: "hello" }],
},
responseFormat: "openai",
responseFactory(captured, seenCalls) {
if (seenCalls.length === 1) {
return new Response(JSON.stringify({ error: { message: "native failed" } }), {
status: 500,
headers: { "Content-Type": "application/json" },
});
}
assert.match(captured.url, /^http:\/\/127\.0\.0\.1:8317\/v1\/chat\/completions$/);
throw new Error("cliproxy retry failed");
},
});
assert.equal(calls.length, 2);
assert.equal(result.success, false);
assert.equal(result.status, 502);
assert.equal(result.error, "[502]: cliproxy retry failed");
});
test("chatCore fallback proxy mode surfaces CLIProxyAPI errors after native executor throws", async () => {
await upstreamProxyDb.upsertUpstreamProxyConfig({
providerId: "github",
mode: "fallback",
enabled: true,
});
clearUpstreamProxyConfigCache("github");
const { calls, result } = await invokeChatCore({
provider: "github",
model: "gpt-4o",
credentials: {
accessToken: "gh-token",
providerSpecificData: {
copilotToken: "mock-token",
copilotTokenExpiresAt: Date.now() + 3600000,
},
},
body: {
model: "gpt-4o",
messages: [{ role: "user", content: "hello" }],
},
responseFormat: "openai",
responseFactory(captured, seenCalls) {
if (seenCalls.length === 1) {
throw new Error("native transport exploded");
}
assert.match(captured.url, /^http:\/\/127\.0\.0\.1:8317\/v1\/chat\/completions$/);
throw new Error("cliproxy transport exploded");
},
});
assert.equal(calls.length, 2);
assert.equal(result.success, false);
assert.equal(result.status, 502);
assert.equal(result.error, "[502]: cliproxy transport exploded");
});
test("chatCore serves a cached idempotent response without hitting the provider twice", async () => {
const sharedHeaders = { "idempotency-key": "unit-idempotent-key" };
const first = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
requestHeaders: sharedHeaders,
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "repeat this safely" }],
},
responseFormat: "openai",
});
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
requestHeaders: sharedHeaders,
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "repeat this safely" }],
},
responseFormat: "openai",
});
assert.equal(first.calls.length, 1);
assert.equal(second.calls.length, 0);
assert.equal(second.result.success, true);
assert.equal(second.result.response.headers.get("X-OmniRoute-Idempotent"), "true");
const payload = (await second.result.response.json()) as any;
assert.equal(payload.choices[0].message.content, "ok");
});
test("chatCore returns a semantic cache HIT for repeated deterministic requests", async () => {
let upstreamHits = 0;
const sharedBody = {
model: "gpt-4o-mini",
stream: false,
temperature: 0,
messages: [{ role: "user", content: "cache this exact answer" }],
};
const first = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(false, "cached-once");
},
});
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(false, "should-not-run");
},
});
assert.equal(first.calls.length, 1);
assert.equal(first.result.response.headers.get("X-OmniRoute-Cache"), "MISS");
assert.equal(second.calls.length, 0);
assert.equal(second.result.response.headers.get("X-OmniRoute-Cache"), "HIT");
assert.equal(upstreamHits, 1);
const payload = (await second.result.response.json()) as any;
assert.equal(payload.choices[0].message.content, "cached-once");
await waitForAsyncSideEffects();
const semanticLog = await waitFor(async () => {
const rows = await getCallLogs({ limit: 10 });
const hit = rows.find((row) => row.cacheSource === "semantic");
if (!hit) return null;
return await getCallLogById(hit.id);
});
assert.ok(semanticLog, "expected semantic cache HIT to be persisted in call logs");
assert.equal(semanticLog.cacheSource, "semantic");
assert.equal(semanticLog.path, "/v1/chat/completions");
assert.equal(semanticLog.status, 200);
});
test("chatCore skips semantic cache when disabled in settings", async () => {
await settingsDb.updateSettings({ semanticCacheEnabled: false });
let upstreamHits = 0;
const sharedBody = {
model: "gpt-4o-mini",
stream: false,
temperature: 0,
messages: [{ role: "user", content: "do not reuse this response locally" }],
};
const first = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(false, `fresh-${upstreamHits}`);
},
});
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(false, `fresh-${upstreamHits}`);
},
});
assert.equal(first.calls.length, 1);
assert.equal(second.calls.length, 1);
assert.equal(upstreamHits, 2);
assert.equal(first.result.response.headers.get("X-OmniRoute-Cache"), "MISS");
assert.equal(second.result.response.headers.get("X-OmniRoute-Cache"), "MISS");
const payload = (await second.result.response.json()) as any;
assert.equal(payload.choices[0].message.content, "fresh-2");
});
test("chatCore attaches OmniRoute response metadata headers to non-stream responses", async () => {
const { result } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
body: {
model: "claude-sonnet-4-6",
stream: false,
messages: [{ role: "user", content: "header metadata" }],
},
responseFormat: "claude",
});
assert.equal(result.success, true);
assert.equal(result.response.headers.get("X-OmniRoute-Provider"), "cc");
assert.equal(result.response.headers.get("X-OmniRoute-Model"), "claude-sonnet-4-6");
assert.equal(result.response.headers.get("X-OmniRoute-Cache-Hit"), "false");
assert.equal(result.response.headers.get("X-OmniRoute-Tokens-In"), "12");
assert.equal(result.response.headers.get("X-OmniRoute-Tokens-Out"), "3");
assert.ok(Number(result.response.headers.get("X-OmniRoute-Latency-Ms")) >= 0);
assert.match(String(result.response.headers.get("X-OmniRoute-Response-Cost")), /^\d+\.\d{10}$/);
});
test("chatCore does not expose provider request credentials in non-stream response headers", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "hide provider credentials" }],
},
responseFormat: "openai",
});
assert.equal(result.success, true);
assert.equal(result.response.headers.get("authorization"), null);
assert.equal(result.response.headers.get("x-api-key"), null);
assert.equal(result.response.headers.get("Content-Type"), "application/json");
assert.equal(result.response.headers.get("X-OmniRoute-Cache"), "MISS");
});
test("chatCore normalizes tool finish reasons and estimates usage when upstream omits it", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "call the tool" }],
},
responseFormat: "openai",
responseFactory() {
return new Response(
JSON.stringify({
id: "chatcmpl_tool_no_usage",
object: "chat.completion",
model: "gpt-4o-mini",
choices: [
{
index: 0,
message: {
role: "assistant",
content: "",
tool_calls: [
{
id: "call_1",
type: "function",
function: {
name: "lookup_weather",
arguments: '{"city":"Sao Paulo"}',
},
},
],
},
finish_reason: "stop",
},
],
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(payload.choices[0].finish_reason, "tool_calls");
assert.ok(payload.usage.total_tokens > 0);
assert.ok(payload.usage.prompt_tokens > 0);
});
test("chatCore bypasses Claude CLI warmup probes before touching the provider", async () => {
const { calls, result } = await invokeChatCore({
model: "gpt-5",
userAgent: "claude-cli/2.1.89",
body: {
model: "gpt-5",
stream: false,
messages: [{ role: "user", content: [{ type: "text", text: "Warmup" }] }],
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(calls.length, 0);
assert.match(payload.choices[0].message.content, /CLI Command Execution/);
});
test("chatCore redirects background utility tasks to a cheaper mapped model", async () => {
setBackgroundDegradationConfig({
enabled: true,
degradationMap: {
...originalBackgroundConfig.degradationMap,
"gpt-5": "gpt-5-mini",
},
detectionPatterns: ["generate a title"],
});
const { call, result } = await invokeChatCore({
provider: "openai",
model: "gpt-5",
body: {
model: "gpt-5",
max_tokens: 16,
messages: [
{ role: "system", content: "Generate a title for the conversation." },
{ role: "user", content: "Discuss release notes" },
],
},
});
assert.equal(result.success, true);
assert.equal(call.body.model, "gpt-5-mini");
});
test("chatCore preserves Codex dual-window scope cooldowns on 429 responses", async () => {
const connection = await providersDb.createProviderConnection({
provider: "codex",
authType: "oauth",
email: "codex@example.com",
accessToken: "codex-token",
isActive: true,
providerSpecificData: {},
});
const resetAt5h = new Date(Date.now() + 60_000).toISOString();
const resetAt7d = new Date(Date.now() + 3_600_000).toISOString();
const { result } = await invokeChatCore({
provider: "codex",
model: "gpt-5.1-codex",
endpoint: "/v1/responses",
connectionId: connection.id,
credentials: {
accessToken: "codex-token",
providerSpecificData: {},
},
body: {
model: "gpt-5.1-codex",
input: "persist quota",
stream: false,
},
responseFactory() {
return new Response(JSON.stringify({ error: { message: "Codex quota exceeded" } }), {
status: 429,
headers: {
"Content-Type": "application/json",
"x-codex-5h-usage": "95",
"x-codex-5h-limit": "100",
"x-codex-5h-reset-at": resetAt5h,
"x-codex-7d-usage": "100",
"x-codex-7d-limit": "1000",
"x-codex-7d-reset-at": resetAt7d,
},
});
},
});
const updated = await providersDb.getProviderConnectionById((connection as any).id);
assert.equal(result.success, false);
assert.equal(result.status, 429);
assert.equal((updated as any).providerSpecificData.codexQuotaState.limit5h, 100);
assert.equal((updated as any).providerSpecificData.codexQuotaState.scope, "codex");
assert.equal(
typeof (updated as any).providerSpecificData.codexScopeRateLimitedUntil.codex,
"string"
);
assert.equal((updated as any).providerSpecificData.codexExhaustedWindow, "5h");
});
test("chatCore 429 lets account fallback apply the configured resilience cooldown", async () => {
await settingsDb.updateSettings({
resilienceSettings: {
connectionCooldown: {
apikey: {
baseCooldownMs: 1000,
useUpstreamRetryHints: false,
maxBackoffSteps: 3,
},
},
},
});
const connection = await providersDb.createProviderConnection({
provider: "openai",
authType: "apikey",
name: "resilience-429",
apiKey: "sk-resilience-429",
isActive: true,
providerSpecificData: {},
});
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
connectionId: connection.id,
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "rate limit me" }],
},
responseFactory() {
return new Response(JSON.stringify({ error: { message: "too many requests" } }), {
status: 429,
headers: { "Content-Type": "application/json" },
});
},
});
const afterCore = await providersDb.getProviderConnectionById((connection as any).id);
assert.equal(result.success, false);
assert.equal(result.status, 429);
assert.equal((afterCore as any).rateLimitedUntil, undefined);
const fallback = await auth.markAccountUnavailable(
(connection as any).id,
result.status,
result.error,
"openai",
"gpt-4o-mini"
);
const afterFallback = await providersDb.getProviderConnectionById((connection as any).id);
const cooldownRemaining =
new Date((afterFallback as any).rateLimitedUntil).getTime() - Date.now();
assert.equal(fallback.shouldFallback, true);
assert.equal(fallback.cooldownMs, 1000);
assert.equal((afterFallback as any).testStatus, "unavailable");
assert.ok(cooldownRemaining > 0 && cooldownRemaining <= 2_000);
});
test("chatCore falls back to the next family model when the requested model is unavailable", async () => {
const { calls, result } = await invokeChatCore({
provider: "openai",
model: "gpt-5.1",
body: {
model: "gpt-5.1",
stream: false,
messages: [{ role: "user", content: "fallback on model unavailable" }],
},
responseFactory(_captured, seenCalls) {
if (seenCalls.length === 1) {
return new Response(JSON.stringify({ error: { message: "model not found" } }), {
status: 404,
headers: { "Content-Type": "application/json" },
});
}
return buildOpenAIResponse(false, "family fallback ok");
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(calls.length, 2);
assert.equal(calls[1].body.model, "gpt-5.1-mini");
assert.equal(payload.choices[0].message.content, "family fallback ok");
});
test("chatCore falls back to a larger-context sibling when the request overflows context", async () => {
saveModelsDevCapabilities({
unknown: {
"gpt-5": capabilityEntry(128_000),
"gpt-5-mini": capabilityEntry(64_000),
"gpt-4o": capabilityEntry(256_000),
},
});
const { calls, result } = await invokeChatCore({
provider: "openai",
model: "gpt-5",
body: {
model: "gpt-5",
stream: false,
messages: [{ role: "user", content: "recover from context overflow" }],
},
responseFactory(_captured, seenCalls) {
if (seenCalls.length === 1) {
return new Response(JSON.stringify({ error: { message: "maximum context exceeded" } }), {
status: 400,
headers: { "Content-Type": "application/json" },
});
}
return buildOpenAIResponse(false, "larger context fallback");
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(calls.length, 2);
assert.equal(calls[1].body.model, "gpt-4o");
assert.equal(payload.choices[0].message.content, "larger context fallback");
});
test("chatCore parses upstream SSE payloads for non-streaming requests", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "parse sse" }],
},
responseFactory() {
return buildOpenAIResponse(true, "sse json");
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(payload.choices[0].message.content, "sse json");
});
test("chatCore rejects malformed non-streaming SSE payloads", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "bad sse" }],
},
responseFactory() {
return new Response("data: not-json\n\ndata: [DONE]\n\n", {
status: 200,
headers: { "Content-Type": "text/event-stream" },
});
},
});
assert.equal(result.success, false);
assert.equal(result.status, 502);
assert.match(result.error, /Invalid SSE response/);
});
test("chatCore rejects malformed non-streaming JSON payloads", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "return valid json" }],
},
responseFactory() {
return new Response("{oops", {
status: 200,
headers: { "Content-Type": "application/json" },
});
},
});
assert.equal(result.success, false);
assert.equal(result.status, 502);
assert.equal(result.error, "Invalid JSON response from provider");
});
test("chatCore falls back after an empty-content success response", async () => {
const { calls, result } = await invokeChatCore({
provider: "openai",
model: "gpt-5.1",
body: {
model: "gpt-5.1",
stream: false,
messages: [{ role: "user", content: "recover from empty content" }],
},
responseFactory(_captured, seenCalls) {
if (seenCalls.length === 1) {
return new Response(
JSON.stringify({
id: "chatcmpl-empty",
object: "chat.completion",
model: "gpt-5.1",
choices: [
{
index: 0,
message: { role: "assistant", content: "" },
finish_reason: "stop",
},
],
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
}
return buildOpenAIResponse(false, "empty-content fallback ok");
},
});
const payload = (await result.response.json()) as any;
assert.equal(result.success, true);
assert.equal(calls.length, 2);
assert.equal(calls[1].body.model, "gpt-5.1-mini");
assert.equal(payload.choices[0].message.content, "empty-content fallback ok");
});
test("chatCore returns a gateway error when the empty-content fallback responds with invalid JSON", async () => {
const { result, calls } = await invokeChatCore({
provider: "openai",
model: "gpt-5.1",
body: {
model: "gpt-5.1",
stream: false,
messages: [{ role: "user", content: "recover from empty content" }],
},
responseFactory(_captured, seenCalls) {
if (seenCalls.length === 1) {
return new Response(
JSON.stringify({
id: "chatcmpl-empty",
object: "chat.completion",
model: "gpt-5.1",
choices: [
{
index: 0,
message: { role: "assistant", content: "" },
finish_reason: "stop",
},
],
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
}
return new Response("{invalid-json", {
status: 200,
headers: { "Content-Type": "application/json" },
});
},
});
assert.equal(result.success, false);
assert.equal(result.status, 502);
assert.equal(result.error, "Provider returned empty content");
assert.equal(calls.length, 2);
assert.equal(calls[1].body.model, "gpt-5.1-mini");
});
test("chatCore records Claude prompt cache and cache usage metadata in call logs", async () => {
await settingsDb.updateSettings({ alwaysPreserveClientCache: "always" });
invalidateCacheControlSettingsCache();
const { result } = await invokeChatCore({
provider: "claude",
model: "claude-sonnet-4-6",
endpoint: "/v1/messages",
credentials: { apiKey: "claude-key", providerSpecificData: {} },
requestHeaders: { "anthropic-beta": "prompt-caching-2024-07-31" },
userAgent: "Claude-Code/1.0.0",
responseFormat: "claude",
body: {
model: "claude-sonnet-4-6",
max_tokens: 64,
system: [{ type: "text", text: "system", cache_control: { type: "ephemeral", ttl: "5m" } }],
messages: [
{
role: "user",
content: [{ type: "text", text: "question", cache_control: { type: "ephemeral" } }],
},
{
role: "assistant",
content: [
{ type: "text", text: "answer", cache_control: { type: "ephemeral", ttl: "10m" } },
],
},
],
tools: [
{
name: "lookup_weather",
description: "Fetch weather",
input_schema: { type: "object" },
cache_control: { type: "ephemeral", ttl: "30m" },
},
],
},
responseFactory() {
return new Response(
JSON.stringify({
id: "msg_json",
type: "message",
role: "assistant",
model: "claude-sonnet-4-6",
content: [{ type: "text", text: "cached answer" }],
stop_reason: "end_turn",
usage: {
input_tokens: 12,
output_tokens: 3,
cache_read_input_tokens: 4,
cache_creation_input_tokens: 2,
},
}),
{
status: 200,
headers: { "Content-Type": "application/json" },
}
);
},
});
const detail = await waitFor(() => getLatestCallLog());
assert.equal(result.success, true);
assert.ok(detail);
assert.equal(detail.requestBody._omniroute.claudePromptCache.applied, true);
// Breakpoints: system[2] (1), message content (1), assistant response (1). Tools cache_control is stripped by base.ts.
assert.equal(detail.requestBody._omniroute.claudePromptCache.totalBreakpoints, 3);
assert.equal(detail.responseBody._omniroute.claudePromptCache.applied, true);
assert.equal(detail.responseBody._omniroute.claudePromptCache.totalBreakpoints, 3);
assert.equal(typeof detail.responseBody._omniroute.claudePromptCache.anthropicBeta, "string");
assert.match(detail.responseBody._omniroute.claudePromptCache.anthropicBeta, /prompt-caching/i);
assert.deepEqual(detail.responseBody._omniroute.claudePromptCacheUsage, {
cacheReadTokens: 4,
cacheCreationTokens: 2,
});
});
test("chatCore propagates budget errors without an executor-level emergency hop", async () => {
// The emergency budget fallback is orchestrated by the routing layer
// (src/sse/handlers/chat.ts), which resolves credentials FOR the emergency
// provider through account selection. The old executor-level hop here re-sent
// the FAILING provider's credentials to the emergency provider's endpoint
// (cross-provider credential leak) — the engine must now surface the budget
// error as-is, with no extra upstream call.
const { calls, result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
max_tokens: 9000,
messages: [{ role: "user", content: "keep the request alive after budget exhaustion" }],
},
responseFactory() {
return new Response(
JSON.stringify({
error: { message: "insufficient funds on this account" },
}),
{
status: 402,
headers: { "Content-Type": "application/json" },
}
);
},
});
assert.equal(result.success, false);
assert.equal(result.status, 402);
assert.equal(calls.length, 1, "no executor-level emergency hop may fire");
const body = (await result.response.json()) as any;
assert.match(String(body?.error?.message ?? ""), /insufficient funds/);
assert.ok(
!calls.some((c: any) => String(c.body?.model ?? "").includes("gpt-oss-120b")),
"emergency fallback model must not be called at executor level"
);
});
test("chatCore injects progress events into streaming responses when requested", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
requestHeaders: { "x-omniroute-progress": "true" },
body: {
model: "gpt-4o-mini",
stream: true,
messages: [{ role: "user", content: "stream with progress" }],
},
responseFactory() {
return buildOpenAIResponse(true, "streamed");
},
});
const streamText = await result.response.text();
assert.equal(result.success, true);
assert.equal(result.response.headers.get("X-OmniRoute-Progress"), "enabled");
assert.match(streamText, /event: progress/);
});
test("chatCore emits final SSE metadata comments before [DONE] on streaming responses", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: {
model: "gpt-4o-mini",
stream: true,
messages: [{ role: "user", content: "stream metadata" }],
},
responseFactory() {
return buildOpenAIResponse(true, "streamed");
},
});
const streamText = await result.response.text();
assert.equal(result.success, true);
assert.equal(result.response.headers.get("X-OmniRoute-Provider"), "openai");
assert.equal(result.response.headers.get("X-OmniRoute-Model"), "gpt-4o-mini");
assert.match(streamText, /: x-omniroute-response-cost=\d+\.\d{10}/);
assert.match(streamText, /: x-omniroute-tokens-in=\d+/);
assert.match(streamText, /: x-omniroute-tokens-out=\d+/);
assert.ok(
streamText.indexOf(": x-omniroute-response-cost=") < streamText.indexOf("data: [DONE]")
);
});
test("buildStreamingResponseHeaders drops upstream compression and framing headers", () => {
const headers = new Headers(
buildStreamingResponseHeaders(
new Headers({
"Content-Type": "text/event-stream",
"Content-Encoding": "gzip",
"Content-Length": "999",
"Transfer-Encoding": "chunked",
"X-Upstream-Trace": "trace-1",
}),
{
provider: "openai",
model: "gpt-4o-mini",
cacheHit: false,
latencyMs: 0,
usage: null,
costUsd: 0,
}
)
);
assert.equal(headers.get("Content-Type"), "text/event-stream");
assert.equal(headers.get("Content-Encoding"), null);
assert.equal(headers.get("Content-Length"), null);
assert.equal(headers.get("Transfer-Encoding"), null);
assert.equal(headers.get("X-Upstream-Trace"), "trace-1");
assert.equal(headers.get("X-OmniRoute-Cache"), "MISS");
});
test("chatCore strips upstream compression and length headers from streaming responses", async () => {
const upstreamPayload = `data: ${JSON.stringify({
id: "chatcmpl-stream-headers",
object: "chat.completion.chunk",
choices: [{ index: 0, delta: { role: "assistant", content: "streamed" } }],
})}\n\ndata: [DONE]\n\n`;
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: {
model: "gpt-4o-mini",
stream: true,
messages: [{ role: "user", content: "stream header sanitization" }],
},
responseFactory() {
return new Response(upstreamPayload, {
status: 200,
headers: {
"Content-Type": "text/event-stream",
"Content-Length": String(Buffer.byteLength(upstreamPayload)),
"X-Upstream-Trace": "trace-1",
},
});
},
});
assert.equal(result.success, true);
assert.equal(result.response.headers.get("Content-Type"), "text/event-stream");
assert.equal(result.response.headers.get("Content-Length"), null);
assert.equal(result.response.headers.get("X-Upstream-Trace"), "trace-1");
assert.equal(result.response.headers.get("X-OmniRoute-Cache"), "MISS");
await result.response.text();
});
test("chatCore maps upstream aborts to request-aborted errors", async () => {
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "abort me" }],
},
responseFactory() {
const error = new Error("request aborted by client");
error.name = "AbortError";
throw error;
},
});
assert.equal(result.success, false);
assert.equal(result.status, 499);
assert.equal(result.error, "Request aborted");
});
test("chatCore maps raw string abort reasons to 499, not 502 (#7907)", async () => {
// abort(reason) rejects the upstream fetch with the raw reason — often a
// bare string with no `name`/`status`. It must map to 499 like a named
// AbortError, not fall through to the 502 provider-failure default.
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "abort me with a string reason" }],
},
responseFactory() {
throw "request_signal_aborted";
},
});
assert.equal(result.success, false);
assert.equal(result.status, 499);
assert.equal(result.error, "Request aborted");
});
// Live incident territory (dashboard log id 1784504040241-6f8b9a): the client had
// ALREADY disconnected before this synthetic error body was ever computed — nothing
// was actually delivered to it. Persisting that body as `clientResponse` (the
// dashboard's "what the client received" field) is misleading, since it implies a
// response was sent when the client never got one. `error` above already records
// the failure reason; `clientResponse`/`responseBody` should stay empty for an abort.
test("chatCore does not log a synthetic clientResponse body for a client abort", async () => {
// clientResponse only ever lands in the persisted pipeline payloads when detailed
// call-log capture is on (attemptLogging.ts's detailedLoggingEnabled gate) — this is
// exactly the setting a real "detailed logging" connection/request has enabled, which
// is why the live incident's artifact JSON had a full pipeline (including the
// misleading clientResponse) to begin with.
await settingsDb.updateSettings({ call_log_pipeline_enabled: true });
await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "abort me, no fake clientResponse" }],
},
responseFactory() {
const error = new Error("request aborted by client");
error.name = "AbortError";
throw error;
},
});
const detail = await waitFor(() => getLatestCallLog());
assert.ok(detail);
assert.equal(detail.status, 499);
assert.match(String(detail.error ?? ""), /Request aborted/);
assert.ok(detail.pipelinePayloads, "expected pipeline payloads when capture is enabled");
assert.equal(
(detail.pipelinePayloads as Record<string, unknown>).clientResponse,
undefined,
"an aborted request never delivered anything to the client — clientResponse must stay unset"
);
});
test("chatCore returns streaming responses without waiting for upstream completion", async () => {
const encoder = new TextEncoder();
let closeUpstream: (() => void) | null = null;
const invocation = invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: {
model: "gpt-4o-mini",
stream: true,
messages: [{ role: "user", content: "do not buffer streaming" }],
},
responseFactory() {
return new Response(
new ReadableStream({
start(controller) {
controller.enqueue(
encoder.encode(
`data: ${JSON.stringify({
id: "chatcmpl-stream",
object: "chat.completion.chunk",
choices: [
{
index: 0,
delta: { role: "assistant", content: "streamed-without-buffering" },
},
],
})}\n\n`
)
);
closeUpstream = () => {
controller.enqueue(encoder.encode("data: [DONE]\n\n"));
controller.close();
};
},
}),
{
status: 200,
headers: { "Content-Type": "text/event-stream" },
}
);
},
});
const raceResult = await Promise.race([
invocation.then(() => "returned"),
// 10s ceiling: a non-buffering streaming impl resolves the invocation as soon
// as the Response is returned (upstream still open), but on a starved CI event
// loop that legitimate early return can exceed a 1s wall-clock budget (flake
// repro: 5/8 runs returned at 1.32.2s under CPU contention → false "blocked").
// The ceiling only bounds the buffered-failure case: a buffering impl never
// resolves until closeUpstream() fires below, so it still trips "blocked".
new Promise((resolve) => setTimeout(() => resolve("blocked"), 10000)),
]);
if (raceResult !== "returned") {
closeUpstream?.();
}
const { result } = await invocation;
assert.equal(raceResult, "returned");
closeUpstream?.();
const streamText = await result.response.text();
assert.equal(result.success, true);
assert.match(streamText, /streamed-without-buffering/);
});
test("chatCore releases account semaphore slots when upstream execution throws", async () => {
const connectionId = "sem-exception";
const semaphoreKey = buildAccountSemaphoreKey({
provider: "openai",
accountKey: connectionId,
});
const { result } = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
connectionId,
credentials: {
apiKey: "sk-test",
maxConcurrent: 1,
providerSpecificData: {},
},
body: {
model: "gpt-4o-mini",
stream: false,
messages: [{ role: "user", content: "executor throws" }],
},
responseFactory() {
throw new Error("simulated upstream network failure");
},
});
await waitForAsyncSideEffects();
assert.equal(result.success, false);
assert.equal(result.status, 502);
assert.equal(getAccountSemaphoreStats()[semaphoreKey], undefined);
});
test("chatCore locks per-model quota failures without dropping quota helper references", async () => {
const model = "gemini-1.5-pro";
const connection = await providersDb.createProviderConnection({
provider: "gemini",
authType: "apikey",
name: "gemini-quota-lock",
apiKey: "gemini-key",
isActive: true,
providerSpecificData: {},
});
try {
const { result } = await invokeChatCore({
provider: "gemini",
model,
connectionId: connection.id,
credentials: {
apiKey: "gemini-key",
providerSpecificData: {},
},
body: {
model,
stream: false,
messages: [{ role: "user", content: "quota lock" }],
},
responseFactory() {
return new Response(
JSON.stringify({ error: { message: "insufficient_quota: quota exhausted" } }),
{
status: 402,
headers: { "Content-Type": "application/json" },
}
);
},
});
assert.equal(result.success, false);
assert.equal(result.status, 402);
assert.equal(isModelLocked("gemini", connection.id, model), true);
} finally {
clearModelLock("gemini", connection.id, model);
}
});
// ── Streaming semantic cache tests ──────────────────────────────────────────
test("chatCore caches streaming response and serves cache HIT on repeat", async () => {
let upstreamHits = 0;
const sharedBody = {
model: "gpt-4o-mini",
stream: true,
temperature: 0,
messages: [{ role: "user", content: "stream-cache-test" }],
};
const first = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(true, "streamed-once");
},
});
assert.equal(first.result.success, true);
// Consume the stream to trigger onStreamComplete and cache write
await first.result.response.text();
await waitForAsyncSideEffects();
// Second request with same body should get cache HIT (JSON, not SSE)
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(true, "should-not-stream");
},
});
assert.equal(upstreamHits, 1, "upstream should be called only once");
assert.equal(second.calls.length, 0, "second request should not reach upstream");
assert.equal(second.result.response.headers.get("X-OmniRoute-Cache"), "HIT");
// #2952 — a streaming client receives the cache HIT as an SSE stream (not a
// raw JSON body), so content + reasoning_content arrive in the streaming shape.
assert.equal(
second.result.response.headers.get("Content-Type"),
"text/event-stream",
"streaming cache HIT should be served as SSE"
);
const sse = await second.result.response.text();
assert.match(sse, /^data:/m, "cache HIT should be SSE-framed");
assert.match(sse, /streamed-once/, "SSE cache HIT should carry the cached content");
});
test("chatCore does not cache streaming response when temperature > 0", async () => {
let upstreamHits = 0;
const sharedBody = {
model: "gpt-4o-mini",
stream: true,
temperature: 0.7,
messages: [{ role: "user", content: "non-deterministic-stream" }],
};
const first = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(true, `hot-${upstreamHits}`);
},
});
await first.result.response.text();
await waitForAsyncSideEffects();
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(true, `hot-${upstreamHits}`);
},
});
await second.result.response.text();
assert.equal(upstreamHits, 2, "both requests should hit upstream");
assert.equal(second.calls.length, 1, "second request should reach upstream");
});
test("chatCore skips streaming cache when X-OmniRoute-No-Cache header is set", async () => {
let upstreamHits = 0;
const sharedBody = {
model: "gpt-4o-mini",
stream: true,
temperature: 0,
messages: [{ role: "user", content: "no-cache-stream" }],
};
const first = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
requestHeaders: { "x-omniroute-no-cache": "true" },
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(true, "bypass-cache");
},
});
await first.result.response.text();
await waitForAsyncSideEffects();
// Verify nothing was cached
const sig = generateSignature("gpt-4o-mini", sharedBody.messages, 0, 1);
const cached = getCachedResponse(sig);
assert.equal(cached, null, "response should not be cached when no-cache header is set");
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
requestHeaders: { "x-omniroute-no-cache": "true" },
body: sharedBody,
responseFormat: "openai",
responseFactory() {
upstreamHits += 1;
return buildOpenAIResponse(true, "bypass-again");
},
});
await second.result.response.text();
assert.equal(upstreamHits, 2, "both requests should hit upstream with no-cache");
});
test("chatCore returns cache HIT as SSE when the client requests streaming", async () => {
const sharedBody = {
model: "gpt-4o-mini",
stream: false,
temperature: 0,
messages: [{ role: "user", content: "json-then-sse-cache" }],
};
// First: non-streaming request populates cache
await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
body: sharedBody,
responseFormat: "openai",
responseFactory() {
return buildOpenAIResponse(false, "cached-json");
},
});
// Second: streaming request should still get cache HIT as JSON
const second = await invokeChatCore({
provider: "openai",
model: "gpt-4o-mini",
accept: "text/event-stream",
body: { ...sharedBody, stream: true },
responseFormat: "openai",
responseFactory() {
return buildOpenAIResponse(true, "should-not-stream");
},
});
assert.equal(second.calls.length, 0, "cached response should prevent upstream call");
assert.equal(second.result.response.headers.get("X-OmniRoute-Cache"), "HIT");
// #2952 — even though the cache was populated by a non-streaming request, a
// later streaming request gets the cached completion SSE-wrapped, so streaming
// clients keep their streaming shape (and reasoning_content) on cache hits.
assert.equal(
second.result.response.headers.get("Content-Type"),
"text/event-stream",
"streaming cache HIT should be served as SSE"
);
const sse = await second.result.response.text();
assert.match(sse, /^data:/m, "cache HIT should be SSE-framed");
assert.match(sse, /cached-json/, "SSE cache HIT should carry the cached content");
});