diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index d4caf28454..ef7b3616f4 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -11,6 +11,7 @@ import { storeSemanticCacheResponse } from "./chatCore/semanticCacheStore.ts"; import { buildNonStreamingResponseHeaders } from "./chatCore/nonStreamingResponseHeaders.ts"; import { maybeConvertJsonBodyToSse } from "./chatCore/jsonBodyToSse.ts"; import { assembleStreamingResponseHeaders } from "./chatCore/streamingResponseHeaders.ts"; +import { storeStreamingSemanticCacheResponse } from "./chatCore/streamingSemanticCacheStore.ts"; import { sanitizeChatRequestBody } from "./chatCore/sanitization.ts"; import { getHeaderValueCaseInsensitive, @@ -3861,32 +3862,17 @@ export async function handleChatCore({ } // Semantic cache: store assembled streaming response for future cache hits - if ( - semanticCacheEnabled && - streamStatus === 200 && - streamResponseBody && - isCacheableForWrite(body, clientRawRequest?.headers) - ) { - try { - const cleanBody = { ...streamResponseBody }; - delete cleanBody._streamed; - if (!isSmallEnoughForSemanticCache(cleanBody)) return; - const sig = generateSignature( - model, - body.messages ?? body.input, - body.temperature, - body.top_p, - apiKeyInfo?.id ?? undefined - ); - const u = streamUsage as Record | null; - const tokensSaved = - (Number(u?.prompt_tokens ?? 0) || 0) + (Number(u?.completion_tokens ?? 0) || 0); - setCachedResponse(sig, model, cleanBody, tokensSaved); - log?.debug?.("CACHE", `Stored streaming response for ${model} (${tokensSaved} tokens)`); - } catch { - // Cache write failed — non-critical - } - } + storeStreamingSemanticCacheResponse({ + enabled: semanticCacheEnabled, + streamStatus, + streamResponseBody, + body, + headers: clientRawRequest?.headers, + model, + apiKeyId: apiKeyInfo?.id ?? undefined, + streamUsage, + log, + }); }; const streamFailureFinalizers = streamFailure.createStreamFailureFinalizers({ diff --git a/open-sse/handlers/chatCore/streamingSemanticCacheStore.ts b/open-sse/handlers/chatCore/streamingSemanticCacheStore.ts new file mode 100644 index 0000000000..3aa16a9781 --- /dev/null +++ b/open-sse/handlers/chatCore/streamingSemanticCacheStore.ts @@ -0,0 +1,95 @@ +/** + * chatCore streaming semantic-cache store (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Extracted from handleChatCore's onStreamComplete callback: after a 200 streaming response is + * assembled, store it under its signature so a future temp=0 request can be served from cache. + * Side-effect only (cache write + debug log), wrapped in fail-open try/catch. Behaviour is + * byte-identical to the previous inline block — including the `_streamed` strip, the early + * skip-on-too-large, and the `Number(...) || 0` token accounting. The early return was the last + * statement of the callback, so returning from this helper is equivalent. + */ +import { + generateSignature as defaultGenerateSignature, + setCachedResponse as defaultSetCachedResponse, + isCacheableForWrite as defaultIsCacheableForWrite, +} from "@/lib/semanticCache"; +import { isSmallEnoughForSemanticCache as defaultIsSmallEnough } from "../../utils/estimateSize.ts"; + +type LoggerLike = { debug?: (...args: unknown[]) => void } | null | undefined; + +type CacheBody = { + messages?: unknown; + input?: unknown; + temperature?: unknown; + top_p?: unknown; +}; + +export interface StreamingSemanticCacheStoreDeps { + isCacheableForWrite: typeof defaultIsCacheableForWrite; + isSmallEnoughForSemanticCache: typeof defaultIsSmallEnough; + generateSignature: typeof defaultGenerateSignature; + setCachedResponse: typeof defaultSetCachedResponse; +} + +const DEFAULT_DEPS: StreamingSemanticCacheStoreDeps = { + isCacheableForWrite: defaultIsCacheableForWrite, + isSmallEnoughForSemanticCache: defaultIsSmallEnough, + generateSignature: defaultGenerateSignature, + setCachedResponse: defaultSetCachedResponse, +}; + +interface StreamingCacheArgs { + enabled: boolean; + streamStatus: number; + streamResponseBody: Record | null | undefined; + body: CacheBody; + headers: unknown; + model: string; + apiKeyId?: string | number; + streamUsage?: Record | null; + log?: LoggerLike; +} + +function streamTokensSaved(streamUsage: Record | null | undefined): number { + const u = streamUsage as Record | null; + return (Number(u?.prompt_tokens ?? 0) || 0) + (Number(u?.completion_tokens ?? 0) || 0); +} + +function writeStreamingCacheEntry( + args: StreamingCacheArgs, + deps: StreamingSemanticCacheStoreDeps +): void { + try { + const cleanBody = { ...(args.streamResponseBody as Record) }; + delete cleanBody._streamed; + if (!deps.isSmallEnoughForSemanticCache(cleanBody)) return; + const sig = deps.generateSignature( + args.model, + args.body.messages ?? args.body.input, + args.body.temperature, + args.body.top_p, + args.apiKeyId ?? undefined + ); + const tokensSaved = streamTokensSaved(args.streamUsage); + deps.setCachedResponse(sig, args.model, cleanBody, tokensSaved); + args.log?.debug?.("CACHE", `Stored streaming response for ${args.model} (${tokensSaved} tokens)`); + } catch { + // Cache write failed — non-critical + } +} + +export function storeStreamingSemanticCacheResponse( + args: StreamingCacheArgs, + deps: StreamingSemanticCacheStoreDeps = DEFAULT_DEPS +): void { + if ( + !args.enabled || + args.streamStatus !== 200 || + !args.streamResponseBody || + !deps.isCacheableForWrite(args.body, args.headers) + ) { + return; + } + writeStreamingCacheEntry(args, deps); +} diff --git a/tests/unit/chatcore-streaming-cache-store.test.ts b/tests/unit/chatcore-streaming-cache-store.test.ts new file mode 100644 index 0000000000..8ee5fb264c --- /dev/null +++ b/tests/unit/chatcore-streaming-cache-store.test.ts @@ -0,0 +1,96 @@ +// Characterization of storeStreamingSemanticCacheResponse — the streaming Phase 9.1 cache store +// extracted from handleChatCore's onStreamComplete (chatCore god-file decomposition, #3501). Deps +// are injected so the gate + `_streamed` strip + signature + token accounting are observable +// without the real cache backend. Locks: the 4-way gate (enabled + status 200 + body + cacheable), +// the `_streamed` strip, skip-on-too-large, `Number(...) || 0` tokens, and fail-open on throw. +import { test } from "node:test"; +import assert from "node:assert/strict"; + +const { storeStreamingSemanticCacheResponse } = await import( + "../../open-sse/handlers/chatCore/streamingSemanticCacheStore.ts" +); + +type Stored = { sig: unknown; model: string; body: Record; tokens: number }; + +function makeDeps(overrides: Record = {}) { + const stored: Stored[] = []; + const deps = { + isCacheableForWrite: () => true, + isSmallEnoughForSemanticCache: () => true, + generateSignature: (...a: unknown[]) => `sig:${JSON.stringify(a)}`, + setCachedResponse: (sig: unknown, model: string, body: Record, tokens: number) => + stored.push({ sig, model, body, tokens }), + ...overrides, + } as Parameters[1]; + return { deps, stored }; +} + +function baseArgs(overrides: Record = {}) { + return { + enabled: true, + streamStatus: 200, + streamResponseBody: { id: "resp-1", _streamed: true }, + body: { messages: [{ role: "user", content: "hi" }], temperature: 0, top_p: 1 }, + headers: undefined, + model: "gpt-x", + apiKeyId: "key-1", + streamUsage: { prompt_tokens: 12, completion_tokens: 8 }, + log: undefined, + ...overrides, + } as Parameters[0]; +} + +test("happy path → stores cleaned body (no _streamed), tokens = prompt + completion", () => { + const { deps, stored } = makeDeps(); + storeStreamingSemanticCacheResponse(baseArgs(), deps); + assert.equal(stored.length, 1); + assert.equal(stored[0].model, "gpt-x"); + assert.equal(stored[0].tokens, 20); + assert.equal("_streamed" in stored[0].body, false); + assert.equal(stored[0].body.id, "resp-1"); +}); + +test("non-200 stream status → no store", () => { + const { deps, stored } = makeDeps(); + storeStreamingSemanticCacheResponse(baseArgs({ streamStatus: 500 }), deps); + assert.equal(stored.length, 0); +}); + +test("disabled → no store", () => { + const { deps, stored } = makeDeps(); + storeStreamingSemanticCacheResponse(baseArgs({ enabled: false }), deps); + assert.equal(stored.length, 0); +}); + +test("missing response body → no store", () => { + const { deps, stored } = makeDeps(); + storeStreamingSemanticCacheResponse(baseArgs({ streamResponseBody: null }), deps); + assert.equal(stored.length, 0); +}); + +test("not cacheable-for-write → no store", () => { + const { deps, stored } = makeDeps({ isCacheableForWrite: () => false }); + storeStreamingSemanticCacheResponse(baseArgs(), deps); + assert.equal(stored.length, 0); +}); + +test("too large → no store (early return inside try)", () => { + const { deps, stored } = makeDeps({ isSmallEnoughForSemanticCache: () => false }); + storeStreamingSemanticCacheResponse(baseArgs(), deps); + assert.equal(stored.length, 0); +}); + +test("missing usage → tokens coerce to 0", () => { + const { deps, stored } = makeDeps(); + storeStreamingSemanticCacheResponse(baseArgs({ streamUsage: null }), deps); + assert.equal(stored[0].tokens, 0); +}); + +test("a throwing dep is swallowed (fail-open, non-critical)", () => { + const { deps } = makeDeps({ + setCachedResponse: () => { + throw new Error("cache write boom"); + }, + }); + assert.doesNotThrow(() => storeStreamingSemanticCacheResponse(baseArgs(), deps)); +});