refactor(chatCore): extrai storeStreamingSemanticCacheResponse (cache-store streaming, #3501) (#4829)

Integrated into release/v3.8.36 (#3501 chatCore extraction stack 12/13)

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<string, unknown> | 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({

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<string, unknown> | null | undefined;
+  body: CacheBody;
+  headers: unknown;
+  model: string;
+  apiKeyId?: string | number;
+  streamUsage?: Record<string, unknown> | null;
+  log?: LoggerLike;
+}
+
+function streamTokensSaved(streamUsage: Record<string, unknown> | null | undefined): number {
+  const u = streamUsage as Record<string, unknown> | 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<string, unknown>) };
+    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);
+}

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<string, unknown>; tokens: number };
+
+function makeDeps(overrides: Record<string, unknown> = {}) {
+  const stored: Stored[] = [];
+  const deps = {
+    isCacheableForWrite: () => true,
+    isSmallEnoughForSemanticCache: () => true,
+    generateSignature: (...a: unknown[]) => `sig:${JSON.stringify(a)}`,
+    setCachedResponse: (sig: unknown, model: string, body: Record<string, unknown>, tokens: number) =>
+      stored.push({ sig, model, body, tokens }),
+    ...overrides,
+  } as Parameters<typeof storeStreamingSemanticCacheResponse>[1];
+  return { deps, stored };
+}
+
+function baseArgs(overrides: Record<string, unknown> = {}) {
+  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<typeof storeStreamingSemanticCacheResponse>[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));
+});