compaction: the service — settle kick, turn-start gate, registry mutex (round 14 slice 2)
- 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)
This commit is contained in:
@@ -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. */
|
||||
|
||||
@@ -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<ModelSlotName, EffectiveSlot | null>;
|
||||
appearance: EffectiveAppearance;
|
||||
graph: Required<GraphSettings>;
|
||||
/** Compaction (owner round 14): resolved concrete — absent record
|
||||
* section resolves to the factory 150k. */
|
||||
compaction: Required<CompactionSettings>;
|
||||
}
|
||||
|
||||
/** 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,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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<GraphSettings>;
|
||||
compaction: Required<CompactionSettings>;
|
||||
}
|
||||
|
||||
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<string, unknown>): 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<string, unknown>): 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<string, unknown>);
|
||||
if (r.compaction !== undefined && r.compaction !== null && typeof r.compaction === "object") {
|
||||
out.compaction = sanitizeCompaction(r.compaction as Record<string, unknown>);
|
||||
}
|
||||
}
|
||||
// Unknown top-level keys ride along untouched (forward-compat). One
|
||||
// known-DEAD key is stripped instead of riding: `chat` (the old
|
||||
|
||||
@@ -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<Response> {
|
||||
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
|
||||
|
||||
@@ -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<string, Promise<void>>;
|
||||
}
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
// 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));
|
||||
}
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user