diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index d8070256ab..493387af45 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -3013,6 +3013,27 @@ export async function handleChatCore({ } } + // Reasoning Replay Cache (#1628): Capture reasoning_content from non-streaming responses + // with tool_calls so it can be replayed on subsequent turns (DeepSeek V4, Kimi K2, etc.) + try { + const firstChoice = translatedResponse?.choices?.[0]; + const msg = firstChoice?.message; + if ( + msg?.role === "assistant" && + Array.isArray(msg.tool_calls) && + msg.tool_calls.length > 0 && + typeof msg.reasoning_content === "string" && + msg.reasoning_content.length > 0 + ) { + const { cacheReasoningBatch } = require("../services/reasoningCache"); + const toolIds = msg.tool_calls.map((tc: { id?: string }) => tc.id).filter(Boolean); + if (toolIds.length > 0) { + cacheReasoningBatch(toolIds, provider, model, msg.reasoning_content); + } + } + } catch { + // Cache capture is non-critical — never block the response + } // Sanitize response for OpenAI SDK compatibility // Strips non-standard fields (x_groq, usage_breakdown, service_tier, etc.) // Extracts and tags into reasoning_content @@ -3239,6 +3260,31 @@ export async function handleChatCore({ }); } + // Reasoning Replay Cache (#1628): Capture reasoning_content from streaming responses + // with tool_calls so it can be replayed on subsequent turns (DeepSeek V4, Kimi K2, etc.) + if (streamStatus === 200 && streamResponseBody) { + try { + const body = streamResponseBody as Record; + const choices = body.choices as { message?: Record }[] | undefined; + const msg = choices?.[0]?.message; + if ( + msg?.role === "assistant" && + Array.isArray(msg.tool_calls) && + msg.tool_calls.length > 0 && + typeof msg.reasoning_content === "string" && + (msg.reasoning_content as string).length > 0 + ) { + const { cacheReasoningBatch } = require("../services/reasoningCache"); + const toolIds = (msg.tool_calls as { id?: string }[]).map((tc) => tc.id).filter(Boolean); + if (toolIds.length > 0) { + cacheReasoningBatch(toolIds, provider, model, msg.reasoning_content as string); + } + } + } catch { + // Cache capture is non-critical — never block the stream + } + } + // Track cache token metrics for streaming responses if (streamUsage && typeof streamUsage === "object") { const inputTokens = streamUsage.prompt_tokens || 0; diff --git a/open-sse/services/reasoningCache.ts b/open-sse/services/reasoningCache.ts new file mode 100644 index 0000000000..4d7d10f857 --- /dev/null +++ b/open-sse/services/reasoningCache.ts @@ -0,0 +1,314 @@ +/** + * Reasoning Replay Cache — In-memory + DB hybrid service. + * + * Captures reasoning_content from thinking-mode model responses (DeepSeek V4, + * Kimi K2, Qwen-Thinking, GLM, etc.) and re-injects it on subsequent turns + * when clients omit it. This prevents the DeepSeek 400 error: + * "The reasoning_content in the thinking mode must be passed back to the API." + * + * Architecture: + * Write: memory + DB simultaneously (fire-and-forget DB write) + * Read: memory first → DB fallback → miss + * + * @see Issue #1628 + */ + +// ──────────────── Provider/Model Detection ──────────────── + +const REASONING_REPLAY_PROVIDERS = new Set([ + "deepseek", + "opencode-go", + "siliconflow", + "nebius", + "deepinfra", + "sambanova", + "fireworks", + "together", +]); + +const REASONING_REPLAY_MODEL_PATTERNS = [ + /deepseek-r1/i, + /deepseek-reasoner/i, + /deepseek-chat/i, + /kimi-k2/i, + /qwq/i, + /qwen.*think/i, + /glm.*think/i, +]; + +/** + * Check if a provider/model combination requires reasoning replay. + */ +export function requiresReasoningReplay(provider: string, model: string): boolean { + if (REASONING_REPLAY_PROVIDERS.has(provider)) return true; + return REASONING_REPLAY_MODEL_PATTERNS.some((p) => p.test(model)); +} + +// ──────────────── In-Memory Cache ──────────────── + +interface MemoryCacheEntry { + reasoning: string; + provider: string; + model: string; + expiresAt: number; + createdAt: number; +} + +const memoryCache = new Map(); +const MAX_MEMORY_ENTRIES = 2000; +const TTL_MS = 2 * 60 * 60 * 1000; // 2 hours + +// ──────────────── Counters ──────────────── + +let hits = 0; +let misses = 0; +let replays = 0; + +// ──────────────── Core Operations ──────────────── + +/** + * Evict the oldest entry from the memory cache when full. + */ +function evictOldest(): void { + let oldestKey: string | null = null; + let oldestTime = Infinity; + for (const [key, entry] of memoryCache) { + if (entry.createdAt < oldestTime) { + oldestTime = entry.createdAt; + oldestKey = key; + } + } + if (oldestKey) memoryCache.delete(oldestKey); +} + +/** + * Remove expired entries from memory cache. + */ +function purgeExpiredMemory(): void { + const now = Date.now(); + for (const [key, entry] of memoryCache) { + if (now >= entry.expiresAt) { + memoryCache.delete(key); + } + } +} + +/** + * Cache a reasoning_content string for one or more tool_call IDs. + * Writes to both memory and DB (DB write is fire-and-forget). + */ +export function cacheReasoning( + toolCallId: string, + provider: string, + model: string, + reasoning: string +): void { + if (!toolCallId || !reasoning) return; + + const now = Date.now(); + + // Memory write + if (memoryCache.size >= MAX_MEMORY_ENTRIES) { + evictOldest(); + } + memoryCache.set(toolCallId, { + reasoning, + provider, + model, + expiresAt: now + TTL_MS, + createdAt: now, + }); + + // DB write (fire-and-forget — don't block the hot path) + try { + const { setReasoningCache } = require("@/lib/db/reasoningCache"); + setReasoningCache(toolCallId, provider, model, reasoning, TTL_MS); + } catch { + // DB write failure is non-fatal; memory cache still serves + } +} + +/** + * Cache reasoning for multiple tool_call IDs (same reasoning content). + */ +export function cacheReasoningBatch( + toolCallIds: string[], + provider: string, + model: string, + reasoning: string +): void { + for (const id of toolCallIds) { + if (id) cacheReasoning(id, provider, model, reasoning); + } +} + +/** + * Look up cached reasoning_content by tool_call_id. + * Memory first → DB fallback → null (miss). + */ +export function lookupReasoning(toolCallId: string): string | null { + if (!toolCallId) { + misses++; + return null; + } + + // 1. Check memory + const mem = memoryCache.get(toolCallId); + if (mem) { + if (Date.now() < mem.expiresAt) { + hits++; + return mem.reasoning; + } + // Expired in memory — remove + memoryCache.delete(toolCallId); + } + + // 2. Fallback to DB + try { + const { getReasoningCache } = require("@/lib/db/reasoningCache"); + const dbResult = getReasoningCache(toolCallId); + if (dbResult) { + hits++; + // Promote back to memory for fast subsequent lookups + memoryCache.set(toolCallId, { + reasoning: dbResult.reasoning, + provider: dbResult.provider, + model: dbResult.model, + expiresAt: Date.now() + TTL_MS, + createdAt: Date.now(), + }); + return dbResult.reasoning; + } + } catch { + // DB read failure is non-fatal + } + + // 3. Miss + misses++; + return null; +} + +/** + * Record that a replay was performed (for dashboard metrics). + */ +export function recordReplay(): void { + replays++; +} + +// ──────────────── Stats & Dashboard ──────────────── + +/** + * Get combined stats from memory + DB + counters. + */ +export function getReasoningCacheServiceStats(): { + memoryEntries: number; + dbEntries: number; + totalEntries: number; + totalChars: number; + hits: number; + misses: number; + replays: number; + replayRate: string; + byProvider: Record; + byModel: Record; + oldestEntry: string | null; + newestEntry: string | null; +} { + // Purge expired memory entries before reporting + purgeExpiredMemory(); + + let dbStats = { + totalEntries: 0, + totalChars: 0, + byProvider: {} as Record, + byModel: {} as Record, + oldestEntry: null as string | null, + newestEntry: null as string | null, + }; + + try { + const { getReasoningCacheStats } = require("@/lib/db/reasoningCache"); + dbStats = getReasoningCacheStats(); + } catch { + // DB unavailable — return memory-only stats + } + + const totalLookups = hits + misses; + const replayRate = totalLookups > 0 ? ((replays / totalLookups) * 100).toFixed(1) : "0.0"; + + return { + memoryEntries: memoryCache.size, + dbEntries: dbStats.totalEntries, + totalEntries: dbStats.totalEntries, + totalChars: dbStats.totalChars, + hits, + misses, + replays, + replayRate: `${replayRate}%`, + byProvider: dbStats.byProvider, + byModel: dbStats.byModel, + oldestEntry: dbStats.oldestEntry, + newestEntry: dbStats.newestEntry, + }; +} + +/** + * Get paginated entries (delegates to DB). + */ +export function getReasoningCacheServiceEntries( + opts: { + limit?: number; + offset?: number; + provider?: string; + model?: string; + } = {} +): unknown[] { + try { + const { getReasoningCacheEntries } = require("@/lib/db/reasoningCache"); + return getReasoningCacheEntries(opts); + } catch { + return []; + } +} + +/** + * Clear all reasoning cache entries (memory + DB). + * Returns count of DB entries removed. + */ +export function clearReasoningCacheAll(provider?: string): number { + // Clear memory + if (provider) { + for (const [key, entry] of memoryCache) { + if (entry.provider === provider) memoryCache.delete(key); + } + } else { + memoryCache.clear(); + } + + // Reset counters + hits = 0; + misses = 0; + replays = 0; + + // Clear DB + try { + const { clearAllReasoningCache } = require("@/lib/db/reasoningCache"); + return clearAllReasoningCache(provider); + } catch { + return 0; + } +} + +/** + * Cleanup expired entries from both memory and DB. + * Called periodically (e.g., every 30 min from health-check). + */ +export function cleanupReasoningCache(): number { + purgeExpiredMemory(); + try { + const { cleanupExpiredReasoning } = require("@/lib/db/reasoningCache"); + return cleanupExpiredReasoning(); + } catch { + return 0; + } +} diff --git a/open-sse/translator/index.ts b/open-sse/translator/index.ts index feb8a5d375..57649cc8c2 100644 --- a/open-sse/translator/index.ts +++ b/open-sse/translator/index.ts @@ -204,21 +204,51 @@ export function translateRequest( result.tools = sanitizeToolDescriptions(result.tools); } - // Inject reasoning_content = "" for DeepSeek/Reasoning models assistant messages with tool_calls - // if omitted by the client, to avoid upstream 400 errors (e.g. "Messages with role 'assistant' that contain tool_calls must also include reasoning_content") + // Reasoning Replay Cache (#1628): Re-inject cached reasoning_content for + // thinking-mode models (DeepSeek V4, Kimi K2, Qwen-Thinking, etc.) when + // clients omit it from the conversation history. Without this, DeepSeek V4 + // returns 400: "The reasoning_content in the thinking mode must be passed + // back to the API." const isReasoner = provider === "deepseek" || provider === "opencode-go" || (typeof model === "string" && /r1|reason|kimi-k2/i.test(model)); if (isReasoner && result.messages && Array.isArray(result.messages)) { + // Lazy-load to avoid circular dependency issues at module load + let lookupReasoning: ((id: string) => string | null) | null = null; + let recordReplay: (() => void) | null = null; + try { + const cache = require("../services/reasoningCache"); + lookupReasoning = cache.lookupReasoning; + recordReplay = cache.recordReplay; + } catch { + // Cache module unavailable — fall through to legacy behavior + } + for (const msg of result.messages) { - if ( - msg.role === "assistant" && - Array.isArray(msg.tool_calls) && - msg.tool_calls.length > 0 && - msg.reasoning_content === undefined - ) { - msg.reasoning_content = ""; + if (msg.role === "assistant" && Array.isArray(msg.tool_calls) && msg.tool_calls.length > 0) { + // Skip if client already provided real reasoning_content + if (typeof msg.reasoning_content === "string" && msg.reasoning_content.length > 0) { + continue; + } + + // Try cache lookup using first tool_call ID + if (lookupReasoning) { + const firstToolId = msg.tool_calls[0]?.id; + if (firstToolId) { + const cached = lookupReasoning(firstToolId); + if (cached) { + msg.reasoning_content = cached; + if (recordReplay) recordReplay(); + continue; + } + } + } + + // Legacy fallback — empty string (works for older DeepSeek versions) + if (msg.reasoning_content === undefined) { + msg.reasoning_content = ""; + } } } } else if ( diff --git a/src/app/(dashboard)/dashboard/cache/components/ReasoningCacheTab.tsx b/src/app/(dashboard)/dashboard/cache/components/ReasoningCacheTab.tsx new file mode 100644 index 0000000000..e4d3ad6d22 --- /dev/null +++ b/src/app/(dashboard)/dashboard/cache/components/ReasoningCacheTab.tsx @@ -0,0 +1,482 @@ +"use client"; + +import { useState, useEffect, useCallback } from "react"; +import { Button, EmptyState } from "@/shared/components"; +import { useNotificationStore } from "@/store/notificationStore"; +import { useTranslations } from "next-intl"; + +// ──────────────── Types ──────────────── + +interface ReasoningCacheEntry { + toolCallId: string; + provider: string; + model: string; + reasoning: string; + charCount: number; + createdAt: string; + expiresAt: string; +} + +interface ReasoningCacheStats { + memoryEntries: number; + dbEntries: number; + totalEntries: number; + totalChars: number; + hits: number; + misses: number; + replays: number; + replayRate: string; + byProvider: Record; + byModel: Record; + oldestEntry: string | null; + newestEntry: string | null; +} + +interface ReasoningCacheData { + stats: ReasoningCacheStats; + entries: ReasoningCacheEntry[]; +} + +// ──────────────── Helpers ──────────────── + +function timeAgo(dateStr: string): string { + const diff = Date.now() - new Date(dateStr).getTime(); + const minutes = Math.floor(diff / 60000); + if (minutes < 1) return "just now"; + if (minutes < 60) return `${minutes}m ago`; + const hours = Math.floor(minutes / 60); + if (hours < 24) return `${hours}h ago`; + const days = Math.floor(hours / 24); + return `${days}d ago`; +} + +function formatChars(chars: number): string { + if (chars >= 1_000_000) return `${(chars / 1_000_000).toFixed(1)}M`; + if (chars >= 1_000) return `${(chars / 1_000).toFixed(1)}K`; + return String(chars); +} + +// ──────────────── Sub-Components ──────────────── + +function StatCard({ + icon, + label, + value, + sub, + accent = "text-text-main", +}: { + icon: string; + label: string; + value: string | number; + sub?: string; + accent?: string; +}) { + return ( +
+
+ + {label} +
+
{value}
+ {sub &&
{sub}
} +
+ ); +} + +function SectionBadge({ + icon, + children, + tone = "neutral", +}: { + icon: string; + children: React.ReactNode; + tone?: "neutral" | "green" | "amber" | "blue"; +}) { + const toneClass = + tone === "green" + ? "border-green-500/20 bg-green-500/10 text-green-300" + : tone === "amber" + ? "border-amber-400/20 bg-amber-400/10 text-amber-300" + : tone === "blue" + ? "border-blue-400/20 bg-blue-400/10 text-blue-300" + : "border-border/40 bg-surface/50 text-text-muted"; + + return ( + + + {children} + + ); +} + +function InfoRow({ icon, children }: { icon: string; children: React.ReactNode }) { + return ( +
+ + {children} +
+ ); +} + +// ──────────────── Main Component ──────────────── + +const REFRESH_INTERVAL_MS = 10_000; + +export default function ReasoningCacheTab() { + const t = useTranslations("cache"); + const notify = useNotificationStore(); + + const [data, setData] = useState(null); + const [loading, setLoading] = useState(true); + const [clearing, setClearing] = useState(false); + const [expandedId, setExpandedId] = useState(null); + + const fetchData = useCallback(async () => { + try { + const res = await fetch("/api/cache/reasoning"); + if (res.ok) { + const json: ReasoningCacheData = await res.json(); + setData(json); + } + } catch (error) { + console.error("[ReasoningCacheTab] Failed to fetch:", error); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + void fetchData(); + const id = setInterval(() => void fetchData(), REFRESH_INTERVAL_MS); + return () => clearInterval(id); + }, [fetchData]); + + const handleClear = async () => { + setClearing(true); + try { + const res = await fetch("/api/cache/reasoning", { method: "DELETE" }); + if (res.ok) { + const result = await res.json(); + notify.success(t("reasoningClearSuccess", { count: result.cleared ?? 0 })); + await fetchData(); + } else { + notify.error(t("reasoningClearError")); + } + } catch { + notify.error(t("reasoningClearError")); + } finally { + setClearing(false); + } + }; + + if (loading) { + return ( +
+
+
+
+ ); + } + + if (!data) { + return ( + void fetchData()} + /> + ); + } + + const { stats, entries } = data; + const providerEntries = Object.entries(stats.byProvider).sort( + ([, a], [, b]) => b.entries - a.entries + ); + const modelEntries = Object.entries(stats.byModel).sort(([, a], [, b]) => b.entries - a.entries); + const totalLookups = stats.hits + stats.misses; + + return ( +
+ {/* Header */} +
+
+ + {t("reasoningCache")} + +
+

{t("reasoningCache")}

+

{t("reasoningCacheDesc")}

+
+
+ +
+ + {/* Stat Cards */} +
+ + + + + +
+ + {/* By Provider */} + {providerEntries.length > 0 && ( +
+

{t("reasoningByProvider")}

+
+ + + + + + + + + + + {providerEntries.map(([prov, d]) => { + const share = + stats.totalEntries > 0 + ? ((d.entries / stats.totalEntries) * 100).toFixed(1) + : "0.0"; + return ( + + + + + + + ); + })} + +
Provider{t("reasoningEntries")}{t("reasoningChars")}Share
{prov} + {d.entries.toLocaleString()} + + {formatChars(d.chars)} + +
+
+
+
+ + {share}% + +
+
+
+
+ )} + + {/* By Model */} + {modelEntries.length > 0 && ( +
+

{t("reasoningByModel")}

+
+ + + + + + + + + + + {modelEntries.map(([mdl, d]) => { + const avgChars = d.entries > 0 ? Math.round(d.chars / d.entries) : 0; + return ( + + + + + + + ); + })} + +
Model{t("reasoningEntries")}Avg Chars{t("reasoningChars")}
{mdl} + {d.entries.toLocaleString()} + + {avgChars.toLocaleString()} + + {formatChars(d.chars)} +
+
+
+ )} + + {/* Recent Entries */} +
+
+

{t("reasoningRecentEntries")}

+

{t("reasoningCacheDesc")}

+
+ + {entries.length === 0 ? ( +
+ {t("reasoningNoData")} +
+ ) : ( +
+
+ {t("reasoningToolCallId")} + Provider + Model + {t("reasoningChars")} + {t("reasoningAge")} + +
+
+ {entries.map((entry) => ( +
+
+
+ {entry.toolCallId} +
+
{entry.provider}
+
+ {entry.model} +
+
+ {entry.charCount.toLocaleString()} +
+
{timeAgo(entry.createdAt)}
+ +
+ + {/* Expanded Detail */} + {expandedId === entry.toolCallId && ( +
+
+ + psychology + + + {t("reasoningDetail")} ({entry.toolCallId}) + +
+ 
+                        {entry.reasoning}
+
+
+ + Provider: {entry.provider} + + + Model: {entry.model} + + + Created:{" "} + + {new Date(entry.createdAt).toLocaleString()} + + + + Expires:{" "} + + {new Date(entry.expiresAt).toLocaleString()} + + + + {t("reasoningChars")}:{" "} + + {entry.charCount.toLocaleString()} + + +
+
+ )} +
+ ))} +
+
+ )} +
+ + {/* Behavior Info */} +
+

{t("reasoningBehavior")}

+
+ {t("reasoningBehaviorCapture")} + {t("reasoningBehaviorReplay")} + {t("reasoningBehaviorFallback")} + {t("reasoningBehaviorTtl")} + {t("reasoningBehaviorModels")} +
+
+
+ ); +} diff --git a/src/app/(dashboard)/dashboard/cache/page.tsx b/src/app/(dashboard)/dashboard/cache/page.tsx index 5b828e1b42..62d7c1cf1a 100644 --- a/src/app/(dashboard)/dashboard/cache/page.tsx +++ b/src/app/(dashboard)/dashboard/cache/page.tsx @@ -5,6 +5,7 @@ import { Card, Button, EmptyState } from "@/shared/components"; import { useNotificationStore } from "@/store/notificationStore"; import { useTranslations } from "next-intl"; import CacheEntriesTab from "./components/CacheEntriesTab"; +import ReasoningCacheTab from "./components/ReasoningCacheTab"; interface SemanticCacheStats { memoryEntries: number; @@ -63,7 +64,7 @@ interface CacheStats { config?: CacheConfig; } -type CacheView = "prompt" | "semantic"; +type CacheView = "prompt" | "semantic" | "reasoning"; function StatCard({ icon, @@ -469,6 +470,18 @@ export default function CachePage() { > {t("semanticCache")} +
{loading && ( @@ -818,6 +831,14 @@ export default function CachePage() {
)} + + {activeView === "reasoning" && ( + +
+ +
+ + )} ); } diff --git a/src/app/api/cache/reasoning/route.ts b/src/app/api/cache/reasoning/route.ts new file mode 100644 index 0000000000..e5d144fa47 --- /dev/null +++ b/src/app/api/cache/reasoning/route.ts @@ -0,0 +1,71 @@ +import { NextRequest, NextResponse } from "next/server"; +import { isAuthenticated } from "@/shared/utils/apiAuth"; + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +/** + * GET /api/cache/reasoning + * + * Returns reasoning replay cache stats + paginated entries. + * Query params: ?provider=deepseek&model=deepseek-reasoner&limit=50&offset=0 + */ +export async function GET(req: NextRequest) { + if (!(await isAuthenticated(req))) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + try { + const { getReasoningCacheServiceStats, getReasoningCacheServiceEntries } = + await import("@omniroute/open-sse/services/reasoningCache"); + + const { searchParams } = new URL(req.url); + const provider = searchParams.get("provider") || undefined; + const model = searchParams.get("model") || undefined; + const limit = parseInt(searchParams.get("limit") || "50", 10); + const offset = parseInt(searchParams.get("offset") || "0", 10); + + const stats = getReasoningCacheServiceStats(); + const entries = getReasoningCacheServiceEntries({ + limit: Math.min(Math.max(limit, 1), 200), + offset: Math.max(offset, 0), + provider, + model, + }); + + return NextResponse.json({ stats, entries }); + } catch (error) { + return NextResponse.json({ error: errorMessage(error) }, { status: 500 }); + } +} + +/** + * DELETE /api/cache/reasoning + * + * Clears reasoning cache entries. + * Query params: ?provider=deepseek (filter by provider) or no params (clear all). + */ +export async function DELETE(req: NextRequest) { + if (!(await isAuthenticated(req))) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + try { + const { clearReasoningCacheAll } = await import("@omniroute/open-sse/services/reasoningCache"); + + const { searchParams } = new URL(req.url); + const provider = searchParams.get("provider") || undefined; + + const cleared = clearReasoningCacheAll(provider); + + return NextResponse.json({ + ok: true, + cleared, + scope: provider ? "provider" : "all", + ...(provider ? { provider } : {}), + }); + } catch (error) { + return NextResponse.json({ error: errorMessage(error) }, { status: 500 }); + } +} diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index 6461ef2959..5e870fb7fb 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -4431,7 +4431,32 @@ "savedCalls": "Saved API Calls", "totalProcessed": "Total Requests Processed", "disabled": "Disabled", - "totalRequests": "Total Requests" + "totalRequests": "Total Requests", + "reasoningCache": "Reasoning Replay", + "reasoningCacheDesc": "Preserves model thinking for multi-turn tool calling flows", + "reasoningEntries": "Active Entries", + "reasoningReplayRate": "Replay Rate", + "reasoningReplays": "Total Replays", + "reasoningCharsCached": "Characters Cached", + "reasoningMisses": "Cache Misses", + "reasoningByProvider": "By Provider", + "reasoningByModel": "By Model", + "reasoningRecentEntries": "Recent Entries", + "reasoningToolCallId": "Tool Call ID", + "reasoningChars": "Characters", + "reasoningAge": "Age", + "reasoningView": "View", + "reasoningDetail": "Reasoning Content", + "reasoningBehavior": "Behavior", + "reasoningBehaviorCapture": "Captures reasoning_content from streaming responses", + "reasoningBehaviorReplay": "Re-injects on next turn when client omits it", + "reasoningBehaviorFallback": "Memory-first with SQLite fallback for crash recovery", + "reasoningBehaviorTtl": "TTL: 2 hours | Max entries: 2,000 (memory)", + "reasoningBehaviorModels": "Supported: DeepSeek, Kimi, Qwen-Thinking, GLM", + "reasoningClearAll": "Clear Reasoning Cache", + "reasoningClearSuccess": "Cleared {count} reasoning cache entries", + "reasoningClearError": "Failed to clear reasoning cache", + "reasoningNoData": "No reasoning entries cached yet. Entries appear when thinking models use tool calling." }, "proxyConfigModal": { "levelGlobal": "Global", diff --git a/src/lib/db/migrations/032_create_reasoning_cache.sql b/src/lib/db/migrations/032_create_reasoning_cache.sql new file mode 100644 index 0000000000..1d1efa1eb5 --- /dev/null +++ b/src/lib/db/migrations/032_create_reasoning_cache.sql @@ -0,0 +1,27 @@ +-- 032_create_reasoning_cache.sql +-- Persistent storage for reasoning_content replay cache. +-- Enables crash recovery and dashboard visibility for the +-- Reasoning Replay Cache feature (Issue #1628). +-- +-- When thinking-mode models (DeepSeek V4, Kimi K2, Qwen-Thinking, etc.) +-- generate tool_calls, their reasoning_content must be replayed on the +-- next turn. This table persists that content keyed by tool_call_id. + +CREATE TABLE IF NOT EXISTS reasoning_cache ( + tool_call_id TEXT PRIMARY KEY, + provider TEXT NOT NULL, + model TEXT NOT NULL, + reasoning TEXT NOT NULL, + char_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + expires_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_reasoning_cache_expires + ON reasoning_cache(expires_at); +CREATE INDEX IF NOT EXISTS idx_reasoning_cache_provider + ON reasoning_cache(provider); +CREATE INDEX IF NOT EXISTS idx_reasoning_cache_model + ON reasoning_cache(model); +CREATE INDEX IF NOT EXISTS idx_reasoning_cache_created + ON reasoning_cache(created_at); diff --git a/src/lib/db/reasoningCache.ts b/src/lib/db/reasoningCache.ts new file mode 100644 index 0000000000..146b3bef68 --- /dev/null +++ b/src/lib/db/reasoningCache.ts @@ -0,0 +1,240 @@ +/** + * reasoningCache.ts — DB domain module for the Reasoning Replay Cache. + * + * Persists reasoning_content from thinking-mode models (DeepSeek V4, Kimi K2, + * Qwen-Thinking, etc.) keyed by tool_call_id. Used for crash recovery and + * dashboard visibility. The hot path uses the in-memory cache in + * open-sse/services/reasoningCache.ts; this module is the persistence layer. + * + * @see Issue #1628 + */ + +import { getDbInstance } from "./core"; + +// ──────────────── Types ──────────────── + +export interface ReasoningCacheEntry { + toolCallId: string; + provider: string; + model: string; + reasoning: string; + charCount: number; + createdAt: string; + expiresAt: string; +} + +export interface ReasoningCacheStats { + totalEntries: number; + totalChars: number; + byProvider: Record; + byModel: Record; + oldestEntry: string | null; + newestEntry: string | null; +} + +// ──────────────── Constants ──────────────── + +const DEFAULT_TTL_MS = 2 * 60 * 60 * 1000; // 2 hours + +// ──────────────── CRUD ──────────────── + +/** + * Store a reasoning_content entry for a given tool_call_id. + * Uses INSERT OR REPLACE to handle duplicate tool_call_ids gracefully. + */ +export function setReasoningCache( + toolCallId: string, + provider: string, + model: string, + reasoning: string, + ttlMs: number = DEFAULT_TTL_MS +): void { + const db = getDbInstance(); + const expiresAt = new Date(Date.now() + ttlMs).toISOString(); + const charCount = reasoning.length; + + db.prepare( + `INSERT OR REPLACE INTO reasoning_cache + (tool_call_id, provider, model, reasoning, char_count, created_at, expires_at) + VALUES (?, ?, ?, ?, ?, datetime('now'), ?)` + ).run(toolCallId, provider, model, reasoning, charCount, expiresAt); +} + +/** + * Retrieve a cached reasoning_content by tool_call_id. + * Returns null if not found or expired. + */ +export function getReasoningCache( + toolCallId: string +): { reasoning: string; provider: string; model: string } | null { + const db = getDbInstance(); + const row = db + .prepare( + `SELECT reasoning, provider, model FROM reasoning_cache + WHERE tool_call_id = ? AND expires_at > datetime('now')` + ) + .get(toolCallId) as { reasoning: string; provider: string; model: string } | undefined; + + return row ?? null; +} + +/** + * Delete a specific reasoning cache entry. + */ +export function deleteReasoningCache(toolCallId: string): void { + const db = getDbInstance(); + db.prepare(`DELETE FROM reasoning_cache WHERE tool_call_id = ?`).run(toolCallId); +} + +/** + * Delete all expired entries. Returns count of rows removed. + */ +export function cleanupExpiredReasoning(): number { + const db = getDbInstance(); + const result = db + .prepare(`DELETE FROM reasoning_cache WHERE expires_at <= datetime('now')`) + .run(); + return result.changes; +} + +/** + * Delete all entries, optionally filtered by provider. + * Returns count of rows removed. + */ +export function clearAllReasoningCache(provider?: string): number { + const db = getDbInstance(); + if (provider) { + const result = db.prepare(`DELETE FROM reasoning_cache WHERE provider = ?`).run(provider); + return result.changes; + } + const result = db.prepare(`DELETE FROM reasoning_cache`).run(); + return result.changes; +} + +// ──────────────── Stats ──────────────── + +/** + * Get aggregate statistics for the reasoning cache. + */ +export function getReasoningCacheStats(): ReasoningCacheStats { + const db = getDbInstance(); + + // Total counts + const totals = db + .prepare( + `SELECT COUNT(*) as total_entries, COALESCE(SUM(char_count), 0) as total_chars + FROM reasoning_cache WHERE expires_at > datetime('now')` + ) + .get() as { total_entries: number; total_chars: number }; + + // By provider + const providerRows = db + .prepare( + `SELECT provider, COUNT(*) as entries, COALESCE(SUM(char_count), 0) as chars + FROM reasoning_cache WHERE expires_at > datetime('now') + GROUP BY provider ORDER BY entries DESC` + ) + .all() as { provider: string; entries: number; chars: number }[]; + + const byProvider: Record = {}; + for (const row of providerRows) { + byProvider[row.provider] = { entries: row.entries, chars: row.chars }; + } + + // By model + const modelRows = db + .prepare( + `SELECT model, COUNT(*) as entries, COALESCE(SUM(char_count), 0) as chars + FROM reasoning_cache WHERE expires_at > datetime('now') + GROUP BY model ORDER BY entries DESC` + ) + .all() as { model: string; entries: number; chars: number }[]; + + const byModel: Record = {}; + for (const row of modelRows) { + byModel[row.model] = { entries: row.entries, chars: row.chars }; + } + + // Oldest/newest + const oldest = db + .prepare( + `SELECT created_at FROM reasoning_cache + WHERE expires_at > datetime('now') ORDER BY created_at ASC LIMIT 1` + ) + .get() as { created_at: string } | undefined; + + const newest = db + .prepare( + `SELECT created_at FROM reasoning_cache + WHERE expires_at > datetime('now') ORDER BY created_at DESC LIMIT 1` + ) + .get() as { created_at: string } | undefined; + + return { + totalEntries: totals.total_entries, + totalChars: totals.total_chars, + byProvider, + byModel, + oldestEntry: oldest?.created_at ?? null, + newestEntry: newest?.created_at ?? null, + }; +} + +// ──────────────── Paginated Entries ──────────────── + +/** + * List reasoning cache entries with optional filters and pagination. + */ +export function getReasoningCacheEntries( + opts: { + limit?: number; + offset?: number; + provider?: string; + model?: string; + } = {} +): ReasoningCacheEntry[] { + const db = getDbInstance(); + const limit = Math.min(opts.limit ?? 50, 200); + const offset = opts.offset ?? 0; + + const conditions: string[] = ["expires_at > datetime('now')"]; + const params: unknown[] = []; + + if (opts.provider) { + conditions.push("provider = ?"); + params.push(opts.provider); + } + if (opts.model) { + conditions.push("model = ?"); + params.push(opts.model); + } + + const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; + + const rows = db + .prepare( + `SELECT tool_call_id, provider, model, reasoning, char_count, created_at, expires_at + FROM reasoning_cache ${where} + ORDER BY created_at DESC + LIMIT ? OFFSET ?` + ) + .all(...params, limit, offset) as { + tool_call_id: string; + provider: string; + model: string; + reasoning: string; + char_count: number; + created_at: string; + expires_at: string; + }[]; + + return rows.map((row) => ({ + toolCallId: row.tool_call_id, + provider: row.provider, + model: row.model, + reasoning: row.reasoning, + charCount: row.char_count, + createdAt: row.created_at, + expiresAt: row.expires_at, + })); +} diff --git a/src/lib/localDb.ts b/src/lib/localDb.ts index bc2ce0a2dd..e8d23b664b 100755 --- a/src/lib/localDb.ts +++ b/src/lib/localDb.ts @@ -321,3 +321,16 @@ export { getAllPersistedCreditBalances, persistCreditBalance, } from "./db/creditBalance"; + +export { + // Reasoning Replay Cache (#1628) + setReasoningCache, + getReasoningCache, + deleteReasoningCache, + cleanupExpiredReasoning, + getReasoningCacheStats, + getReasoningCacheEntries, + clearAllReasoningCache, +} from "./db/reasoningCache"; + +export type { ReasoningCacheEntry, ReasoningCacheStats } from "./db/reasoningCache"; diff --git a/tests/unit/reasoning-cache.test.ts b/tests/unit/reasoning-cache.test.ts new file mode 100644 index 0000000000..6d2cd97148 --- /dev/null +++ b/tests/unit/reasoning-cache.test.ts @@ -0,0 +1,187 @@ +/** + * Unit tests for the Reasoning Replay Cache (Issue #1628). + * + * Covers: memory cache, DB fallback, hit/miss counters, + * provider detection, and cleanup behavior. + */ + +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; + +// ──────────── Direct service import ──────────── + +import { + cacheReasoning, + cacheReasoningBatch, + lookupReasoning, + recordReplay, + getReasoningCacheServiceStats, + clearReasoningCacheAll, + requiresReasoningReplay, + cleanupReasoningCache, +} from "../../open-sse/services/reasoningCache.ts"; + +describe("Reasoning Replay Cache — Service Layer", () => { + before(() => { + // Start each suite with a clean slate + clearReasoningCacheAll(); + }); + + after(() => { + clearReasoningCacheAll(); + }); + + it("should store and retrieve reasoning by tool_call_id", () => { + cacheReasoning( + "call_test_1", + "deepseek", + "deepseek-reasoner", + "The user wants to read the file..." + ); + const result = lookupReasoning("call_test_1"); + assert.equal(result, "The user wants to read the file..."); + }); + + it("should return null for unknown tool_call_id", () => { + const result = lookupReasoning("call_nonexistent"); + assert.equal(result, null); + }); + + it("should return null for empty tool_call_id", () => { + const result = lookupReasoning(""); + assert.equal(result, null); + }); + + it("should skip caching when reasoning is empty", () => { + cacheReasoning("call_empty", "deepseek", "deepseek-chat", ""); + const result = lookupReasoning("call_empty"); + assert.equal(result, null); + }); + + it("should cache reasoning for multiple tool_call_ids (batch)", () => { + cacheReasoningBatch( + ["call_batch_1", "call_batch_2", "call_batch_3"], + "deepseek", + "deepseek-reasoner", + "Batch reasoning content" + ); + assert.equal(lookupReasoning("call_batch_1"), "Batch reasoning content"); + assert.equal(lookupReasoning("call_batch_2"), "Batch reasoning content"); + assert.equal(lookupReasoning("call_batch_3"), "Batch reasoning content"); + }); + + it("should not overwrite if same tool_call_id is cached again", () => { + cacheReasoning("call_overwrite", "deepseek", "deepseek-chat", "First reasoning"); + cacheReasoning("call_overwrite", "deepseek", "deepseek-chat", "Updated reasoning"); + // Second write wins (INSERT OR REPLACE) + const result = lookupReasoning("call_overwrite"); + assert.equal(result, "Updated reasoning"); + }); + + it("should track hits and misses correctly", () => { + clearReasoningCacheAll(); + + cacheReasoning("call_hit_test", "deepseek", "deepseek-chat", "test reasoning"); + + lookupReasoning("call_hit_test"); // hit + lookupReasoning("call_hit_test"); // hit + lookupReasoning("call_miss_test"); // miss + + const stats = getReasoningCacheServiceStats(); + assert.ok(stats.hits >= 2, `Expected at least 2 hits, got ${stats.hits}`); + assert.ok(stats.misses >= 1, `Expected at least 1 miss, got ${stats.misses}`); + }); + + it("should track replays", () => { + clearReasoningCacheAll(); + + recordReplay(); + recordReplay(); + recordReplay(); + + const stats = getReasoningCacheServiceStats(); + assert.ok(stats.replays >= 3, `Expected at least 3 replays, got ${stats.replays}`); + }); + + it("should report correct stats structure", () => { + clearReasoningCacheAll(); + + cacheReasoning("call_stat_1", "deepseek", "deepseek-reasoner", "Reasoning A"); + cacheReasoning("call_stat_2", "kimi", "kimi-k2.5", "Reasoning B from Kimi"); + + const stats = getReasoningCacheServiceStats(); + + assert.equal(typeof stats.memoryEntries, "number"); + assert.equal(typeof stats.dbEntries, "number"); + assert.equal(typeof stats.totalEntries, "number"); + assert.equal(typeof stats.totalChars, "number"); + assert.equal(typeof stats.hits, "number"); + assert.equal(typeof stats.misses, "number"); + assert.equal(typeof stats.replays, "number"); + assert.equal(typeof stats.replayRate, "string"); + assert.ok(stats.replayRate.endsWith("%")); + assert.equal(typeof stats.byProvider, "object"); + assert.equal(typeof stats.byModel, "object"); + }); + + it("should clear all entries", () => { + cacheReasoning("call_clear_1", "deepseek", "deepseek-chat", "Will be cleared"); + cacheReasoning("call_clear_2", "deepseek", "deepseek-chat", "Also cleared"); + + const count = clearReasoningCacheAll(); + assert.ok(count >= 0); + + assert.equal(lookupReasoning("call_clear_1"), null); + assert.equal(lookupReasoning("call_clear_2"), null); + }); + + it("should cleanup expired reasoning (no-op when nothing expired)", () => { + cacheReasoning("call_cleanup_test", "deepseek", "deepseek-chat", "Not expired yet"); + const cleaned = cleanupReasoningCache(); + assert.equal(typeof cleaned, "number"); + // Entry should still be available since TTL is 2 hours + assert.equal(lookupReasoning("call_cleanup_test"), "Not expired yet"); + }); +}); + +describe("Reasoning Replay Cache — Provider Detection", () => { + it("should detect deepseek as requiring replay", () => { + assert.equal(requiresReasoningReplay("deepseek", "deepseek-chat"), true); + }); + + it("should detect opencode-go as requiring replay", () => { + assert.equal(requiresReasoningReplay("opencode-go", "some-model"), true); + }); + + it("should detect siliconflow as requiring replay", () => { + assert.equal(requiresReasoningReplay("siliconflow", "deepseek-r1"), true); + }); + + it("should detect deepseek-r1 model pattern", () => { + assert.equal(requiresReasoningReplay("unknown-provider", "deepseek-r1"), true); + }); + + it("should detect deepseek-reasoner model pattern", () => { + assert.equal(requiresReasoningReplay("unknown-provider", "deepseek-reasoner"), true); + }); + + it("should detect kimi-k2 model pattern", () => { + assert.equal(requiresReasoningReplay("unknown-provider", "kimi-k2.5"), true); + }); + + it("should detect qwq model pattern", () => { + assert.equal(requiresReasoningReplay("unknown-provider", "qwq-32b-preview"), true); + }); + + it("should detect qwen-thinking model pattern", () => { + assert.equal(requiresReasoningReplay("unknown-provider", "qwen3-thinking-235b"), true); + }); + + it("should NOT detect a generic openai model", () => { + assert.equal(requiresReasoningReplay("openai", "gpt-4o"), false); + }); + + it("should NOT detect claude as requiring replay", () => { + assert.equal(requiresReasoningReplay("anthropic", "claude-opus-4"), false); + }); +});