diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 3e0c90fe69..bf6aa9c6a8 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -305,7 +305,12 @@ import { resolveComboContextLimit, } from "../services/contextManager.ts"; import { resolveBackgroundTaskRedirect } from "./chatCore/backgroundRedirect.ts"; -import type { CompressionConfig, CompressionPipelineStep } from "../services/compression/types.ts"; +import type { + CompressionConfig, + CompressionPipelineStep, + CompressionResult, +} from "../services/compression/types.ts"; +import { generateSessionId } from "../services/sessionManager.ts"; import { prepareWebSearchFallbackBody } from "../services/webSearchFallback.ts"; import { resolveInterceptSearch } from "@/lib/db/interceptionRules"; import { @@ -1338,7 +1343,8 @@ export async function handleChatCore({ // that selectCompressionStrategy can only partially apply via the mode string. const cacheCtx = { provider, targetFormat, model: effectiveModel, connectionCacheOverride }; const compressionConfig = resolveCacheAwareConfig(config, compressionInputBody, cacheCtx); - const result = await applyCompressionAsync(compressionInputBody, mode, { + const compressionPrincipalId = apiKeyInfo?.id ? String(apiKeyInfo.id) : undefined; + const compressionOptions = { model: effectiveModel, // #7237: feed the AUTHORITATIVE capability (model spec / models.dev sync / DB // override, with the conservative model-id fragment heuristic only as its @@ -1352,10 +1358,11 @@ export async function handleChatCore({ .supportsVision, // Rota direta oficial ('anthropic') vs agregadores: o engine omniglyph // exige 'direct' — agregadores redimensionam imagens (medido 2026-07-06). - providerTransport: provider === "anthropic" ? "direct" : "aggregator", + providerTransport: + provider === "anthropic" ? ("direct" as const) : ("aggregator" as const), config: compressionConfig, cachingContext: cacheCtx, - principalId: apiKeyInfo?.id ? String(apiKeyInfo.id) : undefined, + principalId: compressionPrincipalId, // F3.3: stream per-engine progress live (best-effort) before compression.completed. onEngineStep: (s) => { try { @@ -1379,7 +1386,52 @@ export async function handleChatCore({ // best-effort live event — never fail the request } }, - }); + }; + const runCompression = (input: Record) => + applyCompressionAsync(input, mode, compressionOptions); + let result: CompressionResult; + if (compressionConfig.liveZone?.enabled === true) { + const { applyLiveZoneCompression } = await import("../services/compression/liveZone.ts"); + const explicitSessionId = + clientRawRequest?.headers && typeof clientRawRequest.headers.get === "function" + ? clientRawRequest.headers.get("x-omniroute-session-id") + : getHeaderValueCaseInsensitive( + clientRawRequest?.headers ?? null, + "x-omniroute-session-id" + ); + const liveZoneSessionId = + explicitSessionId || + generateSessionId(compressionInputBody, { + provider, + connectionId: getCurrentConnectionId() ?? undefined, + }) || + undefined; + result = await applyLiveZoneCompression( + compressionInputBody, + { + principalId: compressionPrincipalId, + sessionId: liveZoneSessionId, + variant: { + mode, + provider, + model: effectiveModel, + config: compressionConfig, + cachePrefix: { + system: compressionInputBody.system, + systemInstruction: compressionInputBody.systemInstruction, + system_instruction: compressionInputBody.system_instruction, + instructions: compressionInputBody.instructions, + tools: compressionInputBody.tools, + toolChoice: compressionInputBody.tool_choice, + }, + }, + ttlMinutes: compressionConfig.cacheMinutes, + }, + runCompression + ); + } else { + result = await runCompression(compressionInputBody); + } if (result.stats) { const annotation = formatCompressionAnnotation(result.stats); if (annotation) { diff --git a/open-sse/services/compression/liveZone.ts b/open-sse/services/compression/liveZone.ts new file mode 100644 index 0000000000..d0c69cd03a --- /dev/null +++ b/open-sse/services/compression/liveZone.ts @@ -0,0 +1,397 @@ +import { createHash } from "node:crypto"; + +import { estimateCompressionTokens } from "./stats.ts"; +import type { CompressionResult, CompressionStats } from "./types.ts"; + +export interface LiveZoneOptions { + principalId?: string; + sessionId?: string; + variant: unknown; + ttlMinutes?: number; +} + +interface LiveZoneEntry { + rawItemDigests: string[]; + rawStableFieldsDigest: string; + transformedPrefix: unknown[]; + transformedStableFields: Record; + stats: CompressionStats | null; + lastAccess: number; + expiresAt: number; + bytes: number; +} + +interface LiveZoneContext { + field: "messages" | "input"; + key: string; + rawItems: unknown[]; + rawItemDigests: string[]; + rawStableFieldsDigest: string; + ttlMs: number; + now: number; +} + +const MAX_ENTRIES = 100; +const MAX_ENTRY_BYTES = 2 * 1024 * 1024; +const MAX_TOTAL_BYTES = 32 * 1024 * 1024; +const DEFAULT_TTL_MINUTES = 5; +const STABLE_PREFIX_FIELDS = [ + "system", + "systemInstruction", + "system_instruction", + "instructions", + "tools", + "tool_choice", +] as const; + +const entries = new Map(); +let totalBytes = 0; + +function serialize(value: unknown): string | null { + try { + const serialized = JSON.stringify(value); + return typeof serialized === "string" ? serialized : null; + } catch { + return null; + } +} + +function digest(value: unknown): string | null { + const serialized = serialize(value); + return serialized === null ? null : createHash("sha256").update(serialized).digest("hex"); +} + +function cloneItems(items: unknown[]): unknown[] | null { + try { + return structuredClone(items); + } catch { + const serialized = serialize(items); + if (serialized === null) return null; + try { + return JSON.parse(serialized) as unknown[]; + } catch { + return null; + } + } +} + +function cloneValue(value: T): T | null { + try { + return structuredClone(value); + } catch { + const serialized = serialize(value); + if (serialized === null) return null; + try { + return JSON.parse(serialized) as T; + } catch { + return null; + } + } +} + +function pickStableFields(body: Record): Record | null { + const fields: Record = {}; + for (const field of STABLE_PREFIX_FIELDS) { + if (Object.prototype.hasOwnProperty.call(body, field)) fields[field] = body[field]; + } + return cloneValue(fields); +} + +function sequenceField(body: Record): "messages" | "input" | null { + if (Array.isArray(body.messages)) return "messages"; + if (Array.isArray(body.input)) return "input"; + return null; +} + +function isToolOutputItem(value: unknown): boolean { + if (!value || typeof value !== "object") return false; + const item = value as Record; + return ( + item.role === "tool" || + item.role === "function" || + item.role === "tool_result" || + item.type === "function_call_output" || + item.type === "computer_call_output" || + item.type === "tool_result" + ); +} + +function makeKey(options: LiveZoneOptions, field: string): string | null { + const principal = options.principalId?.trim(); + const session = options.sessionId?.trim(); + const variant = digest(options.variant); + if (!principal || !session || !variant) return null; + return `${principal}:${session}:${field}:${variant}`; +} + +function deleteEntry(key: string): void { + const existing = entries.get(key); + if (!existing) return; + totalBytes -= existing.bytes; + entries.delete(key); +} + +function prune(now: number): void { + for (const [key, entry] of entries) { + if (now >= entry.expiresAt) deleteEntry(key); + } + while (entries.size > MAX_ENTRIES || totalBytes > MAX_TOTAL_BYTES) { + const oldest = entries.keys().next().value as string | undefined; + if (!oldest) break; + deleteEntry(oldest); + } +} + +function store( + key: string, + rawItemDigests: string[], + rawStableFieldsDigest: string, + result: CompressionResult, + field: "messages" | "input", + now: number, + ttlMs: number +): void { + const transformedItems = result.body[field]; + if (!Array.isArray(transformedItems)) return; + const transformedPrefix = cloneItems(transformedItems); + const transformedStableFields = pickStableFields(result.body); + const stats = cloneValue(result.stats); + if (!transformedPrefix || !transformedStableFields) return; + const serialized = serialize({ transformedPrefix, transformedStableFields, stats }); + if (serialized === null) return; + const bytes = Buffer.byteLength(serialized, "utf8") + rawItemDigests.length * 64; + if (bytes > MAX_ENTRY_BYTES) return; + + deleteEntry(key); + entries.set(key, { + rawItemDigests, + rawStableFieldsDigest, + transformedPrefix, + transformedStableFields, + stats, + lastAccess: now, + expiresAt: now + ttlMs, + bytes, + }); + totalBytes += bytes; + prune(now); +} + +function hasExactRawPrefix(rawItemDigests: string[], entry: LiveZoneEntry): boolean { + if (rawItemDigests.length < entry.rawItemDigests.length) return false; + for (let index = 0; index < entry.rawItemDigests.length; index++) { + if (rawItemDigests[index] !== entry.rawItemDigests[index]) return false; + } + return true; +} + +function restoreStableFields( + body: Record, + stableFields: Record +): Record | null { + const restored = cloneValue(stableFields); + return restored ? { ...body, ...restored } : null; +} + +function withLiveZoneStats( + body: Record, + result: CompressionResult, + frozenItems: number, + liveItems: number +): CompressionResult { + const originalTokens = estimateCompressionTokens(body); + const compressedTokens = estimateCompressionTokens(result.body); + const savingsPercent = + originalTokens > 0 + ? Math.max( + 0, + Math.round(((originalTokens - compressedTokens) / originalTokens) * 10000) / 100 + ) + : 0; + const base = result.stats; + const stats: CompressionStats = { + ...(base ?? { + techniquesUsed: [], + mode: "stacked", + timestamp: Date.now(), + }), + originalTokens, + compressedTokens, + savingsPercent, + techniquesUsed: [...new Set([...(base?.techniquesUsed ?? []), "live-zone-prefix-reuse"])], + liveZone: { + cacheHit: true, + frozenItems, + liveItems, + }, + }; + return { + ...result, + compressed: result.compressed || compressedTokens < originalTokens, + stats, + }; +} + +function hasGlobalHardBudget(variant: unknown): boolean { + if (!variant || typeof variant !== "object") return false; + const config = (variant as Record).config; + if (!config || typeof config !== "object") return false; + const record = config as Record; + return record.targetTokens != null || record.targetRatio != null; +} + +function resolveLiveZoneContext( + body: Record, + options: LiveZoneOptions +): LiveZoneContext | null { + const field = sequenceField(body); + const key = field ? makeKey(options, field) : null; + if (!field || !key) return null; + + const rawItems = body[field] as unknown[]; + const rawItemDigests = rawItems.map(digest); + if (rawItemDigests.some((value) => value === null)) return null; + const rawStableFieldsDigest = digest(pickStableFields(body)); + if (!rawStableFieldsDigest) return null; + const ttlMinutes = Math.min(60, Math.max(1, options.ttlMinutes ?? DEFAULT_TTL_MINUTES)); + const now = Date.now(); + return { + field, + key, + rawItems, + rawItemDigests: rawItemDigests as string[], + rawStableFieldsDigest, + ttlMs: ttlMinutes * 60_000, + now, + }; +} + +async function compressAndStore( + body: Record, + context: LiveZoneContext, + compress: (body: Record) => Promise +): Promise { + const result = await compress(body); + store( + context.key, + context.rawItemDigests, + context.rawStableFieldsDigest, + result, + context.field, + context.now, + context.ttlMs + ); + return result; +} + +async function compressLiveToolOutputs( + body: Record, + field: "messages" | "input", + liveItems: unknown[], + previousStats: CompressionStats | null, + compress: (body: Record) => Promise +): Promise<{ liveResult: CompressionResult; transformedLive: unknown[] } | null> { + const transformedLive = cloneItems(liveItems); + if (!transformedLive) return null; + const liveToolIndexes = liveItems.flatMap((item, index) => + isToolOutputItem(item) ? [index] : [] + ); + if (liveToolIndexes.length === 0) { + return { liveResult: { body, compressed: false, stats: previousStats }, transformedLive }; + } + + const liveToolItems = liveToolIndexes.map((index) => liveItems[index]); + const liveResult = await compress({ ...body, [field]: liveToolItems }); + const transformed = liveResult.body[field]; + if (!Array.isArray(transformed) || transformed.length !== liveToolItems.length) { + return { liveResult: { body, compressed: false, stats: null }, transformedLive }; + } + for (let index = 0; index < liveToolIndexes.length; index++) { + transformedLive[liveToolIndexes[index]] = transformed[index]; + } + return { liveResult, transformedLive }; +} + +async function reuseLiveZoneEntry( + body: Record, + context: LiveZoneContext, + previous: LiveZoneEntry, + compress: (body: Record) => Promise +): Promise { + entries.delete(context.key); + previous.lastAccess = context.now; + entries.set(context.key, previous); + + const frozenItems = previous.rawItemDigests.length; + const liveItems = context.rawItems.slice(frozenItems); + const frozenPrefix = cloneItems(previous.transformedPrefix); + if (!frozenPrefix) return compress(body); + const live = await compressLiveToolOutputs( + body, + context.field, + liveItems, + previous.stats, + compress + ); + if (!live) return compress(body); + const restoredBody = restoreStableFields(live.liveResult.body, previous.transformedStableFields); + if (!restoredBody) return compress(body); + const combinedBody = { + ...restoredBody, + [context.field]: [...frozenPrefix, ...live.transformedLive], + }; + const combinedResult = withLiveZoneStats( + body, + { ...live.liveResult, body: combinedBody }, + frozenItems, + liveItems.length + ); + if (entries.get(context.key) === previous) { + store( + context.key, + context.rawItemDigests, + context.rawStableFieldsDigest, + combinedResult, + context.field, + Date.now(), + context.ttlMs + ); + } + return combinedResult; +} + +/** + * Reuses the byte-identical transformed prefix from the previous request in a session and runs + * compression only over newly appended messages/input items. Any changed prefix, missing identity, + * unsupported body shape, serialization failure, or oversized entry fails open to full compression. + */ +export async function applyLiveZoneCompression( + body: Record, + options: LiveZoneOptions, + compress: (body: Record) => Promise +): Promise { + // A global hard budget needs the complete history to make correct keep/drop decisions. + if (hasGlobalHardBudget(options.variant)) return compress(body); + const context = resolveLiveZoneContext(body, options); + if (!context) return compress(body); + prune(context.now); + const previous = entries.get(context.key); + + if ( + !previous || + previous.rawStableFieldsDigest !== context.rawStableFieldsDigest || + !hasExactRawPrefix(context.rawItemDigests, previous) + ) { + return compressAndStore(body, context, compress); + } + return reuseLiveZoneEntry(body, context, previous, compress); +} + +export function resetLiveZoneCache(): void { + entries.clear(); + totalBytes = 0; +} + +export function getLiveZoneCacheStats(): { entries: number; bytes: number } { + return { entries: entries.size, bytes: totalBytes }; +} diff --git a/open-sse/services/compression/types.ts b/open-sse/services/compression/types.ts index 9bba7973b2..ff8aedd77b 100644 --- a/open-sse/services/compression/types.ts +++ b/open-sse/services/compression/types.ts @@ -131,6 +131,11 @@ export interface ContextEditingConfig { enabled: boolean; } +/** Cache-aligned compression: freeze a previously transformed prefix and process only new items. */ +export interface LiveZoneConfig { + enabled: boolean; +} + export interface CompressionPipelineStep { engine: CompressionEngineId; intensity?: CavemanIntensity | RtkIntensity; @@ -193,6 +198,8 @@ export interface CompressionConfig { ultra?: UltraConfig; /** Provider-delegated context editing (Claude/Anthropic only). */ contextEditing?: ContextEditingConfig; + /** Opt-in cache-aligned live-zone compression (default disabled). */ + liveZone?: LiveZoneConfig; /** Per-engine opt-in toggles for the config panel. */ engines: Record; /** Active combo preset id, or null if none selected. */ @@ -294,6 +301,11 @@ export interface CompressionStats { }>; /** Present only when QuantumLock stabilized ≥1 fragment this run. */ quantumLock?: QuantumLockStats; + liveZone?: { + cacheHit: boolean; + frozenItems: number; + liveItems: number; + }; } export interface CompressionResult { @@ -321,6 +333,7 @@ export const DEFAULT_COMPRESSION_CONFIG: CompressionConfig = { activeComboId: null, ultraEngine: "heuristic", ultraSlmPrewarm: false, + liveZone: { enabled: false }, }; export const DEFAULT_CAVEMAN_CONFIG: CavemanConfig = { diff --git a/src/app/(dashboard)/dashboard/context/settings/CompressionPanel.tsx b/src/app/(dashboard)/dashboard/context/settings/CompressionPanel.tsx index 3e416834c9..3d9bbd9362 100644 --- a/src/app/(dashboard)/dashboard/context/settings/CompressionPanel.tsx +++ b/src/app/(dashboard)/dashboard/context/settings/CompressionPanel.tsx @@ -67,6 +67,7 @@ interface CompressionConfig { // The panel currently surfaces the computed target read-only; mode/policy editors are a // follow-up (the load/save path does not yet populate this field). contextBudget?: ContextBudgetConfig; + liveZone?: { enabled: boolean }; } const CAVEMAN_OUTPUT_LEVELS: CavemanIntensity[] = ["lite", "full", "ultra"]; @@ -81,6 +82,7 @@ const DEFAULT_CONFIG: CompressionConfig = { outputStyles: [], ultraEngine: "heuristic", ultraSlmPrewarm: false, + liveZone: { enabled: false }, }; function normalizeEngines(raw: unknown): Record { @@ -88,11 +90,40 @@ function normalizeEngines(raw: unknown): Record { const source = (raw && typeof raw === "object" ? raw : {}) as Record; for (const id of ENGINE_IDS) { const cur = source[id]; - engines[id] = cur ? { enabled: cur.enabled === true, ...(cur.level ? { level: cur.level } : {}) } : { enabled: false }; + engines[id] = cur + ? { enabled: cur.enabled === true, ...(cur.level ? { level: cur.level } : {}) } + : { enabled: false }; } return engines; } +function LiveZoneToggle({ + enabled, + saving, + onChange, +}: { + enabled: boolean; + saving: boolean; + onChange: (enabled: boolean) => void; +}) { + const t = useTranslations("settings"); + return ( + + ); +} + export default function CompressionPanel() { const t = useTranslations("settings"); // D-A6/§7: locale-gated styles (e.g. terse-cjk → zh) are only OFFERED under their locale. @@ -256,8 +287,7 @@ export default function CompressionPanel() { )} {status === "error" && ( - error{" "} - {t("saveFailed")} + error {t("saveFailed")} )}

{meta.label}

- {meta.description && ( -

{meta.description}

- )} + {meta.description &&

{meta.description}

}
- save({ ultraEngine: e.target.value === "slm" ? "slm" : "heuristic" }) - } + onChange={(e) => save({ ultraEngine: e.target.value === "slm" ? "slm" : "heuristic" })} disabled={saving} className="w-44 rounded border border-border bg-surface px-2 py-1 text-sm text-text-main" > @@ -431,9 +455,7 @@ export default function CompressionPanel() { <>

{t("compressionUltraSlmHint")}

+ save({ liveZone: { enabled } })} + />
); diff --git a/src/i18n/messages/de.json b/src/i18n/messages/de.json index 0c0475c7e1..080af3f6d2 100644 --- a/src/i18n/messages/de.json +++ b/src/i18n/messages/de.json @@ -5163,6 +5163,8 @@ "compressionAutoTrigger": "Auto-Trigger Threshold", "compressionCacheTTL": "Cache TTL", "compressionPreserveSystem": "Preserve System Prompt", + "compressionLiveZoneTitle": "Cache-ausgerichtete Live Zone", + "compressionLiveZoneDesc": "Hält den komprimierten Gesprächspräfix stabil und verarbeitet nur neu angehängte Einträge.", "compressionCavemanConfig": "Caveman Engine Configuration", "compressionCavemanConfigDesc": "Fine-tune the rule-based compression engine", "compressionRoles": "Compress Message Roles", diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index 45fcb0bd84..594ec26981 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -5804,6 +5804,8 @@ "compressionPreserveSystemAlways": "Always", "compressionPreserveSystemWhenNoCache": "When no cache", "compressionPreserveSystemNever": "Never", + "compressionLiveZoneTitle": "Cache-aligned Live Zone", + "compressionLiveZoneDesc": "Keep the compressed conversation prefix stable and process only newly appended items.", "compressionCavemanConfig": "Caveman Engine Configuration", "compressionCavemanConfigDesc": "Fine-tune the rule-based compression engine", "compressionCavemanPanelHint": "Its on/off and level are set in the panel:", diff --git a/src/i18n/messages/pt-BR.json b/src/i18n/messages/pt-BR.json index 4d957b14b1..b19a5cf870 100644 --- a/src/i18n/messages/pt-BR.json +++ b/src/i18n/messages/pt-BR.json @@ -5766,6 +5766,8 @@ "compressionPreserveSystemAlways": "Sempre", "compressionPreserveSystemWhenNoCache": "Quando não há cache", "compressionPreserveSystemNever": "Nunca", + "compressionLiveZoneTitle": "Zona ativa alinhada ao cache", + "compressionLiveZoneDesc": "Mantém estável o prefixo comprimido da conversa e processa apenas os itens recém-adicionados.", "compressionCavemanConfig": "Caveman Engine Configuration", "compressionCavemanConfigDesc": "Fine-tune the rule-based compression engine", "compressionCavemanPanelHint": "Ativado/desativado e o nível são definidos no painel:", diff --git a/src/lib/db/compression.ts b/src/lib/db/compression.ts index 6bb7e0fd1b..f015d3687c 100644 --- a/src/lib/db/compression.ts +++ b/src/lib/db/compression.ts @@ -554,6 +554,7 @@ export async function getCompressionSettings(): Promise { ultra: normalizeUltraConfig(undefined), contextBudget: normalizeContextBudgetConfig(undefined), contextEditing: { ...DEFAULT_CONTEXT_EDITING_CONFIG }, + liveZone: { enabled: false }, engines: {}, activeComboId: null, }; @@ -661,6 +662,9 @@ export async function getCompressionSettings(): Promise { case "contextEditing": config.contextEditing = normalizeContextEditingConfig(parsed); break; + case "liveZone": + config.liveZone = { enabled: toRecord(parsed).enabled === true }; + break; case "engines": storedEngines = parseStoredEnginesMap(parsed); break; diff --git a/src/shared/validation/compressionConfigSchemas.ts b/src/shared/validation/compressionConfigSchemas.ts index b640cf99ff..745d284043 100644 --- a/src/shared/validation/compressionConfigSchemas.ts +++ b/src/shared/validation/compressionConfigSchemas.ts @@ -314,6 +314,7 @@ export const compressionSettingsUpdateSchema = z ultra: ultraConfigSchema.optional(), contextBudget: contextBudgetConfigSchema.optional(), contextEditing: contextEditingConfigSchema.optional(), + liveZone: z.object({ enabled: z.boolean() }).strict().optional(), engines: z.record(z.string(), engineToggleSchema).optional(), enginesExplicit: z.boolean().optional(), activeComboId: z.string().nullable().optional(), diff --git a/tests/unit/compression/compression-preview-api.test.ts b/tests/unit/compression/compression-preview-api.test.ts index 9a34952d15..c751bafa7c 100644 --- a/tests/unit/compression/compression-preview-api.test.ts +++ b/tests/unit/compression/compression-preview-api.test.ts @@ -8,6 +8,10 @@ import { describe("compression preview API contract", () => { it("accepts RTK and stacked preview payloads", () => { + assert.equal( + PreviewCompressionConfigSchema.safeParse({ liveZone: { enabled: true } }).success, + true + ); assert.equal( PreviewRequestSchema.safeParse({ messages: [{ role: "tool", content: "same\nsame\nsame" }], diff --git a/tests/unit/compression/db.test.ts b/tests/unit/compression/db.test.ts index 2df3895713..20ba863dbd 100644 --- a/tests/unit/compression/db.test.ts +++ b/tests/unit/compression/db.test.ts @@ -52,6 +52,7 @@ describe("getCompressionSettings", () => { assert.equal(settings.cacheMinutes, 5); assert.equal(settings.preserveSystemPrompt, true); assert.equal(settings.preserveSystemPromptMode, "always"); + assert.deepEqual(settings.liveZone, { enabled: false }); assert.deepEqual(settings.comboOverrides, {}); assert.equal(settings.ultra?.enabled, false); assert.equal(settings.ultra?.compressionRate, 0.5); @@ -105,6 +106,16 @@ describe("updateCompressionSettings", () => { await updateCompressionSettings({ autoTriggerTokens: 0 } as any); }); + it("round-trips cache-aligned live-zone compression", async () => { + await updateCompressionSettings({ liveZone: { enabled: true } }); + let settings = await getCompressionSettings(); + assert.deepEqual(settings.liveZone, { enabled: true }); + + await updateCompressionSettings({ liveZone: { enabled: false } }); + settings = await getCompressionSettings(); + assert.deepEqual(settings.liveZone, { enabled: false }); + }); + it("updates multiple settings at once", async () => { await updateCompressionSettings({ enabled: true, diff --git a/tests/unit/compression/live-zone.test.ts b/tests/unit/compression/live-zone.test.ts new file mode 100644 index 0000000000..fbddf3e761 --- /dev/null +++ b/tests/unit/compression/live-zone.test.ts @@ -0,0 +1,254 @@ +import { beforeEach, describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + applyLiveZoneCompression, + getLiveZoneCacheStats, + resetLiveZoneCache, +} from "../../../open-sse/services/compression/liveZone.ts"; +import type { CompressionResult } from "../../../open-sse/services/compression/types.ts"; + +function result(body: Record, compressed = true): CompressionResult { + return { + body, + compressed, + stats: { + originalTokens: 100, + compressedTokens: compressed ? 50 : 100, + savingsPercent: compressed ? 50 : 0, + techniquesUsed: compressed ? ["test"] : [], + mode: "rtk", + timestamp: Date.now(), + }, + }; +} + +function toolCompressor(calls: Array>) { + return async (body: Record): Promise => { + calls.push(structuredClone(body)); + const field = Array.isArray(body.messages) ? "messages" : "input"; + const items = body[field] as Array>; + return result({ + ...body, + [field]: items.map((item) => + item.role === "tool" ? { ...item, content: `compressed:${String(item.content)}` } : item + ), + }); + }; +} + +const options = { + principalId: "key-1", + sessionId: "session-1", + variant: { mode: "rtk", config: {} }, +}; + +beforeEach(() => resetLiveZoneCache()); + +describe("cache-aligned live-zone compression", () => { + it("compresses normally once, then sends only appended items through the engine", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const first = { + messages: [ + { role: "system", content: "stable" }, + { role: "user", content: "run tests" }, + { role: "tool", content: "100 noisy lines" }, + ], + }; + const firstResult = await applyLiveZoneCompression(first, options, compress); + const frozenBytes = JSON.stringify((firstResult.body.messages as unknown[]).slice(0, 3)); + + const second = { + messages: [ + ...first.messages, + { role: "assistant", content: "Need the failing test" }, + { role: "tool", content: "50 more noisy lines" }, + ], + }; + const secondResult = await applyLiveZoneCompression(second, options, compress); + + assert.equal(calls.length, 2); + assert.deepEqual(calls[1].messages, [second.messages[4]]); + assert.equal( + JSON.stringify((secondResult.body.messages as unknown[]).slice(0, 3)), + frozenBytes, + "the provider-facing prefix must remain byte-stable" + ); + assert.equal( + (secondResult.body.messages as Array<{ content: string }>)[3].content, + "Need the failing test" + ); + assert.equal( + (secondResult.body.messages as Array<{ content: string }>)[4].content, + "compressed:50 more noisy lines" + ); + assert.deepEqual(secondResult.stats?.liveZone, { + cacheHit: true, + frozenItems: 3, + liveItems: 2, + }); + }); + + it("keeps transformed top-level prefix fields stable and skips unchanged requests", async () => { + const calls: Array> = []; + const compress = async (body: Record): Promise => { + calls.push(structuredClone(body)); + return result({ + ...body, + instructions: `${String(body.instructions)}:${calls.length}`, + }); + }; + const first = { + instructions: "stable instructions", + tools: [{ type: "function", name: "run" }], + messages: [{ role: "user", content: "hello" }], + }; + const firstResult = await applyLiveZoneCompression(first, options, compress); + const unchangedResult = await applyLiveZoneCompression(first, options, compress); + const appendedResult = await applyLiveZoneCompression( + { ...first, messages: [...first.messages, { role: "tool", content: "new output" }] }, + options, + compress + ); + + assert.equal(calls.length, 2, "an unchanged request must not run the compressor again"); + assert.equal(unchangedResult.body.instructions, firstResult.body.instructions); + assert.equal(appendedResult.body.instructions, firstResult.body.instructions); + assert.deepEqual(appendedResult.body.tools, firstResult.body.tools); + assert.deepEqual(calls[1].messages, [{ role: "tool", content: "new output" }]); + }); + + it("fails open to full compression when any raw prefix item changes", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + await applyLiveZoneCompression( + { messages: [{ role: "user", content: "original" }] }, + options, + compress + ); + await applyLiveZoneCompression( + { + messages: [ + { role: "user", content: "edited" }, + { role: "tool", content: "new" }, + ], + }, + options, + compress + ); + + assert.equal((calls[1].messages as unknown[]).length, 2); + }); + + it("fails open when stable system or tool metadata changes", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const messages = [{ role: "user", content: "same" }]; + await applyLiveZoneCompression( + { system_instruction: "first", tools: [{ name: "one" }], messages }, + options, + compress + ); + await applyLiveZoneCompression( + { + system_instruction: "changed", + tools: [{ name: "two" }], + messages: [...messages, { role: "tool", content: "new" }], + }, + options, + compress + ); + + assert.equal((calls[1].messages as unknown[]).length, 2); + }); + + it("isolates cached prefixes by principal and compression variant", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const body = { messages: [{ role: "user", content: "same" }] }; + await applyLiveZoneCompression(body, options, compress); + await applyLiveZoneCompression( + { messages: [...body.messages, { role: "tool", content: "other principal" }] }, + { ...options, principalId: "key-2" }, + compress + ); + await applyLiveZoneCompression( + { messages: [...body.messages, { role: "tool", content: "other mode" }] }, + { ...options, variant: { mode: "caveman", config: {} } }, + compress + ); + + assert.equal((calls[1].messages as unknown[]).length, 2); + assert.equal((calls[2].messages as unknown[]).length, 2); + assert.equal(getLiveZoneCacheStats().entries, 3); + }); + + it("supports Responses input arrays and bypasses caching without an authenticated principal", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const first = { input: [{ role: "user", content: "hello" }] }; + await applyLiveZoneCompression(first, options, compress); + await applyLiveZoneCompression( + { input: [...first.input, { role: "tool", content: "response output" }] }, + options, + compress + ); + assert.deepEqual(calls[1].input, [{ role: "tool", content: "response output" }]); + + await applyLiveZoneCompression(first, { ...options, principalId: undefined }, compress); + await applyLiveZoneCompression(first, { ...options, principalId: undefined }, compress); + assert.equal(getLiveZoneCacheStats().entries, 1); + }); + + it("compresses only new Responses API tool outputs", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const first = { input: [{ type: "message", role: "user", content: "hello" }] }; + await applyLiveZoneCompression(first, options, compress); + const functionCall = { type: "function_call", call_id: "call-1", name: "run" }; + const output = { type: "function_call_output", call_id: "call-1", output: "noisy" }; + const result = await applyLiveZoneCompression( + { input: [...first.input, functionCall, output] }, + options, + compress + ); + + assert.deepEqual(calls[1].input, [output]); + assert.deepEqual((result.body.input as unknown[])[1], functionCall); + }); + + it("compresses tool_result roles and types in the live zone", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const first = { input: [{ role: "user", content: "hello" }] }; + await applyLiveZoneCompression(first, options, compress); + const roleOutput = { role: "tool_result", content: "role output" }; + const typeOutput = { type: "tool_result", content: "type output" }; + await applyLiveZoneCompression( + { input: [...first.input, roleOutput, typeOutput] }, + options, + compress + ); + + assert.deepEqual(calls[1].input, [roleOutput, typeOutput]); + }); + + it("bypasses live-zone reuse when a global hard budget is configured", async () => { + const calls: Array> = []; + const compress = toolCompressor(calls); + const hardBudgetOptions = { + ...options, + variant: { mode: "stacked", config: { targetTokens: 100 } }, + }; + const first = { messages: [{ role: "user", content: "hello" }] }; + await applyLiveZoneCompression(first, hardBudgetOptions, compress); + await applyLiveZoneCompression( + { messages: [...first.messages, { role: "tool", content: "new" }] }, + hardBudgetOptions, + compress + ); + assert.equal((calls[1].messages as unknown[]).length, 2); + assert.equal(getLiveZoneCacheStats().entries, 0); + }); +});