Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 24 additions & 5 deletions open-sse/services/autoCombo/__tests__/speedRanking.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -165,15 +165,34 @@ describe("rankBySpeed — factor breakdown", () => {
});

it("falls back to 0.5 per missing metric so new providers are not crushed", () => {
const ranked = rankBySpeed([candidate({ provider: "fresh", model: "m" })]);
const ranked = rankBySpeed([
candidate({
provider: "fresh",
model: "m",
p95LatencyMs: undefined,
latencyStdDev: undefined,
}),
]);
expect(ranked).toHaveLength(1);
// No telemetry at all → weighted sum lands near 0.5 with reliability multiplier 1
expect(ranked[0].factors.reliability).toBe(1);
expect(ranked[0].factors.health).toBe(1);
expect(ranked[0].factors.ttft).toBe(0.5);
expect(ranked[0].factors.tps).toBe(0.5);
});
});
expect(ranked[0].factors.tps).toBe(0.5);
});

it("uses p95 latency when TTFT and E2E telemetry are unavailable", () => {
const ranked = rankBySpeed([
candidate({ provider: "slow-tail", model: "m", p95LatencyMs: 4000 }),
candidate({ provider: "fast-tail", model: "m", p95LatencyMs: 1000 }),
]);
const fast = ranked.find((entry) => entry.provider === "fast-tail");
const slow = ranked.find((entry) => entry.provider === "slow-tail");

expect(fast?.factors.ttft).toBeGreaterThan(slow?.factors.ttft ?? 1);
expect(fast?.factors.e2e).toBeGreaterThan(slow?.factors.e2e ?? 1);
});
});

describe("rankBySpeed — weight overrides", () => {
it("respects caller weight overrides (e.g. heavy TTFT bias)", () => {
Expand Down Expand Up @@ -223,4 +242,4 @@ describe("pickFastest", () => {
const winner = pickFastest([slow, fast]);
expect(winner?.provider).toBe("fast");
});
});
});
10 changes: 8 additions & 2 deletions open-sse/services/autoCombo/speedRanking.ts
Original file line number Diff line number Diff line change
Expand Up @@ -211,9 +211,15 @@ function speedFactorsFor(
failureRate: number
): SpeedFactors {
return {
ttft: lowerIsBetter(positiveFinite(candidate.avgTtftMs), maxima.ttft),
ttft: lowerIsBetter(
positiveFinite(candidate.avgTtftMs) ?? positiveFinite(candidate.p95LatencyMs),
maxima.ttft
),
tps: higherIsBetter(positiveFinite(candidate.avgTokensPerSecond), maxima.tps),
e2e: lowerIsBetter(positiveFinite(candidate.avgE2ELatencyMs), maxima.e2e),
e2e: lowerIsBetter(
positiveFinite(candidate.avgE2ELatencyMs) ?? positiveFinite(candidate.p95LatencyMs),
maxima.e2e
),
p95: lowerIsBetter(positiveFinite(candidate.p95LatencyMs), maxima.p95),
health: healthScoreFor(candidate.circuitBreakerState),
reliability: clamp01(1 - failureRate),
Expand Down
6 changes: 6 additions & 0 deletions open-sse/utils/earlyStreamKeepalive.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,12 @@

const ENCODER = new TextEncoder();
const KEEPALIVE_FRAME = ENCODER.encode(": omniroute-keepalive\n\n");
// OpenAI-compatible keepalive: a syntactically valid empty streaming chunk.
// Some OpenAI-compatible clients parse every non-empty SSE line as JSON and
// reject legal SSE comments before their first provider chunk arrives.
export const OPENAI_KEEPALIVE_FRAME = ENCODER.encode(
'data: {"id":"omniroute-keepalive","object":"chat.completion.chunk","created":0,"model":"omniroute","choices":[{"index":0,"delta":{},"finish_reason":null}]}\n\n'
);
// Anthropic Messages-format keepalive: a REAL `ping` SSE event, not a comment.
// Anthropic clients (Claude Code, the Anthropic SDK) reset their stream/first-token
// watchdog on real SSE events but ignore SSE comments (`: ...`), so on a slow first
Expand Down
9 changes: 8 additions & 1 deletion scripts/quality/validate-release-green.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,14 @@ export function parseEslintJson(out) {

/** Pull the cognitive-complexity violation count from the gate's output. */
export function parseCognitiveCount(out) {
const m = String(out || "").match(/(\d+)\s+(?:function\(s\) exceed|violações|violations)/i);
const s = String(out || "");
// `check:complexity-ratchets` runs ONE shared ESLint walk and prints BOTH ratchets, with the
// cyclomatic "N violações" summary emitted FIRST — so a bare `\d+ violações` regex would grab
// the cyclomatic count. Prefer the unambiguous machine-readable `cognitiveComplexity=N` line
// (mirrors the cyclomatic `complexity=N` parse used for cycCurrent below).
const machine = s.match(/(?:^|\n)cognitiveComplexity=(\d+)/);
if (machine) return Number(machine[1]);
const m = s.match(/(\d+)\s+(?:function\(s\) exceed|violações|violations)/i);
return m ? Number(m[1]) : null;
}

Expand Down
6 changes: 5 additions & 1 deletion src/app/api/v1/chat/completions/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,10 @@ import { generateRequestId } from "@/shared/utils/requestId";
import { initTranslators } from "@omniroute/open-sse/translator/index.ts";
import { createInjectionGuard } from "@/middleware/promptInjectionGuard";
import { acceptHeaderForcesStream } from "@omniroute/open-sse/utils/aiSdkCompat.ts";
import { withEarlyStreamKeepalive } from "@omniroute/open-sse/utils/earlyStreamKeepalive";
import {
OPENAI_KEEPALIVE_FRAME,
withEarlyStreamKeepalive,
} from "@omniroute/open-sse/utils/earlyStreamKeepalive";
import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
import { checkChatAdmission } from "@/shared/middleware/chatBodyAdmission";
import {
Expand Down Expand Up @@ -132,6 +135,7 @@ export async function POST(request) {
{
signal: request.signal,
thresholdMs: resolveKeepaliveThreshold(parsedBody?.model),
keepaliveFrame: OPENAI_KEEPALIVE_FRAME,
extraHeaders: { "X-Correlation-Id": reqId },
}
);
Expand Down
5 changes: 4 additions & 1 deletion tests/unit/chat-combo-live-test.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -274,7 +274,10 @@ test("chat completions route emits early keepalive while waiting for stream read
assert.match(response.headers.get("content-type") || "", /text\/event-stream/);

const body = await readAll(response);
assert.match(body, /: omniroute-keepalive/);
assert.match(
body,
/data: \{"id":"omniroute-keepalive","object":"chat\.completion\.chunk"/
);
assert.match(body, /OK/);
assert.match(body, /\[DONE\]/);
});
Expand Down
34 changes: 33 additions & 1 deletion tests/unit/early-stream-keepalive.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import assert from "node:assert/strict";
import {
withEarlyStreamKeepalive,
ANTHROPIC_PING_FRAME,
OPENAI_KEEPALIVE_FRAME,
} from "../../open-sse/utils/earlyStreamKeepalive.ts";

async function readAll(response: Response): Promise<string> {
Expand Down Expand Up @@ -68,9 +69,40 @@ test("ANTHROPIC_PING_FRAME is a real Anthropic ping event (not a comment)", () =
assert.doesNotMatch(decoded, /^:/, "must not be an SSE comment");
});

test("OPENAI_KEEPALIVE_FRAME is a JSON-parseable OpenAI streaming chunk", () => {
const decoded = new TextDecoder().decode(OPENAI_KEEPALIVE_FRAME);
assert.match(decoded, /^data: /);
assert.doesNotMatch(decoded, /^:/, "must not be an SSE comment");

const payload = JSON.parse(decoded.slice("data: ".length).trim());
assert.equal(payload.object, "chat.completion.chunk");
assert.deepEqual(payload.choices, [{ index: 0, delta: {}, finish_reason: null }]);
});

test("slow handler emits the custom OpenAI keepalive chunk before the body", async () => {
const slow = new Promise<Response>((resolve) => {
setTimeout(() => resolve(sseResponse("data: [DONE]\n\n")), 120);
});

const result = await withEarlyStreamKeepalive(slow, {
thresholdMs: 25,
intervalMs: 20,
keepaliveFrame: OPENAI_KEEPALIVE_FRAME,
});

const body = await readAll(result);
assert.doesNotMatch(body, /: omniroute-keepalive/);
const firstFrame = body.split("\n\n")[0];
assert.doesNotThrow(() => JSON.parse(firstFrame.slice("data: ".length)));
assert.match(body, /data: \[DONE\]/);
});

test("slow handler emits the custom keepaliveFrame (Anthropic ping) before the body", async () => {
const slow = new Promise<Response>((resolve) => {
setTimeout(() => resolve(sseResponse("event: message_start\ndata: {}\n\ndata: [DONE]\n\n")), 120);
setTimeout(
() => resolve(sseResponse("event: message_start\ndata: {}\n\ndata: [DONE]\n\n")),
120
);
});

const result = await withEarlyStreamKeepalive(slow, {
Expand Down
14 changes: 14 additions & 0 deletions tests/unit/validate-release-green.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,20 @@ test("parseCognitiveCount reads the gate's count (en + pt)", () => {
assert.equal(parseCognitiveCount("no number"), null);
});

test("parseCognitiveCount ignores the cyclomatic count in the combined ratchets output (#7009)", () => {
// `check:complexity-ratchets` runs ONE shared ESLint walk and prints BOTH ratchets.
// The cyclomatic "N violações" summary is emitted FIRST, so a bare `\\d+ violações`
// regex captured 2056 (cyclomatic) instead of 890 (cognitive) — a phantom drift in
// every pre-flight report. Prefer the unambiguous machine-readable `cognitiveComplexity=N`.
const combined = [
"complexity=2056",
"cognitiveComplexity=890",
"[complexity] OK — 2056 violações (baseline 2056)",
"[cognitive-complexity] OK — 890 violações (baseline 890)",
].join("\n");
assert.equal(parseCognitiveCount(combined), 890);
});

test("isDrift flags only growth past the committed baseline (down-direction ratchets)", () => {
assert.equal(isDrift(3900, 3867), true); // grew → drift
assert.equal(isDrift(3867, 3867), false); // equal → ok
Expand Down
Loading