Skip to content

Refactor: Extract chatCore phases into modular files #3598

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
diegosouzapw merged 3 commits into from
Jun 11, 2026
Merged

Refactor: Extract chatCore phases into modular files #3598

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8332f19
refactor: extract chatCore phases into modular files
oyi77 Jun 10, 2026
6d4ff14
Merge remote-tracking branch 'origin/release/v3.8.21' into work-3598
diegosouzapw Jun 11, 2026
9cbae02
fix(chatCore): re-derive idempotencyKey for the save path after modul…
oyi77 Jun 11, 2026
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
350 changes: 48 additions & 302 deletions open-sse/handlers/chatCore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
import { injectMemoryAndSkills } from "./chatCore/memorySkillsInjection.ts";
import { checkIdempotencyCache } from "./chatCore/idempotency.ts";
import { checkSemanticCache } from "./chatCore/semanticCache.ts";
import { sanitizeChatRequestBody } from "./chatCore/sanitization.ts";
Comment on lines +1 to +4

@gemini-code-assist gemini-code-assist Bot Jun 10, 2026

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.

medium

The Repository Style Guide (Rule 9) requires that tests be included when changing production code. Please add unit tests for the new modular functions in open-sse/handlers/chatCore/.

References
  1. Rule 9: Always include tests when changing production code (src/, open-sse/, electron/, bin/).

import { CORS_HEADERS } from "../utils/cors.ts";
import { HEAP_PRESSURE_THRESHOLD_MB } from "../utils/heapPressure.ts";
import { normalizeHeaders } from "../utils/headers.ts";
Expand Down Expand Up @@ -176,7 +180,7 @@ import {
isCacheableForRead,
isCacheableForWrite,
} from "@/lib/semanticCache";
import { getIdempotencyKey, checkIdempotency, saveIdempotency } from "@/lib/idempotencyLayer";
import { getIdempotencyKey, saveIdempotency } from "@/lib/idempotencyLayer";
import { createProgressTransform, wantsProgress } from "../utils/progressTracker.ts";
import { createPiiSseTransform } from "@/lib/streamingPiiTransform";
import { isFeatureFlagEnabled } from "@/shared/utils/featureFlags";
Expand Down Expand Up @@ -209,14 +213,6 @@ import {
import { generateRequestId } from "@/shared/utils/requestId";
import { normalizePayloadForLog } from "@/lib/logPayloads";
import { extractFacts } from "@/lib/memory/extraction";
import { injectMemory, shouldInjectMemory } from "@/lib/memory/injection";
import { retrieveMemories } from "@/lib/memory/retrieval";
import {
DEFAULT_MEMORY_SETTINGS,
getMemorySettings,
toMemoryRetrievalConfig,
} from "@/lib/memory/settings";
import { injectSkills } from "@/lib/skills/injection";
import { handleToolCallExecution } from "@/lib/skills/interception";
import { OMNIROUTE_RESPONSE_HEADERS } from "@/shared/constants/headers";
import {
Expand Down Expand Up @@ -1871,39 +1867,16 @@ export async function handleChatCore({
};

// ── Phase 9.2: Idempotency check ──
const idempotencyKey = getIdempotencyKey(clientRawRequest?.headers);
const cachedIdemp = checkIdempotency(idempotencyKey);
if (cachedIdemp) {
log?.debug?.("IDEMPOTENCY", `Hit for key=${idempotencyKey?.slice(0, 12)}...`);
const idempotentUsage =
cachedIdemp.response && typeof cachedIdemp.response === "object"
? ((cachedIdemp.response as Record<string, unknown>).usage as
| Record<string, unknown>
| undefined)
: undefined;
const idempotentCost = idempotentUsage
? await calculateCost(provider, model, idempotentUsage as Record<string, number>, {
serviceTier: effectiveServiceTier,
})
: 0;
return {
success: true,
response: new Response(JSON.stringify(cachedIdemp.response), {
status: cachedIdemp.status,
headers: {
"Content-Type": "application/json",
"X-OmniRoute-Idempotent": "true",
...buildOmniRouteResponseMetaHeaders({
provider,
model,
cacheHit: false,
latencyMs: Date.now() - startTime,
usage: idempotentUsage,
costUsd: idempotentCost,
}),
},
}),
};
const idempotencyHit = await checkIdempotencyCache({
clientRawRequest,
provider,
model,
effectiveServiceTier,
startTime,
log,
});
if (idempotencyHit) {
return idempotencyHit;
}

// T07: Inject connectionId into credentials so executors can rotate API keys
Expand Down Expand Up @@ -2303,269 +2276,38 @@ export async function handleChatCore({
log?.debug?.("FORMAT", `${sourceFormat} → ${targetFormat} | stream=${stream}`);

// ── Phase 9.1: Semantic cache check (temp=0, any streaming mode) ──
// Streaming responses are cached after assembly; cache hits return JSON regardless of stream flag.
if (semanticCacheEnabled && isCacheableForRead(body, clientRawRequest?.headers)) {
const signature = generateSignature(
model,
body.messages ?? body.input,
body.temperature,
body.top_p
);
const cached = getCachedResponse(signature);
if (cached) {
log?.debug?.("CACHE", `Semantic cache HIT for ${model} (stream=${stream})`);
reqLogger.logConvertedResponse(cached as Record<string, unknown>);
const cachedUsage =
extractUsageFromResponse(cached as Record<string, unknown>, provider) ||
((cached as Record<string, unknown>)?.usage as Record<string, unknown> | undefined);
const cachedCost = cachedUsage
? await calculateCost(provider, model, cachedUsage as Record<string, number>, {
serviceTier: effectiveServiceTier,
})
: 0;
persistAttemptLogs({
status: 200,
tokens: (cached as Record<string, unknown>)?.usage,
responseBody: cached,
providerRequest: null,
providerResponse: null,
clientResponse: cached,
cacheSource: "semantic",
});
trackPendingRequest(model, provider, connectionId, false);
// #2952 — when the client requested a stream, serve the cached completion
// as an SSE stream (not a raw JSON body) so content + reasoning_content
// arrive in the streaming shape the client expects. Cache hits previously
// returned application/json regardless of the stream flag, which made
// OpenAI-compatible streaming clients lose reasoning_content. Non-OpenAI
// shapes (no `choices`) yield "" and fall back to the JSON body unchanged.
const cachedSse = stream ? synthesizeOpenAiSseFromJson(JSON.stringify(cached)) : "";
const cacheHitMetaHeaders = buildOmniRouteResponseMetaHeaders({
provider,
model,
cacheHit: true,
latencyMs: Date.now() - startTime,
usage: cachedUsage,
costUsd: cachedCost,
});
return {
success: true,
response: new Response(cachedSse || JSON.stringify(cached), {
headers: {
"Content-Type": cachedSse ? "text/event-stream" : "application/json",
[OMNIROUTE_RESPONSE_HEADERS.cache]: "HIT",
...cacheHitMetaHeaders,
},
}),
};
}
}

// ── Common input sanitization (runs for ALL paths including passthrough) ──
// #994: Normalize between max_output_tokens and max_tokens for universal compatibility.
// For Responses API targets, max_output_tokens is the canonical field. For others,
// max_tokens is preferred. We handle normalization here to support passthrough
// paths where the translator is skipped.
const prefersResponsesTokenField =
sourceFormat === FORMATS.OPENAI_RESPONSES || targetFormat === FORMATS.OPENAI_RESPONSES;

if (prefersResponsesTokenField) {
if (body.max_output_tokens === undefined) {
if (body.max_completion_tokens !== undefined) {
body.max_output_tokens = body.max_completion_tokens;
delete body.max_completion_tokens;
} else if (body.max_tokens !== undefined) {
body.max_output_tokens = body.max_tokens;
delete body.max_tokens;
}
}
} else if (body.max_output_tokens !== undefined) {
if (body.max_tokens === undefined) {
body.max_tokens = body.max_output_tokens;
}
delete body.max_output_tokens;
}

// #291: Strip empty name fields from messages/input items
// Upstream providers (OpenAI, Codex) reject name:"" with 400 errors.
if (Array.isArray(body.messages)) {
body.messages = body.messages.map((msg: Record<string, unknown>) => {
if (msg.name === "") {
const { name: _n, ...rest } = msg;
return rest;
}
return msg;
});
}
if (Array.isArray(body.input)) {
body.input = body.input.map((item: Record<string, unknown>) => {
if (item.name === "") {
const { name: _n, ...rest } = item;
return rest;
}
return item;
});
}
// #346/#637: Strip tools with empty name
// Clients sometimes forward tool definitions with empty names, causing
// upstream providers to reject with 400 "Invalid 'tools[0].name': empty string."
if (Array.isArray(body.tools)) {
body.tools = body.tools.filter((tool: Record<string, unknown>) => {
// Built-in Responses API tool types (web_search, file_search, computer, etc.)
// are identified solely by their `type` field and carry no name — preserve them.
const toolType = typeof tool.type === "string" ? tool.type : "";
if (toolType && toolType !== "function" && !tool.function && tool.name === undefined) {
return true;
}
const fn = tool.function as Record<string, unknown> | undefined;
const name = fn?.name ?? tool.name;
return name && String(name).trim().length > 0;
});

// Sanitize OpenAI-format function tool schemas before they reach strict
// upstream JSON Schema validators (e.g. Moonshot AI behind
// opencode-go/kimi-k2.6). See toolSchemaSanitizer.ts for the specific bug.
// sanitizeOpenAITool is safe to call on any input — it no-ops non-function
// tools (e.g. Responses API built-ins) and non-object values.
body.tools = body.tools.map((tool) => sanitizeOpenAITool(tool) as (typeof body.tools)[number]);
const cacheHit = await checkSemanticCache({
semanticCacheEnabled,
body,
clientRawRequest,
model,
provider,
stream: !!stream,
reqLogger,
effectiveServiceTier,
connectionId,
startTime,
log,
persistAttemptLogs,
});
if (cacheHit) {
return cacheHit;
}

body = sanitizeChatRequestBody(body, sourceFormat, targetFormat);
const memoryOwnerId = resolveMemoryOwnerId(apiKeyInfo as Record<string, unknown> | null);
const memorySettings = memoryOwnerId
? await getMemorySettings().catch(() => DEFAULT_MEMORY_SETTINGS)
: null;

if (
memoryOwnerId &&
memorySettings &&
shouldInjectMemory(body as Parameters<typeof shouldInjectMemory>[0], {
enabled: memorySettings.enabled && memorySettings.maxTokens > 0,
})
) {
try {
// Plan 21 FAIL #1 fix: extract the last user message and pass it as
// `query`. Without this, `config.query` is undefined in retrieveMemories
// and the semantic/hybrid branches (sqlite-vec + RRF, and Qdrant
// tier-2) never fire from the chat hot path — they only fire in the
// Playground (retrievePreview, which gets `query` as a positional arg).
const lastUserQuery = ((): string => {
// Responses API item types that are NOT user input — never accept
// their text as the retrieval query (e.g. function_call_output is the
// tool's reply, reasoning is the model's chain of thought).
const NON_USER_TYPES = new Set([
"function_call",
"function_call_output",
"tool_call",
"tool_call_output",
"reasoning",
"computer_call",
"computer_call_output",
"web_search_call",
"file_search_call",
]);

function pickFrom(arr: unknown[]): string {
for (let i = arr.length - 1; i >= 0; i--) {
const item = arr[i] as Record<string, unknown> | undefined;
if (!item) continue;
// Chat API: only role==="user" items. Responses API items often
// have type instead of role — skip non-user types like
// function_call_output so the tool's reply doesn't leak into the
// memory query.
if (item.role !== undefined && item.role !== "user") continue;
if (item.role === undefined && typeof item.type === "string") {
if (NON_USER_TYPES.has(item.type)) continue;
}
const content = item.content ?? item.text;
if (typeof content === "string" && content.trim().length > 0) {
return content;
}
if (Array.isArray(content)) {
const parts: string[] = [];
for (const p of content) {
if (typeof p === "string") {
parts.push(p);
} else if (p && typeof p === "object") {
const pp = p as Record<string, unknown>;
// Skip non-text content parts (image_url, tool_use, etc.)
const ptype = typeof pp.type === "string" ? pp.type : "";
if (
ptype &&
ptype !== "text" &&
ptype !== "input_text" &&
ptype !== "output_text"
) {
continue;
}
const t = pp.text ?? pp.input_text;
if (typeof t === "string") parts.push(t);
}
}
if (parts.length > 0) return parts.join(" ").trim();
}
}
return "";
}
const b = body as Record<string, unknown>;
if (Array.isArray(b.messages)) {
const r = pickFrom(b.messages);
if (r) return r;
}
if (Array.isArray(b.input)) {
const r = pickFrom(b.input);
if (r) return r;
}
return "";
})();

const memories = await retrieveMemories(
memoryOwnerId,
toMemoryRetrievalConfig(memorySettings, { query: lastUserQuery })
);
if (memories.length > 0) {
const injected = injectMemory(
body as Parameters<typeof injectMemory>[0],
memories,
provider
);
body = injected as typeof body;
log?.debug?.("MEMORY", `Injected ${memories.length} memories for key=${memoryOwnerId}`);
}
} catch (memErr) {
log?.debug?.(
"MEMORY",
`Memory injection skipped: ${memErr instanceof Error ? memErr.message : String(memErr)}`
);
}
}

if (memoryOwnerId && memorySettings?.skillsEnabled) {
const existingTools = Array.isArray(body.tools) ? body.tools : [];
const mergedTools = injectSkills({
provider: getSkillsProviderForFormat(sourceFormat),
existingTools,
apiKeyId: memoryOwnerId,
model: typeof effectiveModel === "string" ? effectiveModel : undefined,
sourceFormat,
targetFormat,
backgroundReason,
messages: Array.isArray(body.messages)
? body.messages
: Array.isArray(body.input)
? body.input
: undefined,
});

if (mergedTools.length > existingTools.length) {
body = {
...body,
tools: mergedTools,
};
log?.debug?.("SKILLS", `Injected ${mergedTools.length - existingTools.length} skills`);
}
}

trace("post_injection", { provider, model });
const injectionResult = await injectMemoryAndSkills({
body,
memoryOwnerId,
provider,
effectiveModel,
sourceFormat,
targetFormat,
backgroundReason,
log,
});
body = injectionResult.body;
const memorySettings = injectionResult.memorySettings;

// Translate request (pass reqLogger for intermediate logging)
// ── Proactive Context Compression (Phase 4) ──
Expand Down Expand Up @@ -5490,6 +5232,10 @@ export async function handleChatCore({
}

// ── Phase 9.2: Save for idempotency ──
// The idempotency *check* moved into checkIdempotencyCache() during the
// chatCore modularization (#3598); re-derive the key here for the save path.
// getIdempotencyKey is pure (reads idempotency-key/x-request-id headers).
const idempotencyKey = getIdempotencyKey(clientRawRequest?.headers);
saveIdempotency(idempotencyKey, translatedResponse, 200);
reqLogger.logConvertedResponse(translatedResponse);
persistAttemptLogs({
Expand Down
Loading