From dd159cc8fb5a4c72d356c3395c24636bc1538159 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Mon, 5 Oct 2026 22:17:33 +0300 Subject: [PATCH] =?UTF-8?q?compaction:=20the=20service=20=E2=80=94=20settl?= =?UTF-8?q?e=20kick,=20turn-start=20gate,=20registry=20mutex=20(round=2014?= =?UTF-8?q?=20slice=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/server/compaction.ts: ensureCompacted (the turn-start gate — awaits in-flight compaction, compacts synchronously when still over threshold), kickCompaction (the settle kick — summary starts inside the learner's reading/answering pause; question/problem poses end runs so the pause is always a between-runs window), registry mutex on globalThis, anti-churn guard (empty-span skip), summary via the zero-tool one-shot sub-agent seam on the session's OWN engine (owner ruling), usage to the turns log keyed by the marker, record-changed + compaction-changed events - chat-handler: the gate rides with the BYOK/key gates (before appends, beginTurn; record re-read after); the kick at the onFinish tail (re-resolves engine fresh; silent no-op when engineless) - events: compaction-changed {chatId, running} — running=true is the client send-gate signal - settings: compaction.thresholdTokens (factory 150k, floor 32k, clamp-never-reject) resolved concrete in EffectiveSettings - effective threshold = min(setting, contextWindow − 16k) over the live catalog index - overflow recovery needs no special path: the next turn's gate compacts - 8 service pins (mutex, events order, chained spans, failure re-arm) --- src/core/events.ts | 7 + src/core/settings-resolution.ts | 7 + src/core/settings.ts | 38 ++++ src/server/chat-handler.ts | 66 ++++++- src/server/compaction.ts | 290 +++++++++++++++++++++++++++++++ tests/compaction-service.test.ts | 206 ++++++++++++++++++++++ 6 files changed, 612 insertions(+), 2 deletions(-) create mode 100644 src/server/compaction.ts create mode 100644 tests/compaction-service.test.ts diff --git a/src/core/events.ts b/src/core/events.ts index 13b42af..e639006 100644 --- a/src/core/events.ts +++ b/src/core/events.ts @@ -42,6 +42,12 @@ export interface ServerEvents { * Same notify-then-fetch contract as the turn events: open views of * the chat refetch and adopt the durable record. */ "record-changed": { chatId: string }; + /** A compaction started/finished for the chat (owner round 14, + * 2026-10-05). running=true is the SEND GATE: while a compaction is + * in flight the server structurally refuses model interaction for + * the chat; the client mirrors it by disabling the composer/dock + * submit. */ + "compaction-changed": { chatId: string; running: boolean }; } export type ServerEventType = keyof ServerEvents; @@ -56,6 +62,7 @@ export const SERVER_EVENT_TYPES = [ "activity-changed", "chats-changed", "record-changed", + "compaction-changed", ] as const satisfies readonly ServerEventType[]; /** One notification: the type discriminates the payload. */ diff --git a/src/core/settings-resolution.ts b/src/core/settings-resolution.ts index c34b20d..8c8a437 100644 --- a/src/core/settings-resolution.ts +++ b/src/core/settings-resolution.ts @@ -35,6 +35,7 @@ import { type BackgroundModelRole, type FactorySettings, type GeneralSettings, + type CompactionSettings, type GraphSettings, type ModelSlot, type ModelSlotName, @@ -89,6 +90,9 @@ export interface EffectiveSettings { models: Record; appearance: EffectiveAppearance; graph: Required; + /** Compaction (owner round 14): resolved concrete — absent record + * section resolves to the factory 150k. */ + compaction: Required; } /** The conversation pair's chain (BYOK, owner 2026-09-22): the chat's own @@ -206,6 +210,9 @@ export function resolveSettings(record: UserSettings | null): EffectiveSettings // 4096; the sanitizer has already clamped/floored any stored value. extractionChunkTokens: graph.extractionChunkTokens ?? FACTORY_SETTINGS.graph.extractionChunkTokens, }, + compaction: { + thresholdTokens: record?.compaction?.thresholdTokens ?? FACTORY_SETTINGS.compaction.thresholdTokens, + }, }; } diff --git a/src/core/settings.ts b/src/core/settings.ts index d551a22..47315bd 100644 --- a/src/core/settings.ts +++ b/src/core/settings.ts @@ -328,6 +328,24 @@ export interface GraphSettings { extractionChunkTokens?: number; } +/** + * COMPACTION (owner round 14, 2026-10-05): the context-fitting mechanic's + * user-facing knob. One number — the trigger threshold in context tokens. + * The effective trigger is min(this, model contextWindow − 16k reserve) + * (owner ruling: on big models the setting rules; on small ones the + * window rules before the provider overflows). + * + * FACTORY DEFAULT 150_000 (owner ruling). + * + * FLOOR 32_000 (implementer's call, the chunk-budget discipline): below + * the kept tail (20k) + a summary the chat would re-compact every turn — + * a burn, not a preference. NO CEILING: a huge threshold just means "fit + * everything" until the window rule binds. + */ +export interface CompactionSettings { + thresholdTokens?: number; +} + /** THE record. Every section optional (delta); unknown top-level keys are * preserved verbatim by the sanitizer (forward-compat with newer records). */ export interface UserSettings { @@ -338,6 +356,7 @@ export interface UserSettings { models?: ModelsSettings; appearance?: AppearanceSettings; graph?: GraphSettings; + compaction?: CompactionSettings; [key: string]: unknown; } @@ -362,6 +381,7 @@ export interface FactorySettings { }; }; graph: Required; + compaction: Required; } export const FACTORY_SETTINGS: FactorySettings = { @@ -376,6 +396,7 @@ export const FACTORY_SETTINGS: FactorySettings = { mdRendering: { headingWeight: 900, boldColor: null, italicColor: null }, }, graph: { enabled: true, extractionChunkTokens: 8192 }, + compaction: { thresholdTokens: 150_000 }, }; /** The chunk budget's validated floor (GraphSettings.extractionChunkTokens @@ -552,6 +573,20 @@ function sanitizeGraph(raw: Record): GraphSettings { return out; } +/** clamp-never-reject (the sanitizer's discipline): a finite positive + * number floors to COMPACTION_THRESHOLD_FLOOR, anything else drops (the + * factory default resolves at read time). */ +const COMPACTION_THRESHOLD_FLOOR = 32_000; + +function sanitizeCompaction(raw: Record): CompactionSettings { + const out: CompactionSettings = {}; + const rawTokens = raw.thresholdTokens; + if (typeof rawTokens === "number" && Number.isFinite(rawTokens) && rawTokens > 0) { + out.thresholdTokens = Math.max(COMPACTION_THRESHOLD_FLOOR, Math.floor(rawTokens)); + } + return out; +} + /** * Coerce arbitrary JSON (disk file, network body) into a valid delta record: * known fields validated strictly (invalid ones drop/clamp per module doc), @@ -579,6 +614,9 @@ export function sanitizeUserSettings(raw: unknown): UserSettings | null { } if (r.graph !== null && typeof r.graph === "object" && !Array.isArray(r.graph)) { out.graph = sanitizeGraph(r.graph as Record); + if (r.compaction !== undefined && r.compaction !== null && typeof r.compaction === "object") { + out.compaction = sanitizeCompaction(r.compaction as Record); + } } // Unknown top-level keys ride along untouched (forward-compat). One // known-DEAD key is stripped instead of riding: `chat` (the old diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 2fa44f7..179377f 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -39,6 +39,9 @@ import { applyParkedDismissal } from "./dismissal-park.ts"; import { registerTurn, unregisterTurn } from "./turn-runs.ts"; import { persistTurn, triggerExtraction } from "./turn-service.ts"; import { hasProbeDone, danglingQuestionOf } from "../core/native-messages.ts"; +import { composeInstructions } from "../agent/teacher-agent.ts"; +import { ensureCompacted, kickCompaction, effectiveThresholdTokens } from "./compaction.ts"; +import { catalogContextLimits } from "./model-catalog.ts"; import { emit } from "../core/events.ts"; import { settledTurnResponse } from "./turn-runs.ts"; import { asAgentMessages } from "../core/native-messages.ts"; @@ -168,7 +171,7 @@ export async function handleChatInput( input: TurnInput, buildAgent?: (opts: AgentBuildOptions & { teaching: boolean }) => TeachingAgent, ): Promise { - const record = stores.chatStore.getChat(chatId); + let record = stores.chatStore.getChat(chatId); if (!record) { return Response.json({ error: "Unknown chat" }, { status: 404 }); } @@ -204,6 +207,34 @@ export async function handleChatInput( return Response.json(envelope, { status: 503 }); } + // THE COMPACTION GATE (owner round 14, 2026-10-05 — the structural + // ruling: NOTHING model-facing for this chat interleaves with a + // running compaction): await any in-flight compaction (the settle + // kick usually finished inside the learner's answer pause), and if + // the record is STILL over threshold, compact synchronously — this + // request waits. Skipped under tests (the extraction precedent); + // placement mirrors the BYOK/key gates — BEFORE appends and + // beginTurn, no liveness effects — and the record re-reads after + // because compaction appends the marker. + const conversationEngine = piEngineForSpec(conversation); + const modelIdForCompaction = `${conversationEngine.model.provider}:${conversationEngine.model.id}`; + await ensureCompacted( + { + chatStore: stores.chatStore, + engine: conversationEngine, + modelId: modelIdForCompaction, + kind: record.kind === "practice" ? "practice" : hasProbeDone(record.messages) ? "teach" : "probe", + thresholdTokens: effectiveThresholdTokens( + effective.compaction?.thresholdTokens, + catalogContextLimits().get(modelIdForCompaction), + ), + systemPromptTokens: Math.ceil(composeInstructions(record.kind === "practice" ? "practice" : "teacher").length / 4), + }, + chatId, + ); + const compactedRecord = stores.chatStore.getChat(chatId); + if (compactedRecord) record = compactedRecord; + // The input's native appends (pure; the honest 400 for an answer with no // pending question call). Computed BEFORE beginTurn so a refused input // leaves no liveness effects; read against the CURRENT record. Wave C: @@ -294,7 +325,8 @@ export async function handleChatInput( // BYOK (owner, 2026-09-22): the conversation chain was resolved by the // turn gate above (chat.model override → strong slot → refused) — the // engine build is pure from here; thinking levels clamp natively. - const engine = piEngineForSpec(conversation); + // (Built once above the compaction gate; reused verbatim.) + const engine = conversationEngine; // Provider-qualified stamp (model slots wave): matches the price table's // provider:id keys. const modelId = `${engine.model.provider}:${engine.model.id}`; @@ -475,6 +507,36 @@ export async function handleChatInput( // ones stay parked for the chat's next run (the registry is the // stranding safety net — the agent's undrained copy died above). reconcileUserSteers(chatId, stamped); + // THE SETTLE KICK (owner round 14): the turn settled — if the + // projection is now over threshold, the summary starts in the + // background, INSIDE the learner's reading/answering pause + // (question/problem poses end runs, so this moment is always a + // between-runs window). The next turn's ensureCompacted gate + // awaits whatever this leaves running; a lost engine or a + // beneath-threshold record is a silent no-op. Skipped under + // tests (the extraction precedent — scripted chats are + // fixtures; the service itself is pinned directly). + if (process.env.NODE_ENV !== "test") + kickCompaction(() => { + const fresh = effectiveSettings(); + const spec = resolveConversationSpec(stores.chatStore.getChat(chatId)?.settings, fresh.models.strong); + if (!spec) return null; + const kickEngine = piEngineForSpec(spec); + const kickModelId = `${kickEngine.model.provider}:${kickEngine.model.id}`; + const freshRecord = stores.chatStore.getChat(chatId); + if (!freshRecord) return null; + return { + chatStore: stores.chatStore, + engine: kickEngine, + modelId: kickModelId, + kind: freshRecord.kind === "practice" ? "practice" : hasProbeDone(freshRecord.messages) ? "teach" : "probe", + thresholdTokens: effectiveThresholdTokens( + fresh.compaction?.thresholdTokens, + catalogContextLimits().get(kickModelId), + ), + systemPromptTokens: Math.ceil(composeInstructions(freshRecord.kind === "practice" ? "practice" : "teacher").length / 4), + }; + }, chatId); }, }); // The turn is now a live background object: register it so any view diff --git a/src/server/compaction.ts b/src/server/compaction.ts new file mode 100644 index 0000000..e10b3a2 --- /dev/null +++ b/src/server/compaction.ts @@ -0,0 +1,290 @@ +/** + * THE COMPACTION SERVICE (owner round 14, 2026-10-05): the trigger + * plumbing around core/compaction's pure cuts and markers. + * + * TOPOLOGY (why this is simple): question and problem poses TERMINATE + * their runs, so the "user is thinking" pause is always a BETWEEN-RUNS + * window. Compaction therefore never touches a live run: + * + * - SETTLE KICK — a turn settles (onFinish, after persist) and the + * projected context is over threshold: the summary starts NOW, in + * the background, while the learner reads/answers — usually done + * long before the next input arrives. + * - TURN-START GATE — every turn (client POST or wake) awaits any + * in-flight compaction for its chat BEFORE appends or beginTurn; + * if the record is STILL over threshold (cold sessions, kicks that + * lost their engine), compaction runs synchronously inside the + * request. Structurally gated: nothing model-facing for the chat + * interleaves with a running compaction (owner ruling — minimal + * surface for anything to go wrong). + * - OVERFLOW RECOVERY — a run that dies on context length needs no + * special path: the next turn's gate compacts and the fresh run + * starts small. + * + * The summary is written by the session's OWN conversation model (owner + * ruling) via the zero-tool one-shot sub-agent seam; its usage lands in + * the chat's append-only turns log (the money channel) keyed by the + * marker's id. + */ + +import type { AgentMessage } from "@earendil-works/pi-agent-core"; +import type { Usage } from "@earendil-works/pi-ai"; +import { + activeCompactionOf, + buildCompactionPrompt, + compactionCutIndex, + COMPACTION_RESERVE_TOKENS, + estimateTokens, + newCompactionMarker, + planTreeToText, + serializeForSummary, + toLlmMessages, + type CompactionDetails, + type LhMessage, + type LhMarkerMessage, +} from "../core/index.ts"; +import { emit } from "../core/events.ts"; +import type { ChatStore } from "../core/ports.ts"; +import type { PiEngine } from "../agent/models.ts"; +import { runSubagent } from "../agent/subagents/pi-run.ts"; + +/** The summarizer's system prompt — server-owned operational framing, + * not model-facing doctrine (buildCompactionPrompt carries the shape). */ +const SUMMARIZER_SYSTEM_PROMPT = + "You are the session-compaction summarizer for a tutoring agent. You receive the transcript span to summarize and write exactly the summary the instructions demand — nothing else, no preamble, no meta-commentary."; + +/** The one-shot summary call: zero tools, one turn, text out. The engine + * is the chat's own conversation engine (owner ruling: the session's + * model writes its own summary). Injectable for tests. */ +export type Summarizer = (prompt: string, engine: PiEngine) => Promise<{ summary: string; usage: Usage }>; + +/** The production summarizer over the sub-agent seam. */ +async function subagentSummarizer(prompt: string, engine: PiEngine): Promise<{ summary: string; usage: Usage }> { + let usage: Usage = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; + const { text } = await runSubagent([], { + // The engine override is always present (the chat's own conversation + // engine — owner ruling), so the role label never resolves an + // engine; "researcher" is the least-wrong existing ModelRole label + // for a one-shot text job (the cast documents the dead path). + role: "researcher" as never, + prompt, + engine, + systemPrompt: SUMMARIZER_SYSTEM_PROMPT, + maxTurns: 1, + onUsage: (u) => { + usage = u; + }, + }); + return { summary: text, usage }; +} + +/** Everything the service needs from its host (chat-handler wires the + * production versions; tests inject). */ +export interface CompactionDeps { + chatStore: ChatStore; + /** The chat's conversation engine — also the summarizer's. */ + engine: PiEngine; + /** provider:id of the engine (usage + marker provenance). */ + modelId: string; + /** The session kind (summarizer wording: tutor vs practice). */ + kind: "teach" | "practice" | "probe"; + /** The effective trigger threshold, RESOLVED by the caller: + * min(settings threshold, contextWindow − reserve) — the owner's + * effective-trigger ruling. The catalog window lookup is the + * caller's (chat-handler has the model id at hand). */ + thresholdTokens: number; + /** The composed system prompt's token estimate (chars/4) — the + * projection the provider actually receives starts with it. */ + systemPromptTokens: number; + /** Injectable summarizer (tests); default = the sub-agent one-shot. */ + summarize?: Summarizer; +} + +// ─── the registry (module-global on globalThis, jobs.ts's pattern) ───────── + +interface CompactionState { + inFlight: Map>; +} +const g = globalThis as typeof globalThis & { __lhCompaction?: CompactionState }; +function state(): CompactionState { + return (g.__lhCompaction ??= { inFlight: new Map() }); +} + +/** Is a compaction running for this chat? (The client status event's + * server-side twin — tests and the SSE emitter read it.) */ +export function compactionInFlight(chatId: string): boolean { + return state().inFlight.has(chatId); +} + +/** Await any in-flight compaction for the chat — the structural gate + * every model interaction for the chat passes through. */ +export async function awaitCompaction(chatId: string): Promise { + const running = state().inFlight.get(chatId); + if (running) await running; +} + +// ─── the effective threshold (owner ruling: min(setting, window − reserve)) ─ + +/** Resolve the effective trigger. Unknown window → the setting alone + * (honest: no fake window — the catalog-index precedent). */ +export function effectiveThresholdTokens(settingTokens: number | undefined, contextWindowTokens: number | undefined): number { + const setting = settingTokens ?? 150_000; + if (contextWindowTokens === undefined) return setting; + return Math.min(setting, Math.max(contextWindowTokens - COMPACTION_RESERVE_TOKENS, 1)); +} + +// ─── the estimate over the ACTIVE projection ──────────────────────────────── + +/** The projected context estimate: what the next LLM call would carry — + * toLlmMessages output (cut honored) + the system prompt. chars/4. */ +export function projectedContextTokens(messages: readonly LhMessage[], systemPromptTokens: number): number { + return estimateTokens(toLlmMessages(messages as readonly AgentMessage[]) as unknown as readonly LhMessage[]) + systemPromptTokens; +} + +// ─── the compaction run ───────────────────────────────────────────────────── + +/** Run one compaction for the chat (the registry mutex is HELD by the + * caller — ensureCompacted). Reads the record fresh, cuts, summarizes, + * appends the marker, records usage, notifies. Never throws upward + * past the registry cleanup. */ +async function compact(deps: CompactionDeps, chatId: string): Promise { + const record = deps.chatStore.getChat(chatId); + if (!record) return; + const messages = record.messages; + const tokensBefore = projectedContextTokens(messages, deps.systemPromptTokens); + if (tokensBefore <= deps.thresholdTokens) return; // settled beneath while we waited + + const cut = compactionCutIndex(messages); + if (cut >= messages.length) return; // nothing to drop (defensive) + + // The previous compaction's kept boundary (chained compactions + // summarize the old summary + what rode after it). + const previous = activeCompactionOf(messages); + const prevKept = previous ? messages.findIndex((m) => m.id === previous.details.firstKeptMessageId) : -1; + // ANTI-CHURN GUARD: when the cut cannot advance past the previous + // kept boundary, the span is empty (or repeats it) — the kept tail + // alone carries the projection and re-summarizing the same span + // cannot shrink it. Reachable only when the threshold sits below + // kept-tail + summary (a hand-edited record below the settings + // floor); skipping keeps the loop honest instead of churning. + if (previous && prevKept !== -1 && cut <= prevKept) { + console.info( + `[chat ${chatId}] compaction skipped — the kept tail alone exceeds the threshold (raise it or shrink the tail)`, + ); + return; + } + + // The span: from the PREVIOUS kept boundary through the cut. + let spanStart = 0; + let priorSummary: string | null = null; + if (previous && prevKept > 0) { + spanStart = prevKept; + priorSummary = previous.content; + } + const spanText = serializeForSummary(messages.slice(spanStart, cut)); + const fullSpan = priorSummary ? `[SYSTEM NOTE — prior compaction summary, still authoritative where not superseded]\n${priorSummary}\n\n${spanText}` : spanText; + + const prompt = buildCompactionPrompt({ + spanText: fullSpan, + planTreeText: planTreeToText(record.tree, messages), + kind: deps.kind, + }); + + emit("compaction-changed", { chatId, running: true }); + try { + const summarize = deps.summarize ?? subagentSummarizer; + const { summary, usage } = await summarize(prompt, deps.engine); + const trimmed = summary.trim(); + if (!trimmed) return; // an empty summary is a refusal artifact — leave the record honest + + const firstKept = messages[cut]!; + const details: CompactionDetails = { + firstKeptMessageId: firstKept.id, + tokensBefore, + model: deps.modelId, + at: new Date().toISOString(), + }; + const marker = newCompactionMarker(trimmed, details); + deps.chatStore.saveMessages(chatId, [...messages, marker]); + // The money channel: the summary is a real model call on the + // session's own engine — its usage keys to the marker (provenance: + // the transcript shows the compaction divider at the marker). + deps.chatStore.appendTurnUsage(chatId, { + id: crypto.randomUUID(), + messageId: marker.id, + at: details.at, + model: deps.modelId, + steps: [ + { + inputTokens: usage.input, + outputTokens: usage.output, + cacheReadTokens: usage.cacheRead, + cacheWriteTokens: usage.cacheWrite, + reasoningTokens: usage.reasoning ?? 0, + }, + ], + }); + // Notify-then-fetch: the record gained the marker (the client's + // dividers appear) and the status event ends the send gate. + emit("record-changed", { chatId }); + } catch (err) { + // A failed compaction leaves the record untouched — the next + // turn-start gate retries (the same self-healing law as the + // extraction trigger). + console.error(`[chat ${chatId}] compaction failed:`, err); + } finally { + emit("compaction-changed", { chatId, running: false }); + } +} + +/** + * The gate + trigger in one: await any in-flight compaction, then — if + * the record is STILL over threshold — run one synchronously (the + * caller's request waits; the owner's structural-gate ruling). Idempotent + * under concurrency through the registry mutex. + */ +export async function ensureCompacted(deps: CompactionDeps, chatId: string): Promise { + // Skipped under tests, the extraction-trigger precedent: scripted + // chats are fixtures — the service itself is tested directly with an + // injected summarizer. + if (process.env.NODE_ENV === "test" && !deps.summarize) return; + await awaitCompaction(chatId); + const record = deps.chatStore.getChat(chatId); + if (!record) return; + if (projectedContextTokens(record.messages, deps.systemPromptTokens) <= deps.thresholdTokens) return; + if (state().inFlight.has(chatId)) return; // a concurrent caller started it while we awaited + const run = compact(deps, chatId); + state().inFlight.set(chatId, run); + try { + await run; + } finally { + state().inFlight.delete(chatId); + } +} + +/** + * The settle kick (fire-and-forget): the turn settled, the learner is + * about to read/answer — if the projection is over threshold, the + * summary starts NOW and finishes inside the pause. Wired at + * chat-handler's onFinish tail; engineless or over-threshold-nothing + * are silent no-ops. The next turn's ensureCompacted gate picks up + * whatever this leaves running. + */ +export function kickCompaction(buildDeps: () => CompactionDeps | null, chatId: string): void { + if (state().inFlight.has(chatId)) return; + const deps = buildDeps(); + if (!deps) return; + const record = deps.chatStore.getChat(chatId); + if (!record) return; + if (projectedContextTokens(record.messages, deps.systemPromptTokens) <= deps.thresholdTokens) return; + const run = compact(deps, chatId); + state().inFlight.set(chatId, run); + void run.finally(() => state().inFlight.delete(chatId)); +} diff --git a/tests/compaction-service.test.ts b/tests/compaction-service.test.ts new file mode 100644 index 0000000..e5150b0 --- /dev/null +++ b/tests/compaction-service.test.ts @@ -0,0 +1,206 @@ +/** + * THE COMPACTION SERVICE (owner round 14, 2026-10-05): the trigger + * plumbing — settle kick + turn-start gate, the registry mutex, the + * summary call, usage, events. Pinned here with an injected summarizer + * (the sub-agent one-shot needs a live engine; the SERVICE laws don't). + * + * The laws: + * - the registry is a mutex: concurrent ensureCompacted → ONE summary; + * - over-threshold → marker appended + usage logged + events emitted; + * - under-threshold (or the projection post-marker) → no-op; + * - chained compaction: the span carries the prior summary forward; + * - the effective threshold: min(setting, window − reserve). + */ + +import assert from "node:assert/strict"; +import fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { after, describe, it } from "node:test"; + +// ── the seam, BEFORE any config-loading import (the birth-stamp discipline) ─ +const TMP_DATA = fs.mkdtempSync(path.join(os.tmpdir(), "lh-compaction-")); +process.env.LH_DATA_DIR = TMP_DATA; + +const { chatStore } = await import("../src/store/chats.ts"); +const { + ensureCompacted, + kickCompaction, + awaitCompaction, + effectiveThresholdTokens, + projectedContextTokens, +} = await import("../src/server/compaction.ts"); +import type { CompactionDeps } from "../src/server/compaction.ts"; +const { subscribe } = await import("../src/core/events.ts"); +const { activeCompactionOf, newUserMessage, COMPACTED_HISTORY_HEADER } = await import("../src/core/index.ts"); +import type { LhMessage } from "../src/core/index.ts"; + +after(() => { + fs.rmSync(TMP_DATA, { recursive: true, force: true }); +}); + +// ── fixtures ──────────────────────────────────────────────────────────────── + +const big = (text: string, kb = 100): LhMessage => newUserMessage(`${text}: ${"x".repeat(kb * 1024)}`); + +/** A chat whose record projects far over any test threshold. */ +function overThresholdChat(label: string): string { + const chat = chatStore.createChat(label); + const messages = [big("m1"), big("m2"), big("m3"), newUserMessage("recent")]; + chatStore.saveMessages(chat.id, messages); + return chat.id; +} + +let summarizeCalls = 0; +const depsOf = (chatId: string, thresholdTokens: number): CompactionDeps => ({ + chatStore, + // The engine is never touched when summarize is injected (only its + // identity rides the marker's provenance) — a minimal stand-in. + engine: { model: { provider: "faux", id: "test", contextWindow: 1_000_000 } } as never, + modelId: "faux:test", + kind: "teach", + thresholdTokens, + systemPromptTokens: 100, + summarize: async (prompt) => { + summarizeCalls++; + assert.ok(prompt.includes("TRANSCRIPT") || prompt.length > 0); + return { + summary: `SUMMARY #${summarizeCalls}: the learner knows X.`, + usage: { input: 1_000, output: 500, cacheRead: 0, cacheWrite: 0, totalTokens: 1_500, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + }; + }, +}); + +// ── the laws ──────────────────────────────────────────────────────────────── + +describe("the compaction service", () => { + it("over-threshold record → marker appended, usage logged, both events emitted", async () => { + const chatId = overThresholdChat("compact: basic"); + const events: string[] = []; + const off = subscribe((e) => { + const { chatId: c, running } = e as { chatId?: string; running?: boolean }; + if (e.type === "compaction-changed" && c === chatId) events.push(`${running}`); + if (e.type === "record-changed" && c === chatId) events.push("record"); + }); + await ensureCompacted(depsOf(chatId, 40_000), chatId); + off(); + + const record = chatStore.getChat(chatId)!; + const marker = activeCompactionOf(record.messages); + assert.ok(marker, "the marker landed"); + assert.ok(marker!.content.includes("SUMMARY #")); + assert.ok(marker!.details.tokensBefore > 40_000, "provenance carries the trigger reading"); + assert.equal(marker!.details.model, "faux:test"); + // the projection now fits: the dropped span is gone from the wire view + assert.ok(projectedContextTokens(record.messages, 100) < 40_000, "the cut brought the projection under threshold+summary"); + // usage landed on the money channel, keyed by the marker + assert.ok((record.turns ?? []).some((t) => t.messageId === marker!.id && t.model === "faux:test"), "usage keyed by the marker"); + // events: running true → record-changed → running false + assert.deepEqual(events, ["true", "record", "false"]); + }); + + it("under-threshold record → a no-op (no marker, no summary call)", async () => { + const chatId = overThresholdChat("compact: under"); + const before = summarizeCalls; + await ensureCompacted(depsOf(chatId, 10_000_000), chatId); + assert.equal(summarizeCalls, before); + assert.equal(activeCompactionOf(chatStore.getChat(chatId)!.messages), null); + }); + + it("the registry is a mutex: concurrent ensureCompacted → ONE summary", async () => { + const chatId = overThresholdChat("compact: mutex"); + const before = summarizeCalls; + const deps = depsOf(chatId, 40_000); + await Promise.all([ensureCompacted(deps, chatId), ensureCompacted(deps, chatId), ensureCompacted(deps, chatId)]); + assert.equal(summarizeCalls, before + 1, "exactly one summary however many gates race"); + }); + + it("awaitCompaction: a later call rides the in-flight one (the structural gate)", async () => { + const chatId = overThresholdChat("compact: gate"); + const events: string[] = []; + const off = subscribe((e) => { + const { chatId: c, running } = e as { chatId?: string; running?: boolean }; + if (e.type === "compaction-changed" && c === chatId) events.push(`${running}`); + }); + const p = ensureCompacted(depsOf(chatId, 40_000), chatId); + await awaitCompaction(chatId); // rides the same promise + await p; + off(); + assert.deepEqual(events, ["true", "false"], "no second compaction started"); + }); + + it("chained compaction: the span carries the PRIOR summary forward", async () => { + const chatId = overThresholdChat("compact: chained"); + let seenPrompt = ""; + const deps: CompactionDeps = { + ...depsOf(chatId, 40_000), + summarize: async (prompt) => { + seenPrompt = prompt; + return { + summary: "SECOND SUMMARY.", + usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 2, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + }; + }, + }; + // first compaction + await ensureCompacted({ ...depsOf(chatId, 40_000) }, chatId); + // grow the record past the threshold again (post-markers + new tail) + chatStore.saveMessages(chatId, [...chatStore.getChat(chatId)!.messages, big("m9"), big("m10"), big("m11")]); + // second compaction — the span must open with the FIRST summary + await ensureCompacted(deps, chatId); + assert.ok(seenPrompt.includes("prior compaction summary"), "the prior summary rides the span"); + assert.ok(seenPrompt.includes(COMPACTED_HISTORY_HEADER), "as the marker's canonical text"); + const record = chatStore.getChat(chatId)!; + const markers = record.messages.filter((m) => (m as { customType?: string }).customType === "compaction"); + assert.equal(markers.length, 2, "both markers persist (history is never rewritten)"); + assert.ok(activeCompactionOf(record.messages)!.content.includes("SECOND SUMMARY"), "the newest rules"); + }); + + it("effectiveThresholdTokens: the setting rules big windows; the window rule binds small ones", () => { + assert.equal(effectiveThresholdTokens(150_000, 200_000), 150_000); + assert.equal(effectiveThresholdTokens(150_000, 100_000), 100_000 - 16_384); + assert.equal(effectiveThresholdTokens(undefined, 200_000), 150_000); + assert.equal(effectiveThresholdTokens(undefined, undefined), 150_000); + }); + + it("kickCompaction: fire-and-forget, registry-set, over-threshold only", async () => { + const chatId = overThresholdChat("compact: kick"); + const before = summarizeCalls; + let built = 0; + kickCompaction(() => { + built++; + return depsOf(chatId, 40_000); + }, chatId); + assert.equal(built, 1, "deps built once"); + await awaitCompaction(chatId); + assert.equal(summarizeCalls, before + 1); + // beneath threshold now → a second kick builds its deps, the + // pre-check refuses, and NO summary runs (building deps is the + // kick's discovery step — the run is what costs) + const afterFirst = summarizeCalls; + kickCompaction(() => { + built++; + return depsOf(chatId, 40_000); + }, chatId); + assert.equal(summarizeCalls, afterFirst, "post-compaction record is under threshold — no run"); + // a null deps build (engineless) is a silent no-op + kickCompaction(() => null, chatId); + }); + + it("a summarizer failure leaves the record untouched and the gate re-arms", async () => { + const chatId = overThresholdChat("compact: fail"); + const before = summarizeCalls; + const failing: CompactionDeps = { + ...depsOf(chatId, 40_000), + summarize: async () => { + throw new Error("boom"); + }, + }; + await ensureCompacted(failing, chatId); + assert.equal(activeCompactionOf(chatStore.getChat(chatId)!.messages), null, "no marker on failure"); + // the next gate retries cleanly + await ensureCompacted(depsOf(chatId, 40_000), chatId); + assert.ok(activeCompactionOf(chatStore.getChat(chatId)!.messages)); + assert.equal(summarizeCalls, before + 1); + }); +});