Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
19 changes: 19 additions & 0 deletions apps/gateway/src/anthropic/anthropic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import { HTTPException } from "hono/http-exception";
import { streamSSE } from "hono/streaming";

import { app } from "@/app.js";
import { extractAnthropicSessionId } from "@/lib/session-id.js";

import { logger, toError } from "@llmgateway/logger";

Expand Down Expand Up @@ -124,6 +125,16 @@ const anthropicRequestSchema = z.object({
description: "Whether to stream the response",
example: false,
}),
metadata: z
.object({
user_id: z.string().optional(),
})
.passthrough()
.optional()
.openapi({
description:
"Anthropic request metadata. Claude Code embeds the session id in user_id, which the gateway uses for sticky routing.",
}),
});

const anthropicContentBlockSchema = z.object({
Expand Down Expand Up @@ -501,6 +512,13 @@ anthropic.openapi(messages, async (c) => {
// Get user-agent for forwarding
const userAgent = c.req.header("User-Agent") ?? "";

// Sticky-routing session id: prefer an explicit header, otherwise derive it
// from Anthropic's metadata.user_id (Claude Code embeds the session id here)
// and forward it to the chat completions endpoint, which routes on it.
const sessionId =
c.req.header("x-session-id")?.trim() ||
extractAnthropicSessionId(anthropicRequest.metadata?.user_id);

// Make internal request to the existing chat completions endpoint using app.request()
const response = await app.request("/v1/chat/completions", {
method: "POST",
Expand All @@ -513,6 +531,7 @@ anthropic.openapi(messages, async (c) => {
"x-source": c.req.header("x-source") ?? "",
"x-debug": c.req.header("x-debug") ?? "",
"HTTP-Referer": c.req.header("HTTP-Referer") ?? "",
...(sessionId ? { "x-session-id": sessionId } : {}),
},
body: JSON.stringify(openaiRequest),
});
Expand Down
20 changes: 20 additions & 0 deletions apps/gateway/src/chat/chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -305,6 +305,7 @@ function collapseProvidersToBestRegionPerProvider(
metricsMap: Map<string, ProviderMetrics>;
isStreaming: boolean;
promptTokens?: number;
sessionId?: string;
},
): ProviderModelMapping[] {
const providersById = new Map<string, ProviderModelMapping[]>();
Expand Down Expand Up @@ -1095,7 +1096,17 @@ chat.openapi(completions, async (c) => {
effort,
web_search,
plugins,
user,
} = validationResult.data;

// Sticky-routing session key: explicit header wins, then OpenAI-native body
// fields (prompt_cache_key, then user). When present, provider selection
// pins this session to a single provider to keep upstream prompt caches warm.
const sessionId =
c.req.header("x-session-id")?.trim() ||
prompt_cache_key ||
user ||
undefined;
Comment on lines +1167 to +1172

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Normalize body-derived session IDs before sticky routing.

prompt_cache_key and user are not trimmed, so whitespace variants can pin to different providers unexpectedly. Normalize them like the header path.

💡 Proposed fix
 const sessionId =
 	c.req.header("x-session-id")?.trim() ||
-	prompt_cache_key ||
-	user ||
+	prompt_cache_key?.trim() ||
+	user?.trim() ||
 	undefined;
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
const sessionId =
c.req.header("x-session-id")?.trim() ||
prompt_cache_key ||
user ||
undefined;
const sessionId =
c.req.header("x-session-id")?.trim() ||
prompt_cache_key?.trim() ||
user?.trim() ||
undefined;
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/gateway/src/chat/chat.ts` around lines 1105 - 1109, The sessionId
selection currently trims only the header value but not the body-derived values,
so prompt_cache_key and user can contain leading/trailing whitespace and break
sticky routing; update the sessionId assignment to normalize (trim)
prompt_cache_key and user the same way as c.req.header does, ensuring the
ternary/OR chain that builds sessionId (using variables sessionId,
prompt_cache_key, user, and c.req.header) calls .trim() or an equivalent
normalization on each non-header branch before falling back to undefined.

let {
messages,
temperature,
Expand Down Expand Up @@ -2044,6 +2055,7 @@ chat.openapi(completions, async (c) => {
metricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand All @@ -2054,6 +2066,7 @@ chat.openapi(completions, async (c) => {
metricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down Expand Up @@ -2272,6 +2285,7 @@ chat.openapi(completions, async (c) => {
metricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down Expand Up @@ -2462,6 +2476,7 @@ chat.openapi(completions, async (c) => {
metricsMap: allMetricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down Expand Up @@ -2617,6 +2632,7 @@ chat.openapi(completions, async (c) => {
metricsMap: allMetricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down Expand Up @@ -2646,6 +2662,7 @@ chat.openapi(completions, async (c) => {
metricsMap: allMetricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down Expand Up @@ -2834,6 +2851,7 @@ chat.openapi(completions, async (c) => {
metricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand All @@ -2844,6 +2862,7 @@ chat.openapi(completions, async (c) => {
metricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down Expand Up @@ -3059,6 +3078,7 @@ chat.openapi(completions, async (c) => {
metricsMap,
isStreaming: stream,
promptTokens: routingPromptTokens,
sessionId,
},
);

Expand Down
10 changes: 10 additions & 0 deletions apps/gateway/src/chat/schemas/completions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,16 @@ export const completionsRequestSchema = z.object({
"OpenAI prompt cache retention policy. OpenAI supports in_memory and 24h for eligible models.",
example: "24h",
}),
user: z
.string()
.nullable()
.optional()
.transform((val) => (val === null ? undefined : val))
.openapi({
description:
"OpenAI end-user identifier. Used as a fallback sticky-routing session key when no x-session-id header or prompt_cache_key is provided.",
example: "user-123",
}),
tools: z
.array(
z.union([
Expand Down
27 changes: 27 additions & 0 deletions apps/gateway/src/lib/session-id.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
import { describe, expect, it } from "vitest";

import { extractAnthropicSessionId } from "./session-id.js";

describe("extractAnthropicSessionId", () => {
it("extracts the session segment from a Claude Code user_id", () => {
expect(
extractAnthropicSessionId(
"user_abc123_account_def456_session_9f8e7d6c-1234-4abc-9def-0123456789ab",
),
).toBe("session_9f8e7d6c-1234-4abc-9def-0123456789ab");
});

it("falls back to the whole value when no session segment is present", () => {
expect(extractAnthropicSessionId("tenant-42")).toBe("tenant-42");
});

it("returns undefined for empty or missing values", () => {
expect(extractAnthropicSessionId(undefined)).toBeUndefined();
expect(extractAnthropicSessionId("")).toBeUndefined();
expect(extractAnthropicSessionId(" ")).toBeUndefined();
});

it("trims surrounding whitespace", () => {
expect(extractAnthropicSessionId(" tenant-42 ")).toBe("tenant-42");
});
});
31 changes: 31 additions & 0 deletions apps/gateway/src/lib/session-id.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/**
* Session id resolution for sticky routing.
*
* The gateway pins a logical session to a single upstream provider so that
* provider-side prompt caches stay warm across a multi-turn conversation. The
* sticky key is sourced, in priority order, from the explicit `x-session-id`
* header, then provider-native fields the coding agents already send.
*/

/**
* Extract a stable session id from Anthropic's `metadata.user_id`.
*
* Claude Code populates `metadata.user_id` with a structured string that embeds
* the session UUID, e.g. `user_<hash>_account_<hash>_session_<uuid>`. We pull
* out the `session_<uuid>` segment so the key is stable per conversation rather
* than per user. When no session segment is present we fall back to the whole
* value, which is still stable for the caller.
*/
export function extractAnthropicSessionId(
userId: string | undefined,
): string | undefined {
if (!userId) {
return undefined;
}
const trimmed = userId.trim();
if (!trimmed) {
return undefined;
}
const match = trimmed.match(/session_[A-Za-z0-9-]+/);
return match ? match[0] : trimmed;
}
96 changes: 96 additions & 0 deletions packages/actions/src/get-cheapest-from-available-providers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,53 @@ export interface ProviderSelectionOptions {
* weighted score.
*/
promptTokens?: number;
/**
* Sticky-routing session identifier. When provided, the provider (and
* region) is chosen deterministically via rendezvous hashing so that all
* requests for the same session pin to the same provider, maximizing
* upstream prompt-cache hits. Price/uptime scoring is bypassed; the session
* only moves to another provider when its pinned provider is no longer in
* the available list (e.g. excluded by health filtering or removed by the
* retry-fallback loop after the provider failed).
*/
sessionId?: string;
}

/**
* FNV-1a 32-bit hash mapped to the unit interval [0, 1). Deterministic across
* processes, no crypto needed.
*/
function hashToUnitInterval(input: string): number {
let hash = 0x811c9dc5;
for (let i = 0; i < input.length; i++) {
hash ^= input.charCodeAt(i);
hash = Math.imul(hash, 0x01000193);
}
return (hash >>> 0) / 0xffffffff;
}

/**
* Rendezvous (highest-random-weight) hashing: deterministically pick one
* provider for a session. Unlike modulo hashing, removing a provider only
* reassigns the sessions that were pinned to it — every other session keeps
* its provider.
*/
function selectStickyProvider<T extends AvailableModelProvider>(
providers: T[],
sessionId: string,
): T {
let best = providers[0];
let bestWeight = -1;
for (const provider of providers) {
const weight = hashToUnitInterval(
`${sessionId}|${provider.providerId}|${provider.region ?? ""}`,
);
if (weight > bestWeight) {
bestWeight = weight;
best = provider;
}
}
return best;
}

function findProviderMapping<P extends ModelWithPricing["providers"][number]>(
Expand Down Expand Up @@ -352,6 +399,55 @@ export function getCheapestFromAvailableProviders<
return null;
}

// Sticky routing: when a session id is provided, pin the session to a single
// provider via rendezvous hashing. Bypasses scoring and exploration so the
// upstream prompt cache stays warm; the session only moves if its provider
// leaves the available list (health filtering or retry-fallback exclusion).
const sessionId = options?.sessionId?.trim();
if (sessionId) {
const stickyProvider = selectStickyProvider(stableProviders, sessionId);
return {
provider: stickyProvider,
metadata: {
availableProviders: stableProviders.map((p) => p.providerId),
selectedProvider: stickyProvider.providerId,
selectionReason: "session-sticky",
providerScores: stableProviders.map((provider) => {
const providerInfo = findProviderMapping(
modelWithPricing.providers,
provider,
);
const providerDef = getProviderDefinition(provider.providerId);
const priority = providerDef?.priority ?? 1;
const metrics = metricsMap?.get(
metricsKey(
modelWithPricing.id,
provider.providerId,
provider.region,
),
);

return {
providerId: provider.providerId,
region: provider.region,
score: 0,
uptime: metrics?.uptime,
latency: metrics?.averageLatency,
throughput: metrics?.throughput,
price: getProviderSelectionPrice(
providerInfo,
videoPricing,
).toNumber(),
priority,
cacheSupported: providerSupportsCaching(
providerInfo as ProviderModelMapping | undefined,
),
};
}),
},
};
}

// Epsilon-greedy exploration: randomly select a provider 1% of the time
// This ensures all providers get periodic traffic and build up metrics
// Skip during tests to keep behavior deterministic
Expand Down
Loading
Loading