diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index 190cbb44452..0f1320b43d3 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -18,6 +18,8 @@ import { handleForcedSSEToJson } from "./chatCore/sseToJsonHandler.js"; import { handleNonStreamingResponse } from "./chatCore/nonStreamingHandler.js"; import { handleStreamingResponse, buildOnStreamComplete } from "./chatCore/streamingHandler.js"; import { detectClientTool, isNativePassthrough } from "../utils/clientDetector.js"; +import { captureSessionId, extractClientSessionId, getRecentThoughts, saveRecentThought } from "../utils/sessionManager.js"; +import { injectContinuity } from "../rtk/continuity.js"; import { dedupeTools } from "../utils/toolDeduper.js"; import { injectCaveman } from "../rtk/caveman.js"; import { injectPonytail } from "../rtk/ponytail.js"; @@ -34,11 +36,18 @@ import { prefetchRemoteImages } from "../translator/concerns/prefetch.js"; * @param {object} options.credentials - Provider credentials * @param {string} options.sourceFormatOverride - Override detected source format (e.g. "openai-responses") */ -export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, sourceFormatOverride, providerThinking }) { +export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, sourceFormatOverride, providerThinking, continuityEnabled, continuityCount }) { const { provider, model } = modelInfo; const requestStartTime = Date.now(); const sourceFormat = sourceFormatOverride || detectFormat(body); + // Continuity: session ID must be provider-independent so combo switching shares the same thought buffer. + // Use apiKey as scope (stable per client across all combo providers) instead of connectionId (provider-specific). + const continuitySessionId = captureSessionId(body, credentials, apiKey); + // Thought buffer key: decoupled from session ID (which changes on context summarization + // because assistantTextSessionId hashes message content that shifts on truncation/summary). + // Uses client-provided session ID when available (stable across summarization), falls back to apiKey. + const thoughtKey = extractClientSessionId(credentials?.rawHeaders, body) || apiKey || continuitySessionId; // Check for bypass patterns (warmup, skip, cc naming) const bypassResponse = handleBypassRequest(body, model, userAgent, ccFilterNaming); @@ -185,6 +194,15 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred log?.debug?.("PONYTAIL", `${ponytailLevel} | ${finalFormat}`); } + // Continuity: inject framed recent thoughts into system prompt + if (continuityEnabled) { + const recentThoughts = getRecentThoughts(thoughtKey, continuityCount); + if (recentThoughts && recentThoughts.length > 0) { + injectContinuity(translatedBody, finalFormat, recentThoughts); + log?.debug?.("CONTINUITY", `injected ${recentThoughts.length} thoughts into system | ${finalFormat}`); + } + } + const executor = getExecutor(provider); trackPendingRequest(model, provider, connectionId, true); appendRequestLog({ model, provider, connectionId, status: "PENDING" }).catch(() => { }); @@ -308,7 +326,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred return createErrorResult(statusCode, errMsg, resetsAtMs); } - const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess }; + const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, continuityEnabled, saveThinking: continuityEnabled ? (thinking) => { if (thinking) saveRecentThought(thoughtKey, thinking); } : null }; const appendLog = (extra) => appendRequestLog({ model, provider, connectionId, ...extra }).catch(() => { }); const trackDone = () => trackPendingRequest(model, provider, connectionId, false); @@ -326,8 +344,14 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred } // Streaming response - const { onStreamComplete } = buildOnStreamComplete({ ...sharedCtx }); - return handleStreamingResponse({ ...sharedCtx, providerResponse, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, streamController, onStreamComplete }); + const { onStreamComplete: _origOnStreamComplete } = buildOnStreamComplete({ ...sharedCtx }); + const onStreamComplete = (contentObj, usage, ttftAt) => { + if (contentObj?.thinking && sharedCtx.saveThinking) { + sharedCtx.saveThinking(contentObj.thinking); + } + return _origOnStreamComplete(contentObj, usage, ttftAt); + }; + return handleStreamingResponse({ ...sharedCtx, providerResponse, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, streamController, onStreamComplete, continuityEnabled }); } export function isTokenExpiringSoon(expiresAt, bufferMs = 5 * 60 * 1000) { diff --git a/open-sse/handlers/chatCore/nonStreamingHandler.js b/open-sse/handlers/chatCore/nonStreamingHandler.js index 3996f94f942..17b377821d2 100644 --- a/open-sse/handlers/chatCore/nonStreamingHandler.js +++ b/open-sse/handlers/chatCore/nonStreamingHandler.js @@ -4,6 +4,8 @@ import { fromOpenAIFinish } from "../../translator/concerns/finishReason.js"; import { ollamaBodyToOpenAI } from "../../translator/response/ollama-to-openai.js"; import { addBufferToUsage, filterUsageForFormat } from "../../utils/usageTracking.js"; import { createErrorResult } from "../../utils/error.js"; +import { extractThinking } from "../../utils/thinkingExtractor.js"; +import { stripTaggedThinking } from "../../utils/taggedThinkingNormalizer.js"; import { HTTP_STATUS } from "../../config/runtimeConfig.js"; import { parseSSEToOpenAIResponse } from "./sseToJsonHandler.js"; import { buildRequestDetail, extractRequestConfig, extractUsageFromResponse, saveUsageStats } from "./requestDetail.js"; @@ -198,7 +200,7 @@ export function translateNonStreamingResponse(responseBody, targetFormat, source /** * Handle non-streaming response from provider. */ -export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog }) { +export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog, saveThinking, continuityEnabled }) { trackDone(); const contentType = providerResponse.headers.get("content-type") || ""; let responseBody; @@ -258,6 +260,16 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m if (!translatedResponse.created) translatedResponse.created = Math.floor(Date.now() / 1000); } + // Strip content-embedded thinking tags (keep inner text as visible content) and extract thinking (continuity) + if (continuityEnabled) { + stripTaggedThinking(translatedResponse); + + const extractedThinking = extractThinking(translatedResponse); + if (extractedThinking && saveThinking) { + saveThinking(extractedThinking); + } + } + // Strip Azure-specific fields if (!isClaudeMessageResponse) { delete translatedResponse.prompt_filter_results; diff --git a/open-sse/handlers/chatCore/sseToJsonHandler.js b/open-sse/handlers/chatCore/sseToJsonHandler.js index 1e0edebad85..c042c819331 100644 --- a/open-sse/handlers/chatCore/sseToJsonHandler.js +++ b/open-sse/handlers/chatCore/sseToJsonHandler.js @@ -4,6 +4,8 @@ import { HTTP_STATUS } from "../../config/runtimeConfig.js"; import { FORMATS } from "../../translator/formats.js"; import { PROVIDERS } from "../../config/providers.js"; import { buildRequestDetail, extractRequestConfig, saveUsageStats } from "./requestDetail.js"; +import { stripTaggedThinking, stripTags } from "../../utils/taggedThinkingNormalizer.js"; +import { extractThinking } from "../../utils/thinkingExtractor.js"; // Responses-API providers (e.g. codex) may emit SSE without content-type + use Responses output shape const isResponsesProvider = (p) => PROVIDERS[p]?.format === FORMATS.OPENAI_RESPONSES; @@ -102,7 +104,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) { * Handle case: provider forced streaming but client wants JSON. * Supports both Codex/Responses API SSE and standard Chat Completions SSE. */ -export async function handleForcedSSEToJson({ providerResponse, sourceFormat, provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, trackDone, appendLog }) { +export async function handleForcedSSEToJson({ providerResponse, sourceFormat, provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, trackDone, appendLog, continuityEnabled, saveThinking }) { const contentType = providerResponse.headers.get("content-type") || ""; const isSSE = contentType.includes("text/event-stream") || (contentType === "" && isResponsesProvider(provider)); if (!isSSE) return null; // not handled here @@ -126,7 +128,12 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr appendLog({ tokens: usage, status: "200 OK" }); saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint }); - const { msgItem, textContent } = pickAssistantMessageForChatCompletion(jsonResponse.output); + const { msgItem, textContent: rawTextContent } = pickAssistantMessageForChatCompletion(jsonResponse.output); + let textContent = rawTextContent; + if (continuityEnabled) { + const stripped = stripTags(textContent); + if (stripped !== null) textContent = stripped; + } const totalLatency = Date.now() - requestStartTime; saveRequestDetail(buildRequestDetail({ @@ -183,6 +190,12 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr }; } + // Continuity: extract thinking for session buffer + if (continuityEnabled) { + const thinkingForBuffer = extractThinking(jsonResponse); + if (thinkingForBuffer && saveThinking) saveThinking(thinkingForBuffer); + } + return { success: true, response: new Response(JSON.stringify(finalResp), { headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } }) }; } catch (err) { console.error("[ChatCore] Responses API SSE→JSON failed:", err); @@ -215,6 +228,13 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr status: "success" }, { endpoint: clientRawRequest?.endpoint || null })).catch(() => {}); + // Continuity: strip tagged thinking + extract thinking for session buffer + if (continuityEnabled) { + stripTaggedThinking(parsed); + const extractedThinking = extractThinking(parsed); + if (extractedThinking && saveThinking) saveThinking(extractedThinking); + } + // Strip reasoning_content only when content is non-empty. // When content is empty (e.g. thinking models that used all tokens for reasoning), // reasoning_content is the only useful output and must be preserved. diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index 41be3d7d752..9484c50e2b4 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -22,7 +22,7 @@ const CODEX_SOURCE_TO_TARGET = { /** * Determine which SSE transform stream to use based on provider/format. */ -function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey }) { +function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, continuityEnabled }) { const isDroidCLI = userAgent?.toLowerCase().includes("droid") || userAgent?.toLowerCase().includes("codex-cli"); // Responses-API providers (e.g. codex) emit Responses SSE → translate into client format const isResponsesProvider = PROVIDERS[provider]?.format === FORMATS.OPENAI_RESPONSES; @@ -30,20 +30,20 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, if (needsCodexTranslation) { const codexTarget = CODEX_SOURCE_TO_TARGET[sourceFormat] || FORMATS.OPENAI; - return createSSETransformStreamWithLogger(FORMATS.OPENAI_RESPONSES, codexTarget, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey); + return createSSETransformStreamWithLogger(FORMATS.OPENAI_RESPONSES, codexTarget, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, continuityEnabled); } if (needsTranslation(targetFormat, sourceFormat)) { - return createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey); + return createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, continuityEnabled); } - return createPassthroughStreamWithLogger(provider, reqLogger, model, connectionId, body, onStreamComplete, apiKey); + return createPassthroughStreamWithLogger(provider, reqLogger, model, connectionId, body, onStreamComplete, apiKey, continuityEnabled); } /** * Handle streaming response — pipe provider SSE through transform stream to client. */ -export function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete }) { +export function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete, continuityEnabled }) { if (onRequestSuccess) { Promise.resolve() .then(onRequestSuccess) @@ -60,7 +60,7 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou console.warn('[STREAM] ' + provider + ' | ' + model + ' | unexpected Content-Type: ' + upstreamContentType); } - const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey }); + const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, continuityEnabled }); // Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event const isResponsesPassthrough = sourceFormat === FORMATS.OPENAI_RESPONSES && targetFormat === FORMATS.OPENAI_RESPONSES; diff --git a/open-sse/rtk/continuity.js b/open-sse/rtk/continuity.js new file mode 100644 index 00000000000..47eac3174c1 --- /dev/null +++ b/open-sse/rtk/continuity.js @@ -0,0 +1,8 @@ +import { injectSystemPrompt } from "./systemInject.js"; +import { buildContinuityPrompt } from "./continuityPrompt.js"; + +export function injectContinuity(body, format, recentThoughts) { + const framedContext = buildContinuityPrompt(recentThoughts); + if (!framedContext) return; + injectSystemPrompt(body, format, framedContext); +} diff --git a/open-sse/rtk/continuityPrompt.js b/open-sse/rtk/continuityPrompt.js new file mode 100644 index 00000000000..8d8c91aa6c5 --- /dev/null +++ b/open-sse/rtk/continuityPrompt.js @@ -0,0 +1,32 @@ +function maxBacktickRun(text) { + let max = 0, current = 0; + for (const ch of text) { + if (ch === '`') { current++; max = Math.max(max, current); } + else { current = 0; } + } + return max; +} + +export function buildContinuityPrompt(recentThoughts) { + if (!recentThoughts || recentThoughts.length === 0) return null; + const maxRun = recentThoughts.reduce((m, t) => Math.max(m, maxBacktickRun(t)), 0); + const fenceLen = Math.max(5, maxRun + 1); + const fence = '`'.repeat(fenceLen); + return [ + "[HOST CONTINUATION CHECKPOINT]", + "", + "", + "The block below is private continuation context from your immediately preceding turns.", + "It is the causal context behind the prior visible conversation history.", + "Use it to understand why that history unfolded as it did, then continue from the active user input below.", + "", + fence + "text", + recentThoughts.join("\n\n---\n\n"), + fence, + "", + "", + "", + "[HOST RESUME]", + "Continuity restored. Continue from the active input below, using the checkpoint as context for the prior visible history." + ].join("\n"); +} diff --git a/open-sse/utils/continuitySelfCheck.mjs b/open-sse/utils/continuitySelfCheck.mjs new file mode 100644 index 00000000000..28f5bdf3171 --- /dev/null +++ b/open-sse/utils/continuitySelfCheck.mjs @@ -0,0 +1,113 @@ +// continuitySelfCheck.mjs +// Run: node open-sse/utils/continuitySelfCheck.mjs +// No framework. No dependencies. Bails on first failure via assertion. +import assert from "node:assert/strict"; +import { stripTags, stripTaggedThinking } from "./taggedThinkingNormalizer.js"; +import { extractThinking } from "./thinkingExtractor.js"; +import { buildContinuityPrompt } from "../rtk/continuityPrompt.js"; + +const checks = []; + +function check(name, fn) { + checks.push({ name, fn }); +} + +// --- stripTags ----------------------------------------------------------- +check("stripTags: removes known tag markers, keeps inner text", () => { + assert.equal(stripTags("hello"), "hello"); + assert.equal(stripTags("ab"), "ab"); + assert.equal(stripTags("rx"), "rx"); + assert.equal(stripTags("no tags here"), null); + assert.equal(stripTags(123), null); +}); + +// --- stripTaggedThinking ------------------------------------------------- +check("stripTaggedThinking: OpenAI delta.content string leaf", () => { + const obj = { choices: [{ delta: { content: "hi" } }] }; + assert.equal(stripTaggedThinking(obj), true); + assert.equal(obj.choices[0].delta.content, "hi"); +}); + +check("stripTaggedThinking: Claude delta.text + Responses top-level delta string", () => { + const obj = { + delta: { text: "a" }, + response: { type: "response.output_text.delta", delta: "b" }, + }; + assert.equal(stripTaggedThinking(obj), true); + assert.equal(obj.delta.text, "a"); + assert.equal(obj.response.delta, "b"); +}); + +check("stripTaggedThinking: no-op returns false", () => { + const obj = { choices: [{ delta: { content: "clean" } }] }; + assert.equal(stripTaggedThinking(obj), false); +}); + +// --- extractThinking ----------------------------------------------------- +check("extractThinking: OpenAI reasoning_content", () => { + const r = { choices: [{ message: { reasoning_content: "abc" } }] }; + assert.equal(extractThinking(r), "abc"); +}); + +check("extractThinking: Claude content[].type=thinking", () => { + const r = { content: [{ type: "thinking", thinking: "head" }, { type: "text", text: "body" }] }; + assert.equal(extractThinking(r), "head"); +}); + +check("extractThinking: Gemini thought part", () => { + const r = { candidates: [{ content: { parts: [{ text: "g", thought: true }, { text: "c" }] } }] }; + assert.equal(extractThinking(r), "g"); +}); + +check("extractThinking: Responses reasoning item", () => { + const r = { output: [{ type: "reasoning", content: "r1" }, { type: "message", content: [{ text: "m" }] }] }; + assert.equal(extractThinking(r), "r1"); +}); + +check("extractThinking: null on empty", () => { + assert.equal(extractThinking({ choices: [{ message: { content: "x" } }] }), null); +}); + +check("extractThinking: dedupes identical reasoning across fields", () => { + const r = { message: { reasoning_content: "dup", thinking: "dup" } }; + assert.equal(extractThinking(r), "dup"); +}); + +// --- buildContinuityPrompt ---------------------------------------------- +check("buildContinuityPrompt: null on empty", () => { + assert.equal(buildContinuityPrompt([]), null); + assert.equal(buildContinuityPrompt(null), null); +}); + +check("buildContinuityPrompt: wraps content in checkpoint markers", () => { + const p = buildContinuityPrompt(["my-thought"]); + assert.ok(p.includes("[HOST CONTINUATION CHECKPOINT]")); + assert.ok(p.includes("")); + assert.ok(p.includes("my-thought")); + assert.ok(p.includes("[HOST RESUME]")); +}); + +check("buildContinuityPrompt: fence length grows when thoughts contain backticks", () => { + const short = buildContinuityPrompt(["abc"]); + const fenceStartShort = short.match(/(`{5,})text/)[1].length; + // 6 backticks exceed the default minimum (5), forcing fence to grow to 7 + const long = buildContinuityPrompt(["``````code``````"]); + const fenceStartLong = long.match(/(`{5,})text/)[1].length; + assert.ok(fenceStartLong > fenceStartShort, "fence must expand to avoid collision"); + assert.ok(fenceStartLong > 6, "fence must be longer than the longest backtick run in content"); +}); + +// --- runner -------------------------------------------------------------- +let failed = 0; +for (const { name, fn } of checks) { + try { + fn(); + console.log(`ok - ${name}`); + } catch (err) { + failed++; + console.error(`FAIL - ${name}`); + console.error(err); + } +} +console.log(`\n${checks.length - failed}/${checks.length} checks passed`); +if (failed > 0) process.exit(1); diff --git a/open-sse/utils/sessionManager.js b/open-sse/utils/sessionManager.js index 05f908964d3..8fd24112b6e 100644 --- a/open-sse/utils/sessionManager.js +++ b/open-sse/utils/sessionManager.js @@ -13,6 +13,7 @@ import { MEMORY_CONFIG } from "../config/runtimeConfig.js"; // Runtime storage: Key = connectionId, Value = { sessionId, lastUsed } const runtimeSessionStore = new Map(); +const recentThoughtsStore = new Map(); // Periodically evict entries that haven't been used within TTL const cleanupInterval = setInterval(() => { @@ -131,7 +132,7 @@ function extractAntigravitySession(body) { return m ? normalizeSessionId(m[1]) : null; } -function extractClientSessionId(headers, body) { +export function extractClientSessionId(headers, body) { const claude = extractClaudeCodeSession(body?.metadata?.user_id); if (claude) return `claude:${claude}`; const antigravity = extractAntigravitySession(body); @@ -229,3 +230,52 @@ const assistantCleanup = setInterval(() => { } }, MEMORY_CONFIG.sessionCleanupIntervalMs); if (assistantCleanup.unref) assistantCleanup.unref(); + +const MAX_RECENT_THOUGHT_CHARS = 32000; + +function evictLeastRecentlyUsedThoughtSession() { + let oldestKey = null; + let oldestTs = Infinity; + for (const [sid, entry] of recentThoughtsStore) { + if (entry.lastUsed < oldestTs) { + oldestTs = entry.lastUsed; + oldestKey = sid; + } + } + if (oldestKey) recentThoughtsStore.delete(oldestKey); +} + +export function saveRecentThought(sessionId, raw) { + if (!sessionId || raw == null || raw === "") return; + if (typeof raw === "string" && raw.length > MAX_RECENT_THOUGHT_CHARS) { + console.warn(`[continuity] skipping oversize reasoning checkpoint (${raw.length} chars)`); + return; + } + let entry = recentThoughtsStore.get(sessionId); + if (!entry) { + if (recentThoughtsStore.size >= 1000) evictLeastRecentlyUsedThoughtSession(); + entry = { thoughts: [], lastUsed: Date.now() }; + recentThoughtsStore.set(sessionId, entry); + } + entry.thoughts.push(raw); + if (entry.thoughts.length > 100) entry.thoughts.shift(); + entry.lastUsed = Date.now(); +} + +export function getRecentThoughts(sessionId, count = 100) { + const entry = recentThoughtsStore.get(sessionId); + if (!entry) return []; + entry.lastUsed = Date.now(); + return entry.thoughts.slice(-count); +} + +// Cleanup expired thought entries independently of runtimeSessionStore +const thoughtsCleanup = setInterval(() => { + const now = Date.now(); + for (const [sid, entry] of recentThoughtsStore) { + if (now - entry.lastUsed > MEMORY_CONFIG.sessionTtlMs) { + recentThoughtsStore.delete(sid); + } + } +}, MEMORY_CONFIG.sessionCleanupIntervalMs); +if (thoughtsCleanup.unref) thoughtsCleanup.unref(); diff --git a/open-sse/utils/stream.js b/open-sse/utils/stream.js index aea9614790e..baa760a5577 100644 --- a/open-sse/utils/stream.js +++ b/open-sse/utils/stream.js @@ -5,6 +5,7 @@ import { extractUsage, hasValidUsage, estimateUsage, logUsage, addBufferToUsage, import { parseSSELine, hasValuableContent, fixInvalidId, formatSSE } from "./streamHelpers.js"; import { getOpenAIResponsesEventName, isOpenAIResponsesTerminalEvent, formatIncompleteOpenAIResponsesStreamFailure } from "./responsesStreamHelpers.js"; import { dbg, isDebugEnabled } from "./debugLog.js"; +import { stripTaggedThinking } from "./taggedThinkingNormalizer.js"; import { SSE_DONE, SSE_HEADERS, SSE_HEADERS_NO_BUFFER } from "./sseConstants.js"; @@ -48,7 +49,8 @@ export function createSSEStream(options = {}) { connectionId = null, body = null, onStreamComplete = null, - apiKey = null + apiKey = null, + continuityEnabled = false } = options; let buffer = ""; @@ -130,20 +132,70 @@ export function createSSEStream(options = {}) { } } - if (!hasValuableContent(parsed, FORMATS.OPENAI)) { + if (!hasValuableContent(parsed)) { continue; } - const delta = parsed.choices?.[0]?.delta; - const content = delta?.content; - const reasoning = delta?.reasoning_content; - if (content && typeof content === "string") { - totalContentLength += content.length; - accumulatedContent += content; + let tagsStripped = false; + if (continuityEnabled) { + tagsStripped = stripTaggedThinking(parsed); } - if (reasoning && typeof reasoning === "string") { - totalContentLength += reasoning.length; - accumulatedThinking += reasoning; + + const unwrapped = parsed.response || parsed; + if (unwrapped.delta?.text) { + totalContentLength += unwrapped.delta.text.length; + accumulatedContent += unwrapped.delta.text; + } + if (unwrapped.delta?.thinking) { + totalContentLength += unwrapped.delta.thinking.length; + accumulatedThinking += unwrapped.delta.thinking; + } + if (typeof unwrapped.delta === "string" && unwrapped.type) { + totalContentLength += unwrapped.delta.length; + if (unwrapped.type.includes("reasoning")) { + accumulatedThinking += unwrapped.delta; + } else { + accumulatedContent += unwrapped.delta; + } + } + if (unwrapped.choices?.[0]?.delta?.content) { + totalContentLength += unwrapped.choices[0].delta.content.length; + accumulatedContent += unwrapped.choices[0].delta.content; + } + if (unwrapped.choices?.[0]?.delta?.reasoning_content) { + totalContentLength += unwrapped.choices[0].delta.reasoning_content.length; + accumulatedThinking += unwrapped.choices[0].delta.reasoning_content; + } + if (unwrapped.candidates?.[0]?.content?.parts) { + for (const part of unwrapped.candidates[0].content.parts) { + if (part.text && typeof part.text === "string") { + if (part.thought === true) { + totalContentLength += part.text.length; + accumulatedThinking += part.text; + } else { + totalContentLength += part.text.length; + accumulatedContent += part.text; + } + } + } + } + // Ollama format: { message: { content, thinking } } + if (unwrapped.message?.content) { + totalContentLength += unwrapped.message.content.length; + accumulatedContent += unwrapped.message.content; + } + if (unwrapped.message?.thinking) { + totalContentLength += unwrapped.message.thinking.length; + accumulatedThinking += unwrapped.message.thinking; + } + // CommandCode format (AI SDK v5): { type:"text-delta"|"reasoning-delta", text } + if (unwrapped.type === "text-delta" && typeof unwrapped.text === "string") { + totalContentLength += unwrapped.text.length; + accumulatedContent += unwrapped.text; + } + if (unwrapped.type === "reasoning-delta" && typeof unwrapped.text === "string") { + totalContentLength += unwrapped.text.length; + accumulatedThinking += unwrapped.text; } const extracted = extractUsage(parsed); @@ -151,20 +203,24 @@ export function createSSEStream(options = {}) { usage = extracted; } - const isFinishChunk = parsed.choices?.[0]?.finish_reason; + const isFinishChunk = unwrapped.choices?.[0]?.finish_reason; if (isFinishChunk && !hasValidUsage(parsed.usage)) { const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI); parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI); - output = `data: ${JSON.stringify(parsed)}\n`; + const nl = line.endsWith("\r") ? "\r\n" : "\n"; + output = `data: ${JSON.stringify(parsed)}${nl}`; usage = estimated; injectedUsage = true; } else if (isFinishChunk && usage) { const buffered = addBufferToUsage(usage); parsed.usage = filterUsageForFormat(buffered, FORMATS.OPENAI); - output = `data: ${JSON.stringify(parsed)}\n`; + const nl = line.endsWith("\r") ? "\r\n" : "\n"; + output = `data: ${JSON.stringify(parsed)}${nl}`; + usage = buffered; injectedUsage = true; - } else if (idFixed || fieldsInjected) { - output = `data: ${JSON.stringify(parsed)}\n`; + } else if (idFixed || fieldsInjected || tagsStripped) { + const nl = line.endsWith("\r") ? "\r\n" : "\n"; + output = `data: ${JSON.stringify(parsed)}${nl}`; injectedUsage = true; } } catch { @@ -194,6 +250,10 @@ export function createSSEStream(options = {}) { const parsed = parseSSELine(trimmed, targetFormat); if (!parsed) continue; + if (continuityEnabled) { + stripTaggedThinking(parsed); + } + // Responses API same-format passthrough: preserve event framing + track terminal state const isOpenAIResponsesStream = targetFormat === FORMATS.OPENAI_RESPONSES; const keepsOpenAIResponsesFormat = isOpenAIResponsesStream && sourceFormat === FORMATS.OPENAI_RESPONSES; @@ -227,42 +287,71 @@ export function createSSEStream(options = {}) { continue; } + const unwrapped = parsed.response || parsed; + // Claude format - content - if (parsed.delta?.text) { - totalContentLength += parsed.delta.text.length; - accumulatedContent += parsed.delta.text; + if (unwrapped.delta?.text) { + totalContentLength += unwrapped.delta.text.length; + accumulatedContent += unwrapped.delta.text; } // Claude format - thinking - if (parsed.delta?.thinking) { - totalContentLength += parsed.delta.thinking.length; - accumulatedThinking += parsed.delta.thinking; + if (unwrapped.delta?.thinking) { + totalContentLength += unwrapped.delta.thinking.length; + accumulatedThinking += unwrapped.delta.thinking; + } + // Responses API format - delta is a top-level string + if (typeof unwrapped.delta === "string" && unwrapped.type) { + totalContentLength += unwrapped.delta.length; + if (unwrapped.type.includes("reasoning")) { + accumulatedThinking += unwrapped.delta; + } else { + accumulatedContent += unwrapped.delta; + } } // OpenAI format - content - if (parsed.choices?.[0]?.delta?.content) { - totalContentLength += parsed.choices[0].delta.content.length; - accumulatedContent += parsed.choices[0].delta.content; + if (unwrapped.choices?.[0]?.delta?.content) { + totalContentLength += unwrapped.choices[0].delta.content.length; + accumulatedContent += unwrapped.choices[0].delta.content; } // OpenAI format - reasoning - if (parsed.choices?.[0]?.delta?.reasoning_content) { - totalContentLength += parsed.choices[0].delta.reasoning_content.length; - accumulatedThinking += parsed.choices[0].delta.reasoning_content; + if (unwrapped.choices?.[0]?.delta?.reasoning_content) { + totalContentLength += unwrapped.choices[0].delta.reasoning_content.length; + accumulatedThinking += unwrapped.choices[0].delta.reasoning_content; } // Gemini format - if (parsed.candidates?.[0]?.content?.parts) { - for (const part of parsed.candidates[0].content.parts) { + if (unwrapped.candidates?.[0]?.content?.parts) { + for (const part of unwrapped.candidates[0].content.parts) { if (part.text && typeof part.text === "string") { - totalContentLength += part.text.length; - // Check if this is thinking content if (part.thought === true) { + totalContentLength += part.text.length; accumulatedThinking += part.text; } else { + totalContentLength += part.text.length; accumulatedContent += part.text; } } } } + // Ollama format: { message: { content, thinking } } + if (unwrapped.message?.content) { + totalContentLength += unwrapped.message.content.length; + accumulatedContent += unwrapped.message.content; + } + if (unwrapped.message?.thinking) { + totalContentLength += unwrapped.message.thinking.length; + accumulatedThinking += unwrapped.message.thinking; + } + // CommandCode format (AI SDK v5): { type:"text-delta"|"reasoning-delta", text } + if (unwrapped.type === "text-delta" && typeof unwrapped.text === "string") { + totalContentLength += unwrapped.text.length; + accumulatedContent += unwrapped.text; + } + if (unwrapped.type === "reasoning-delta" && typeof unwrapped.text === "string") { + totalContentLength += unwrapped.text.length; + accumulatedThinking += unwrapped.text; + } // Extract usage const extracted = extractUsage(parsed); @@ -450,7 +539,7 @@ export function createSSEStream(options = {}) { }); } -export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null, onStreamComplete = null, apiKey = null) { +export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null, onStreamComplete = null, apiKey = null, continuityEnabled = false) { return createSSEStream({ mode: STREAM_MODE.TRANSLATE, targetFormat, @@ -462,11 +551,12 @@ export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, p connectionId, body, onStreamComplete, - apiKey + apiKey, + continuityEnabled }); } -export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null, onStreamComplete = null, apiKey = null) { +export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null, onStreamComplete = null, apiKey = null, continuityEnabled = false) { return createSSEStream({ mode: STREAM_MODE.PASSTHROUGH, provider, @@ -475,6 +565,7 @@ export function createPassthroughStreamWithLogger(provider = null, reqLogger = n connectionId, body, onStreamComplete, - apiKey + apiKey, + continuityEnabled }); } diff --git a/open-sse/utils/streamHelpers.js b/open-sse/utils/streamHelpers.js index a7a19180fed..3d8b04d629e 100644 --- a/open-sse/utils/streamHelpers.js +++ b/open-sse/utils/streamHelpers.js @@ -34,23 +34,26 @@ export function parseSSELine(line, format = null) { } // Check if chunk has valuable content (not empty) -export function hasValuableContent(chunk, format) { +export function hasValuableContent(chunk, format = null) { + const unwrapped = chunk.response || chunk; + const fmt = format || (unwrapped.candidates !== undefined ? FORMATS.GEMINI : (unwrapped.choices === undefined && (unwrapped.type !== undefined || unwrapped.delta !== undefined) ? FORMATS.CLAUDE : FORMATS.OPENAI)); + // OpenAI format - if (format === FORMATS.OPENAI && chunk.choices?.[0]?.delta) { - const delta = chunk.choices[0].delta; + if (fmt === FORMATS.OPENAI && unwrapped.choices?.[0]?.delta) { + const delta = unwrapped.choices[0].delta; return delta.content && delta.content !== "" || delta.reasoning_content && delta.reasoning_content !== "" || delta.tool_calls && delta.tool_calls.length > 0 || - chunk.choices[0].finish_reason || + unwrapped.choices[0].finish_reason || delta.role; } // Claude format - if (format === FORMATS.CLAUDE) { - const isContentBlockDelta = chunk.type === "content_block_delta"; - const hasText = chunk.delta?.text && chunk.delta.text !== ""; - const hasThinking = chunk.delta?.thinking && chunk.delta.thinking !== ""; - const hasInputJson = chunk.delta?.partial_json && chunk.delta.partial_json !== ""; + if (fmt === FORMATS.CLAUDE) { + const isContentBlockDelta = unwrapped.type === "content_block_delta"; + const hasText = unwrapped.delta?.text && unwrapped.delta.text !== ""; + const hasThinking = unwrapped.delta?.thinking && unwrapped.delta.thinking !== ""; + const hasInputJson = unwrapped.delta?.partial_json && unwrapped.delta.partial_json !== ""; if (isContentBlockDelta && !hasText && !hasThinking && !hasInputJson) { return false; diff --git a/open-sse/utils/taggedThinkingNormalizer.js b/open-sse/utils/taggedThinkingNormalizer.js new file mode 100644 index 00000000000..c15abb3fce0 --- /dev/null +++ b/open-sse/utils/taggedThinkingNormalizer.js @@ -0,0 +1,45 @@ +// continuity: simple tag stripper — removes //// +// tag markers from content. Inner text stays as visible content. No migration to +// thinking fields. Ceiling: streaming chunk-boundary tag splits may leak; upgrade to +// stateful buffering only if observed in practice. +const TAG_RE = /<\/?(?:think|thinking|thought|reasoning|analysis)>/gi; + +// Returns null when no change, or the stripped string when tags were removed. +export function stripTags(text) { + if (typeof text !== "string") return null; + const result = text.replace(TAG_RE, ""); + return result === text ? null : result; +} + +export function stripTaggedThinking(obj) { + if (!obj || typeof obj !== "object") return false; + let changed = false; + + if (Array.isArray(obj)) { + for (const item of obj) { + if (stripTaggedThinking(item)) changed = true; + } + return changed; + } + + if (typeof obj.content === "string") { + const stripped = stripTags(obj.content); + if (stripped !== null) { obj.content = stripped; changed = true; } + } + if (typeof obj.text === "string") { + const stripped = stripTags(obj.text); + if (stripped !== null) { obj.text = stripped; changed = true; } + } + if (typeof obj.delta === "string") { + const stripped = stripTags(obj.delta); + if (stripped !== null) { obj.delta = stripped; changed = true; } + } + + for (const key of ["response", "choices", "message", "delta", "output", "content", "candidates", "parts"]) { + if (obj[key]) { + if (stripTaggedThinking(obj[key])) changed = true; + } + } + + return changed; +} diff --git a/open-sse/utils/thinkingExtractor.js b/open-sse/utils/thinkingExtractor.js new file mode 100644 index 00000000000..beb105463bd --- /dev/null +++ b/open-sse/utils/thinkingExtractor.js @@ -0,0 +1,35 @@ +export function extractThinking(value) { + if (!value) return null; + const out = []; + collectThinking(value, out); + return [...new Set(out.filter(Boolean))].join("\n") || null; +} + +function collectThinking(obj, out) { + if (!obj || typeof obj !== "object") return; + + if (Array.isArray(obj)) { + for (const item of obj) collectThinking(item, out); + return; + } + + if (typeof obj.reasoning_content === "string") out.push(obj.reasoning_content); + if (typeof obj.thinking === "string") out.push(obj.thinking); + if (obj.thought === true && typeof obj.text === "string") out.push(obj.text); + + if (obj.type === "thinking" && typeof obj.thinking === "string") out.push(obj.thinking); + if (obj.type === "reasoning") { + if (typeof obj.text === "string") out.push(obj.text); + if (typeof obj.content === "string") out.push(obj.content); + if (typeof obj.summary === "string") out.push(obj.summary); + } + + if (obj.response) collectThinking(obj.response, out); + if (obj.choices) collectThinking(obj.choices, out); + if (obj.message) collectThinking(obj.message, out); + if (obj.delta) collectThinking(obj.delta, out); + if (obj.output) collectThinking(obj.output, out); + if (obj.content) collectThinking(obj.content, out); + if (obj.candidates) collectThinking(obj.candidates, out); + if (obj.parts) collectThinking(obj.parts, out); +} diff --git a/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js b/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js index 637ddced85e..dbf9b9af2a4 100644 --- a/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js +++ b/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js @@ -28,6 +28,8 @@ export default function TokenSaverClient() { const [cavemanLevel, setCavemanLevel] = useState("full"); const [ponytailEnabled, setPonytailEnabled] = useState(false); const [ponytailLevel, setPonytailLevel] = useState("full"); + const [continuityEnabled, setContinuityEnabled] = useState(false); + const [continuityCount, setContinuityCount] = useState(3); const [locale, setLocale] = useState("en"); const { copied, copy } = useCopyToClipboard(); @@ -152,6 +154,20 @@ export default function TokenSaverClient() { patchSetting({ ponytailLevel: level }); }; + const handleContinuityEnabled = (value) => { + setContinuityEnabled(value); + patchSetting({ continuityEnabled: value }); + }; + + const handleContinuityCount = (e) => { + let val = parseInt(e.target.value, 10); + if (isNaN(val)) val = 1; + if (val < 1) val = 1; + if (val > 100) val = 100; + setContinuityCount(val); + patchSetting({ continuityCount: val }); + }; + useEffect(() => { const loadSettings = async () => { try { @@ -165,6 +181,8 @@ export default function TokenSaverClient() { setCavemanLevel(data.cavemanLevel || "full"); setPonytailEnabled(!!data.ponytailEnabled); setPonytailLevel(data.ponytailLevel || "full"); + setContinuityEnabled(!!data.continuityEnabled); + setContinuityCount(parseInt(data.continuityCount) || 3); refreshHeadroomStatus(); } } catch {} @@ -358,6 +376,45 @@ export default function TokenSaverClient() { /> + + {/* Continuity Feature (experimental, opt-in) */} +
+
+

+ Continuity{" "} + (Experimental) +

+

+ Replay recent reasoning checkpoints into the system prompt for cross-model continuation +

+
+
+ {continuityEnabled && ( +
+
+
+ Count: + +
+
+

+ Replaying up to {continuityCount} reasoning checkpoints into the system prompt. +

+
+ )} + handleContinuityEnabled(!continuityEnabled)} + /> +
+