diff --git a/agents/codelayer/src/agent.ts b/agents/codelayer/src/agent.ts index 99b0697..c2a20bc 100644 --- a/agents/codelayer/src/agent.ts +++ b/agents/codelayer/src/agent.ts @@ -350,6 +350,7 @@ function mergeHooks(base: ReturnType, hooks?: preToolUse: [...base.preToolUse, ...(hooks?.preToolUse ?? [])], postToolUse: [...saneDefaultOutputTruncationHooks, ...fileStatePostHooks, ...(hooks?.postToolUse ?? [])], preRequest: [...base.preRequest, ...(hooks?.preRequest ?? [])], + compaction: [...base.compaction, ...(hooks?.compaction ?? [])], } } diff --git a/agents/codelayer/src/coding-subagent-tool.ts b/agents/codelayer/src/coding-subagent-tool.ts index fc4e7a1..aefff38 100644 --- a/agents/codelayer/src/coding-subagent-tool.ts +++ b/agents/codelayer/src/coding-subagent-tool.ts @@ -173,6 +173,7 @@ function mergeHooks( preToolUse: [...base.preToolUse, ...(hooks?.preToolUse ?? [])], postToolUse: [...saneDefaultOutputTruncationHooks, ...fileStatePostHooks, ...(hooks?.postToolUse ?? [])], preRequest: [...base.preRequest, ...(hooks?.preRequest ?? [])], + compaction: [...base.compaction, ...(hooks?.compaction ?? [])], } } diff --git a/packages/agentlayer-core/src/agent-run.ts b/packages/agentlayer-core/src/agent-run.ts index 850ac47..ce82451 100644 --- a/packages/agentlayer-core/src/agent-run.ts +++ b/packages/agentlayer-core/src/agent-run.ts @@ -1,8 +1,9 @@ import type { ModelMessage } from 'ai' import type { RunResult } from './agent' +import type { CompactionTrigger } from './compaction' import type { ApprovalRequest } from './hooks' import type { ApprovalDecision } from './state' -import type { TokenUsageEvent } from './token-usage' +import type { ModelTokenUsage, TokenUsageEvent } from './token-usage' // ── AgentEvent — discriminated union for the async iterator ────────────────── @@ -85,6 +86,17 @@ export type AgentEvent = stepIndex: number finishReason?: string } & AgentEventMeta) + | ({ + type: 'compaction' + trigger: CompactionTrigger + priorContextWindowTokens?: number + replacedMessageCount: number + retainedMessageCount: number + summaryUsage: { + model: string + usage: Omit + } + } & AgentEventMeta) export class AgentRun implements AsyncIterable { private state: { type: 'pending' } | { type: 'resolved'; value: RunResult } = { type: 'pending' } diff --git a/packages/agentlayer-core/src/agent.ts b/packages/agentlayer-core/src/agent.ts index 0ce25bd..48d4500 100644 --- a/packages/agentlayer-core/src/agent.ts +++ b/packages/agentlayer-core/src/agent.ts @@ -1,31 +1,47 @@ import type { FinishReason as AiSdkFinishReason, LanguageModel, ModelMessage, TextStreamPart, ToolChoice } from 'ai' import { streamText, tool as toAiSdkTool } from 'ai' import { type AgentEvent, AgentRun } from './agent-run' +import { + type AutoCompactConfig, + buildCompactionRequestText, + buildTurnPrefixCompactionRequestText, + COMPACTION_SYSTEM_PROMPT, + type CompactionTrigger, + compactionSummaryMessage, + isContextOverflowError, + parseCompactCommand, + planCompaction, + resolveCompactionMaxOutputTokens, + resolveCompactionPolicy, + shouldCompactForThreshold, +} from './compaction' import type { Tool } from './define-tool' import { AgentError, InvalidMessagesError } from './errors' import { type ExecuteToolCallContext, executeToolCall, type ToolCallResult } from './execute-tool-call' import { type ApprovalHook, type ApprovalRequest, + type CompactionHook, type HookStateOperation, type PendingToolCall, type PostToolUseHook, type PreRequestHook, type PreToolUseHook, runApprovalHooks, + runCompactionHooks, runPostToolUseHooks, runPreRequestHooks, runPreToolUseHooks, type StopOptions, type ToolInfo, } from './hooks' -import { type AgentLayerToolOutput, buildToolResultMessage } from './messages' +import { type AgentLayerToolOutput, buildToolResultMessage, userMessage } from './messages' import { type ModelKey, ModelProvider } from './models' import { sanitizeTextForModelState, sanitizeToolOutputForModelState } from './sanitize-text' import type { AgentState, ApprovalDecision, ApprovalHistoryEntry, TerminalChildMap } from './state' import type { Step, StepToolResult, StopResult, StopTiming, StopWhen } from './stop-conditions' import { shouldStop } from './stop-conditions' -import { extractUsage, getModelKey, type TokenUsage, TokenUsageAccumulator } from './token-usage' +import { extractUsage, getModelKey, type ModelTokenUsage, type TokenUsage, TokenUsageAccumulator } from './token-usage' export type ProviderOptions = Parameters[0]['providerOptions'] export type ProviderOptionsFactory = (ctx: { runId: string; promptCacheKey?: string }) => ProviderOptions @@ -76,6 +92,8 @@ export interface AgentConfig> = Rec onStop?: (result: RunResult) => void | Promise /** Explicit context window limit override. When not set, resolved from models.dev if available. */ contextWindowLimit?: number + /** Provider-neutral compaction policy. Omitted means enabled with defaults. */ + autoCompact?: AutoCompactConfig /** Called when an approval is requested. Fires before the event is pushed to the iterator. Observe-only, errors swallowed. */ onApprovalRequested?: ( approval: ApprovalRequest, @@ -88,6 +106,7 @@ export interface AgentConfig> = Rec preToolUse?: PreToolUseHook[] postToolUse?: PostToolUseHook[] preRequest?: PreRequestHook[] + compaction?: CompactionHook[] } } @@ -115,6 +134,18 @@ export interface RunOptions { promptCacheKey?: string } +export interface CompactOptions { + state: AgentState + signal?: AbortSignal + stream?: boolean + promptCacheKey?: string + additionalInstructions?: string + /** Used by loop integrations; direct calls default to manual. */ + trigger?: CompactionTrigger + /** Override native-tail retention for this explicit checkpoint. */ + keepRecentTokens?: number +} + function convertTools(tools: Record>) { return Object.fromEntries( Object.entries(tools).map(([name, t]) => [ @@ -228,6 +259,12 @@ interface ExecutedModelStep { usage: Awaited } +interface PerformedCompaction { + state: AgentState + inferenceMessages: ModelMessage[] + event: Extract +} + function classifyOutcomes(outcomes: ToolOutcome[]): ClassifiedOutcomes { return { asks: outcomes.filter((o): o is ToolOutcome & { kind: 'ask' } => o.kind === 'ask'), @@ -258,6 +295,7 @@ export class Agent> = Record> = Record> = Record> = Record { + const accumulator = new TokenUsageAccumulator((modelKey) => this.modelProvider.getModelPricing(modelKey)) + try { + if (options.signal?.aborted) { + this.finishRun(agentRun, { + state: options.state, + newMessages: [], + finishReason: 'interrupted', + tokenUsage: accumulator.snapshot(), + contextWindowLimit: this.contextWindowLimit, + }) + return + } + + const performed = await this.performCompaction(options, (model, usage) => accumulator.add(model, usage)) + for (const message of performed.inferenceMessages) agentRun.push(message) + agentRun.pushEvent(performed.event) + this.finishRun(agentRun, { + state: performed.state, + newMessages: performed.inferenceMessages, + finishReason: 'complete', + tokenUsage: accumulator.snapshot(), + contextWindowLimit: this.contextWindowLimit, + }) + } catch (error) { + if (options.signal?.aborted && error instanceof Error && error.name === 'AbortError') { + this.finishRun(agentRun, { + state: options.state, + newMessages: [], + finishReason: 'interrupted', + tokenUsage: accumulator.snapshot(), + contextWindowLimit: this.contextWindowLimit, + }) + return + } + const agentError = + error instanceof AgentError + ? error + : new AgentError('unexpected_error', error instanceof Error ? error.message : String(error)) + this.finishRun(agentRun, { + state: options.state, + newMessages: [], + finishReason: 'error', + error: agentError, + tokenUsage: accumulator.snapshot(), + contextWindowLimit: this.contextWindowLimit, + }) + } + } + + private async performCompaction( + options: CompactOptions, + onUsage?: (model: string, usage: Omit) => void, + ): Promise { + const modelKey = getModelKey(this.model) + const modelLimits = this.modelProvider.getModelLimits(modelKey as ModelKey) + if (this.contextWindowLimit === undefined && modelLimits) this.contextWindowLimit = modelLimits.context + + const trigger = options.trigger ?? 'manual' + const policy = resolveCompactionPolicy(this.autoCompact) + const activeMessages = options.state.compaction ? options.state.messages.slice(1) : options.state.messages + const requiredToolCallIds = new Set((options.state.pendingToolCalls ?? []).map((pending) => pending.toolCallId)) + const plan = planCompaction(activeMessages, { + keepRecentTokens: options.keepRecentTokens ?? policy.keepRecentTokens, + requiredToolCallIds, + }) + if ( + plan === null || + (!plan.isSplitTurn && plan.conversationText.trim().length === 0) || + (plan.isSplitTurn && !plan.turnPrefixConversationText?.trim()) + ) { + throw new AgentError('unexpected_error', 'Compaction requires a coherent message prefix to summarize.') + } + + const providerOptions = this.resolveProviderOptions( + crypto.randomUUID(), + options.promptCacheKey ?? this.promptCacheKey, + ) + const summaryUsage: Omit = { + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + } + const inferenceMessages: ModelMessage[] = [] + const summarize = async (requestText: string, turnPrefix: boolean): Promise => { + const requestMessage = userMessage(requestText) + const result = streamText({ + model: this.model, + providerOptions, + messages: [requestMessage], + system: COMPACTION_SYSTEM_PROMPT, + maxOutputTokens: resolveCompactionMaxOutputTokens({ + reserveTokens: policy.reserveTokens, + modelOutputLimit: modelLimits?.output, + turnPrefix, + }), + abortSignal: options.signal ?? new AbortController().signal, + }) + + let streamError: unknown + for await (const part of result.fullStream) { + if (part.type === 'error') streamError = part.error + } + if (streamError) throw streamError + const [response, usage] = await Promise.all([result.response, result.usage]) + const callUsage = extractUsage(usage) + onUsage?.(modelKey, callUsage) + for (const key of [ + 'inputTokens', + 'outputTokens', + 'cacheReadTokens', + 'cacheWriteTokens', + 'reasoningTokens', + ] as const) { + summaryUsage[key] += callUsage[key] + } + const responseMessages = response.messages.filter((message) => message.role !== 'tool') + const summary = responseMessages + .filter((message) => message.role === 'assistant') + .flatMap((message) => + typeof message.content === 'string' + ? [message.content] + : message.content.filter((part) => part.type === 'text').map((part) => part.text), + ) + .join('\n') + .trim() + if (summary.length === 0) + throw new AgentError('unexpected_error', 'Compaction summarizer returned an empty summary.') + inferenceMessages.push(requestMessage, ...responseMessages) + return summary + } + + let historySummary = 'No prior history.' + if (plan.conversationText.trim() || options.state.compaction) { + historySummary = await summarize( + buildCompactionRequestText({ + conversationText: plan.conversationText, + ...(options.state.compaction ? { previousSummary: options.state.compaction.summary } : {}), + ...(policy.compactionPrompt ? { compactionPrompt: policy.compactionPrompt } : {}), + ...(policy.compactionUpdatePrompt ? { compactionUpdatePrompt: policy.compactionUpdatePrompt } : {}), + ...(options.additionalInstructions + ? { additionalInstructions: options.additionalInstructions } + : {}), + }), + false, + ) + } + const summary = plan.isSplitTurn + ? `${historySummary}\n\n---\n\n**Turn Context (split turn):**\n\n${await summarize( + buildTurnPrefixCompactionRequestText(plan.turnPrefixConversationText!), + true, + )}` + : historySummary + + const priorCheckpoint = options.state.compaction + const replacedMessageCount = plan.replacedMessages.length + (priorCheckpoint ? 1 : 0) + const retainedMessages = structuredClone(plan.retainedMessages) + const toolState = this.hooks?.compaction?.length + ? await runCompactionHooks(this.hooks.compaction, { + toolState: options.state.toolState ?? {}, + replacedMessages: plan.replacedMessages, + retainedMessages, + trigger, + }) + : options.state.toolState + const { contextWindowTokens: _staleContext, toolState: _oldToolState, ...stateWithoutContext } = options.state + const state: AgentState = { + ...stateWithoutContext, + messages: [compactionSummaryMessage(summary), ...retainedMessages], + ...(toolState !== undefined && Object.keys(toolState).length > 0 ? { toolState } : {}), + compaction: { + version: 1, + summary, + trigger, + replacedMessageCount, + retainedMessageCount: retainedMessages.length, + totalReplacedMessageCount: + (priorCheckpoint?.totalReplacedMessageCount ?? 0) + plan.replacedMessages.length, + ...(options.state.contextWindowTokens !== undefined + ? { priorContextWindowTokens: options.state.contextWindowTokens } + : {}), + }, + } + return { + state, + inferenceMessages, + event: { + type: 'compaction', + trigger, + ...(options.state.contextWindowTokens !== undefined + ? { priorContextWindowTokens: options.state.contextWindowTokens } + : {}), + replacedMessageCount, + retainedMessageCount: retainedMessages.length, + summaryUsage: { model: modelKey, usage: summaryUsage }, + }, + } + } + private async executeLoop(options: RunOptions, agentRun: AgentRun): Promise { // Hoist mutable state above try/catch so the error path can capture progress const allMessages: ModelMessage[] = [...options.state.messages] @@ -323,6 +572,7 @@ export class Agent> = Record = { ...(options.state.subAgents ?? {}) } const terminalChildren: TerminalChildMap = { ...(options.state.terminalChildren ?? {}) } + let compaction = options.state.compaction const sink = new MessageSink(allMessages, newMessages, agentRun) const promptCacheKey = options.promptCacheKey ?? this.promptCacheKey const providerOptions = this.resolveProviderOptions(crypto.randomUUID(), promptCacheKey) @@ -342,13 +592,6 @@ export class Agent> = Record { - if (event.type === 'tokenUsage' && event.agentId) { - accumulator.add(event.usage.model, event.usage.usage) - } - } - // Helper to build a state snapshot from current messages + optional pending + carried history const buildState = ( pendingToolCalls?: PendingToolCall[], @@ -364,6 +607,7 @@ export class Agent> = Record 0 ? { toolState } : {}), ...(Object.keys(effectiveSubAgents).length > 0 ? { subAgents: effectiveSubAgents } : {}), ...(Object.keys(terminalChildren).length > 0 ? { terminalChildren } : {}), + ...(compaction !== undefined ? { compaction } : {}), ...(contextWindowTokens > 0 ? { contextWindowTokens } : {}), } } @@ -371,6 +615,33 @@ export class Agent> = Record => { + const performed = await this.performCompaction( + { + state, + signal, + stream: options.stream, + promptCacheKey, + trigger, + ...(additionalInstructions ? { additionalInstructions } : {}), + }, + (model, usage) => accumulator.add(model, usage), + ) + allMessages.splice(0, allMessages.length, ...performed.state.messages) + for (const key of Object.keys(toolState)) delete toolState[key] + Object.assign(toolState, performed.state.toolState ?? {}) + compaction = performed.state.compaction + contextWindowTokens = 0 + for (const message of performed.inferenceMessages) { + newMessages.push(message) + agentRun.push(message) + } + agentRun.pushEvent(performed.event) + } const toolCtx: ExecuteToolCallContext = { tools: this.tools, @@ -383,11 +654,18 @@ export class Agent> = Record contextWindowTokens, getContextWindowLimit: () => this.contextWindowLimit, + onChildTokenUsage: (usage) => { + for (const [model, modelUsage] of Object.entries(usage.byModel)) { + const { estimatedCostUsd: _estimatedCostUsd, ...rawUsage } = modelUsage + accumulator.add(model, rawUsage) + } + }, createSubAgentFork: () => ({ agent: this.createForkAgent(), state: structuredClone({ messages: allMessages, ...((inputApprovalHistory?.length ?? 0) > 0 ? { approvalHistory: inputApprovalHistory } : {}), + ...(compaction !== undefined ? { compaction } : {}), }), }), createSubAgentForkAgent: () => this.createForkAgent(), @@ -411,6 +689,8 @@ export class Agent> = Record> = Record index !== manualCommand.messageIndex, + ) + await applyCompaction( + 'manual', + { ...buildState(), messages: messagesWithoutCommand }, + manualCommand.additionalInstructions, + ) + } else if ( + shouldCompactForThreshold({ + contextWindowTokens: contextWindowTokens > 0 ? contextWindowTokens : undefined, + policy: compactionPolicy, contextWindowLimit: this.contextWindowLimit, }) - if (hookResult.transformed) { - requestMessages = hookResult.messages - if (hookResult.persist) { - allMessages.length = 0 - allMessages.push(...hookResult.messages) + ) { + await applyCompaction('threshold', buildState()) + } + + let result: ExecutedModelStep + for (;;) { + // Compaction runs before request transforms, including on an overflow retry. + let requestMessages: ModelMessage[] = allMessages + if (this.hooks?.preRequest?.length) { + const hookResult = await runPreRequestHooks(this.hooks.preRequest, { + messages: allMessages, + contextWindowTokens, + contextWindowLimit: this.contextWindowLimit, + }) + if (hookResult.transformed) { + requestMessages = hookResult.messages + if (hookResult.persist) { + allMessages.length = 0 + allMessages.push(...hookResult.messages) + } } } - } - const result = await this.executeModelStep( - requestMessages, - signal, - options.stream, - stepIndex, - agentRun, - providerOptions, - ) + try { + result = await this.executeModelStep( + requestMessages, + signal, + options.stream, + stepIndex, + agentRun, + providerOptions, + ) + break + } catch (error) { + if (!compactionPolicy.enabled || overflowRecoveryUsed || !isContextOverflowError(error)) + throw error + overflowRecoveryUsed = true + await applyCompaction('overflow', buildState()) + } + } // Only push non-tool messages from the AI SDK response. // The agent manages tool result creation itself; the AI SDK may diff --git a/packages/agentlayer-core/src/compaction/engine.ts b/packages/agentlayer-core/src/compaction/engine.ts new file mode 100644 index 0000000..337374f --- /dev/null +++ b/packages/agentlayer-core/src/compaction/engine.ts @@ -0,0 +1,303 @@ +import type { ModelMessage } from 'ai' +import type { ResolvedCompactionPolicy } from './policy' +import { resolveCompactionThreshold } from './policy' + +export const SERIALIZED_TOOL_RESULT_MAX_CHARS = 2_000 +const FILE_PART_ESTIMATE_CHARS = 4_800 + +type MessagePart = Record & { type: string } + +export interface CompactCommand { + messageIndex: number + additionalInstructions?: string +} + +function safeStringify(value: unknown): string { + try { + return JSON.stringify(value) ?? String(value) + } catch { + return String(value) + } +} + +function contentParts(content: unknown): MessagePart[] { + if (typeof content === 'string') return [{ type: 'text', text: content }] + if (!Array.isArray(content)) return [] + return content.filter( + (part): part is MessagePart => typeof part === 'object' && part !== null && typeof part.type === 'string', + ) +} + +function textContent(content: unknown): string | undefined { + if (typeof content === 'string') return content + if (!Array.isArray(content)) return undefined + const parts = contentParts(content) + if (parts.some((part) => part.type !== 'text')) return undefined + return parts.map(partText).join('') +} + +/** Parse the final user message when it begins with a standalone `/compact` command. */ +export function parseCompactCommand(messages: ReadonlyArray): CompactCommand | undefined { + const messageIndex = messages.length - 1 + const message = messages[messageIndex] + if (!message || message.role !== 'user') return undefined + const text = textContent(message.content) + if (text === undefined) return undefined + const match = /^\s*\/compact(?:\s+([\s\S]*?))?\s*$/.exec(text) + if (!match) return undefined + const additionalInstructions = match[1]?.trim() + return { + messageIndex, + ...(additionalInstructions ? { additionalInstructions } : {}), + } +} + +/** Only a positive estimate recorded after the latest checkpoint can trigger proactive compaction. */ +export function shouldCompactForThreshold(input: { + contextWindowTokens?: number + policy: ResolvedCompactionPolicy + contextWindowLimit?: number +}): boolean { + if (!input.policy.enabled || input.contextWindowTokens === undefined || input.contextWindowTokens <= 0) return false + const threshold = resolveCompactionThreshold(input.policy, input.contextWindowLimit) + return threshold !== undefined && input.contextWindowTokens >= threshold +} + +function partText(part: MessagePart): string { + if (typeof part.text === 'string') return part.text + if (typeof part.value === 'string') return part.value + return safeStringify(part) +} + +function toolOutput(part: MessagePart): unknown { + const output = part.output + if (typeof output === 'object' && output !== null && 'value' in output) { + return (output as { value: unknown }).value + } + return output +} + +function serializeValue(value: unknown): string { + return typeof value === 'string' ? value : safeStringify(value) +} + +/** Estimate a provider-neutral message footprint with Fold's chars/4 heuristic. */ +export function estimateMessageTokens(message: ModelMessage): number { + const chars = contentParts(message.content).reduce((total, part) => { + switch (part.type) { + case 'text': + case 'reasoning': + return total + partText(part).length + case 'file': + case 'image': + return total + FILE_PART_ESTIMATE_CHARS + case 'tool-call': + return total + String(part.toolName ?? '').length + safeStringify(part.input).length + case 'tool-result': + return total + serializeValue(toolOutput(part)).length + default: + return total + safeStringify(part).length + } + }, 0) + return Math.max(1, Math.ceil(chars / 4)) +} + +function toolCallIds(message: ModelMessage): string[] { + if (message.role !== 'assistant') return [] + return contentParts(message.content) + .filter((part) => part.type === 'tool-call' && typeof part.toolCallId === 'string') + .map((part) => part.toolCallId as string) +} + +function toolResultIds(message: ModelMessage): string[] { + if (message.role !== 'tool') return [] + return contentParts(message.content) + .filter((part) => part.type === 'tool-result' && typeof part.toolCallId === 'string') + .map((part) => part.toolCallId as string) +} + +/** Ensure every retained tool result has its retained assistant tool call. */ +export function hasValidToolCallResultPairs(messages: ReadonlyArray): boolean { + const calls = new Set(messages.flatMap(toolCallIds)) + return messages.flatMap(toolResultIds).every((id) => calls.has(id)) +} + +function containsRequiredToolCalls( + messages: ReadonlyArray, + requiredToolCallIds: ReadonlySet, +): boolean { + if (requiredToolCallIds.size === 0) return true + const calls = new Set(messages.flatMap(toolCallIds)) + return [...requiredToolCallIds].every((id) => calls.has(id)) +} + +function isValidCutRole(message: ModelMessage): boolean { + return message.role === 'user' || message.role === 'assistant' +} + +export interface FindCompactionCutOptions { + keepRecentTokens: number + requiredToolCallIds?: ReadonlySet +} + +/** + * Return the first retained message index. The cut keeps complete tool-call/result groups and any + * assistant calls referenced by pending state. + */ +export function findCompactionCut(messages: ReadonlyArray, options: FindCompactionCutOptions): number { + if (messages.length < 2) return 0 + + let accumulated = 0 + let boundary = -1 + for (let index = messages.length - 1; index >= 0; index--) { + accumulated += estimateMessageTokens(messages[index]!) + if (accumulated >= Math.max(1, options.keepRecentTokens)) { + boundary = index + break + } + } + + if (boundary <= 0) return 0 + const required = options.requiredToolCallIds ?? new Set() + const candidates: number[] = [] + for (let index = boundary; index < messages.length; index++) candidates.push(index) + for (let index = boundary - 1; index > 0; index--) candidates.push(index) + + for (const index of candidates) { + if (!isValidCutRole(messages[index]!)) continue + const retained = messages.slice(index) + if (hasValidToolCallResultPairs(retained) && containsRequiredToolCalls(retained, required)) return index + } + return 0 +} + +function truncateToolResult(serialized: string): string { + if (serialized.length <= SERIALIZED_TOOL_RESULT_MAX_CHARS) return serialized + return `${serialized.slice(0, SERIALIZED_TOOL_RESULT_MAX_CHARS)}[... ${serialized.length - SERIALIZED_TOOL_RESULT_MAX_CHARS} more characters truncated]` +} + +function serializeUserContent(content: unknown): string { + return contentParts(content) + .map((part) => { + if (part.type === 'text') return partText(part) + if (part.type === 'file' || part.type === 'image') return '[attached file]' + return safeStringify(part) + }) + .join('\n') +} + +/** Flatten replaced native messages into a provider-neutral bounded transcript. */ +export function serializeConversation(messages: ReadonlyArray): string { + const lines: string[] = [] + for (const message of messages) { + if (message.role === 'user') { + lines.push(`[User]: ${serializeUserContent(message.content)}`) + continue + } + if (message.role === 'system') { + lines.push(`[System note]: ${serializeUserContent(message.content)}`) + continue + } + if (message.role === 'assistant') { + const parts = contentParts(message.content) + for (const part of parts.filter((candidate) => candidate.type === 'reasoning')) { + lines.push(`[Assistant thinking]: ${partText(part)}`) + } + const text = parts.filter((candidate) => candidate.type === 'text').map(partText) + if (text.length > 0) lines.push(`[Assistant]: ${text.join('\n')}`) + const calls = parts.filter((candidate) => candidate.type === 'tool-call') + if (calls.length > 0) { + lines.push( + `[Assistant tool calls]: ${calls + .map((part) => `${String(part.toolName ?? 'tool')}(${safeStringify(part.input)})`) + .join('; ')}`, + ) + } + continue + } + if (message.role === 'tool') { + for (const part of contentParts(message.content)) { + if (part.type !== 'tool-result') continue + lines.push(`[Tool result]: ${truncateToolResult(serializeValue(toolOutput(part)))}`) + } + } + } + return lines.join('\n') +} + +export interface CompactionPlan { + cutIndex: number + /** Older completed history summarized with the normal initial/update prompt. */ + historyMessages: ModelMessage[] + /** Discarded prefix of a current oversized turn, summarized separately. */ + turnPrefixMessages: ModelMessage[] + isSplitTurn: boolean + replacedMessages: ModelMessage[] + retainedMessages: ModelMessage[] + conversationText: string + turnPrefixConversationText?: string +} + +function findTurnStart(messages: ReadonlyArray, cutIndex: number): number { + for (let index = cutIndex; index >= 0; index--) { + if (messages[index]?.role === 'user') return index + } + return -1 +} + +/** Build a pure summary-prefix/native-tail plan, or return null when no coherent prefix exists. */ +export function planCompaction( + messages: ReadonlyArray, + options: FindCompactionCutOptions, +): CompactionPlan | null { + const cutIndex = findCompactionCut(messages, options) + if (cutIndex <= 0) return null + const turnStartIndex = messages[cutIndex]?.role === 'user' ? -1 : findTurnStart(messages, cutIndex) + const isSplitTurn = turnStartIndex >= 0 && turnStartIndex < cutIndex + const historyMessages = messages.slice(0, isSplitTurn ? turnStartIndex : cutIndex) + const turnPrefixMessages = isSplitTurn ? messages.slice(turnStartIndex, cutIndex) : [] + const replacedMessages = messages.slice(0, cutIndex) + const retainedMessages = messages.slice(cutIndex) + return { + cutIndex, + historyMessages, + turnPrefixMessages, + isSplitTurn, + replacedMessages, + retainedMessages, + conversationText: serializeConversation(historyMessages), + ...(isSplitTurn ? { turnPrefixConversationText: serializeConversation(turnPrefixMessages) } : {}), + } +} + +export const CONTEXT_OVERFLOW_PATTERNS: ReadonlyArray = [ + /context[_ ]length[_ ]exceeded/i, + /model_context_window_exceeded/i, + /prompt is too long/i, + /input is too long/i, + /exceeds the context window/i, + /maximum context length/i, + /maximum prompt length/i, + /context window exceeds/i, + /exceeds the available context/i, + /reduce the length of the messages/i, + /request entity too large/i, + /too large for model/i, +] + +const NON_OVERFLOW_PATTERNS: ReadonlyArray = [/rate.?limit/i, /too many requests/i, /quota/i] + +function errorText(error: unknown): string { + if (error instanceof Error) return error.message + if (typeof error === 'string') return error + return safeStringify(error) +} + +/** Recognize provider context overflow while excluding rate-limit and quota failures. */ +export function isContextOverflowError(error: unknown): boolean { + const message = errorText(error) + return ( + !NON_OVERFLOW_PATTERNS.some((pattern) => pattern.test(message)) && + CONTEXT_OVERFLOW_PATTERNS.some((pattern) => pattern.test(message)) + ) +} diff --git a/packages/agentlayer-core/src/compaction/index.ts b/packages/agentlayer-core/src/compaction/index.ts new file mode 100644 index 0000000..764e904 --- /dev/null +++ b/packages/agentlayer-core/src/compaction/index.ts @@ -0,0 +1,3 @@ +export * from './engine' +export * from './policy' +export * from './prompts' diff --git a/packages/agentlayer-core/src/compaction/policy.ts b/packages/agentlayer-core/src/compaction/policy.ts new file mode 100644 index 0000000..75cc386 --- /dev/null +++ b/packages/agentlayer-core/src/compaction/policy.ts @@ -0,0 +1,93 @@ +/** Why a compaction checkpoint was created. */ +export type CompactionTrigger = 'manual' | 'threshold' | 'overflow' + +/** Provider-neutral compaction configuration for an Agent. */ +export type AutoCompactConfig = + | { enabled: false } + | { + enabled?: true + /** Override the automatic compaction threshold. */ + thresholdTokens?: number + /** Override the context window used for compaction budget calculations. */ + contextWindow?: number + /** Tokens kept free below the output allowance. */ + reserveTokens?: number + /** Recent native-message context retained after a checkpoint. */ + keepRecentTokens?: number + /** Replace the default instruction for an initial checkpoint while retaining request framing. */ + compactionPrompt?: string + /** Replace the default instruction for an incremental checkpoint while retaining request framing. */ + compactionUpdatePrompt?: string + } + +export interface ResolvedCompactionPolicy { + enabled: boolean + thresholdTokens?: number + contextWindow?: number + reserveTokens: number + keepRecentTokens: number + compactionPrompt?: string + compactionUpdatePrompt?: string +} + +export const DEFAULT_COMPACTION_RESERVE_TOKENS = 16_384 +export const DEFAULT_COMPACTION_KEEP_RECENT_TOKENS = 20_000 +export const MAX_COMPACTION_OUTPUT_TOKENS = 32_000 +export const COMPACTION_OUTPUT_BUDGET_RATIO = 0.8 +export const TURN_PREFIX_OUTPUT_BUDGET_RATIO = 0.5 + +/** Resolve the provider-neutral output cap for a history or split-turn-prefix summary. */ +export function resolveCompactionMaxOutputTokens(input: { + reserveTokens?: number + modelOutputLimit?: number + turnPrefix?: boolean +}): number { + const outputBudget = Math.max(1, Math.floor(input.reserveTokens ?? DEFAULT_COMPACTION_RESERVE_TOKENS)) + const ratio = input.turnPrefix ? TURN_PREFIX_OUTPUT_BUDGET_RATIO : COMPACTION_OUTPUT_BUDGET_RATIO + const reservedCap = Math.max(1, Math.floor(outputBudget * ratio)) + return input.modelOutputLimit === undefined + ? reservedCap + : Math.max(1, Math.min(reservedCap, Math.floor(input.modelOutputLimit))) +} + +/** Resolve omitted configuration to the default-enabled policy. */ +export function resolveCompactionPolicy(config?: AutoCompactConfig): ResolvedCompactionPolicy { + if (config?.enabled === false) { + return { + enabled: false, + reserveTokens: DEFAULT_COMPACTION_RESERVE_TOKENS, + keepRecentTokens: DEFAULT_COMPACTION_KEEP_RECENT_TOKENS, + } + } + + return { + enabled: true, + ...(config?.thresholdTokens !== undefined ? { thresholdTokens: config.thresholdTokens } : {}), + ...(config?.contextWindow !== undefined ? { contextWindow: config.contextWindow } : {}), + reserveTokens: config?.reserveTokens ?? DEFAULT_COMPACTION_RESERVE_TOKENS, + keepRecentTokens: config?.keepRecentTokens ?? DEFAULT_COMPACTION_KEEP_RECENT_TOKENS, + ...(config?.compactionPrompt !== undefined ? { compactionPrompt: config.compactionPrompt } : {}), + ...(config?.compactionUpdatePrompt !== undefined + ? { compactionUpdatePrompt: config.compactionUpdatePrompt } + : {}), + } +} + +/** Fold-compatible usable context budget before automatic compaction. */ +export function compactionUsableTokens(input: { contextWindow: number; reserveTokens?: number }): number { + const contextWindow = Math.max(1, Math.floor(input.contextWindow)) + const outputBudget = Math.min(MAX_COMPACTION_OUTPUT_TOKENS, Math.floor(contextWindow / 4)) + const reserve = Math.min(input.reserveTokens ?? DEFAULT_COMPACTION_RESERVE_TOKENS, Math.floor(contextWindow / 8)) + return Math.max(1, contextWindow - outputBudget - reserve) +} + +/** Resolve an explicit threshold or derive one from the effective context window. */ +export function resolveCompactionThreshold( + policy: ResolvedCompactionPolicy, + contextWindowLimit?: number, +): number | undefined { + if (policy.thresholdTokens !== undefined) return Math.max(1, Math.floor(policy.thresholdTokens)) + const contextWindow = policy.contextWindow ?? contextWindowLimit + if (contextWindow === undefined) return undefined + return compactionUsableTokens({ contextWindow, reserveTokens: policy.reserveTokens }) +} diff --git a/packages/agentlayer-core/src/compaction/prompts.ts b/packages/agentlayer-core/src/compaction/prompts.ts new file mode 100644 index 0000000..a7a17c0 --- /dev/null +++ b/packages/agentlayer-core/src/compaction/prompts.ts @@ -0,0 +1,134 @@ +import type { ModelMessage, UserModelMessage } from 'ai' + +export const COMPACTION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified. + +Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.` + +export const DEFAULT_COMPACTION_PROMPT = `The messages above are a conversation to summarize. Create a structured context checkpoint summary that another LLM will use to continue the work. + +Use this EXACT format: + +## Goal +[What is the user trying to accomplish? Can be multiple items if the session covers different tasks.] + +## Constraints & Preferences +- [Any constraints, preferences, or requirements mentioned by user] +- [Or "(none)" if none were mentioned] + +## Progress +### Done +- [x] [Completed tasks/changes] + +### In Progress +- [ ] [Current work] + +### Blocked +- [Issues preventing progress, if any] + +## Key Decisions +- **[Decision]**: [Brief rationale] + +## Next Steps +1. [Ordered list of what should happen next] + +## Critical Context +- [Any data, examples, or references needed to continue] +- [Or "(none)" if not applicable] + +Keep each section concise. Preserve exact file paths, function names, and error messages.` + +export const DEFAULT_COMPACTION_UPDATE_PROMPT = `The messages above are NEW conversation messages to incorporate into the existing summary provided in tags. + +Update the existing structured summary with new information. RULES: +- PRESERVE all existing information from the previous summary +- ADD new progress, decisions, and context from the new messages +- UPDATE the Progress section: move items from "In Progress" to "Done" when completed +- UPDATE "Next Steps" based on what was accomplished +- PRESERVE exact file paths, function names, and error messages +- If something is no longer relevant, you may remove it + +Use this EXACT format: + +## Goal +[Preserve existing goals, add new ones if the task expanded] + +## Constraints & Preferences +- [Preserve existing, add new ones discovered] + +## Progress +### Done +- [x] [Include previously done items AND newly completed items] + +### In Progress +- [ ] [Current work - update based on progress] + +### Blocked +- [Current blockers - remove if resolved] + +## Key Decisions +- **[Decision]**: [Brief rationale] (preserve all previous, add new) + +## Next Steps +1. [Update based on current state] + +## Critical Context +- [Preserve important context, add new if needed] + +Keep each section concise. Preserve exact file paths, function names, and error messages.` + +export const TURN_PREFIX_COMPACTION_PROMPT = `This is the PREFIX of a turn that was too large to keep. The SUFFIX (recent work) is retained. + +Summarize the prefix to provide context for the retained suffix: + +## Original Request +[What did the user ask for in this turn?] + +## Early Progress +- [Key decisions and work done in the prefix] + +## Context for Suffix +- [Information needed to understand the retained recent work] + +Be concise. Focus on what's needed to understand the kept suffix.` + +export interface CompactionRequestTextInput { + conversationText: string + previousSummary?: string + /** Replaces the default initial-checkpoint instruction. */ + compactionPrompt?: string + /** Replaces the default incremental-checkpoint instruction. */ + compactionUpdatePrompt?: string + additionalInstructions?: string +} + +export function compactionInstruction(input: CompactionRequestTextInput): string { + return input.previousSummary === undefined + ? (input.compactionPrompt ?? DEFAULT_COMPACTION_PROMPT) + : (input.compactionUpdatePrompt ?? DEFAULT_COMPACTION_UPDATE_PROMPT) +} + +/** Build the fixed conversation/previous-summary/instruction request framing. */ +export function buildCompactionRequestText(input: CompactionRequestTextInput): string { + const previousBlock = + input.previousSummary === undefined + ? '' + : `\n${input.previousSummary}\n\n\n` + const guidance = input.additionalInstructions?.trim() + const guidanceBlock = guidance ? `\n\nAdditional user guidance for this summary:\n${guidance}` : '' + return `\n${input.conversationText}\n\n\n${previousBlock}${compactionInstruction(input)}${guidanceBlock}` +} + +/** Build the concise request used only for a discarded prefix of an oversized current turn. */ +export function buildTurnPrefixCompactionRequestText(conversationText: string): string { + return `\n${conversationText}\n\n\n${TURN_PREFIX_COMPACTION_PROMPT}` +} + +/** Canonical provider-neutral message that replaces a compacted prefix in AgentState. */ +export function compactionSummaryMessage(summary: string): UserModelMessage { + return { role: 'user', content: `\n${summary}\n` } +} + +/** Identify the canonical summary message associated with checkpoint metadata. */ +export function isCompactionSummaryMessage(message: ModelMessage, summary: string): boolean { + return message.role === 'user' && message.content === compactionSummaryMessage(summary).content +} diff --git a/packages/agentlayer-core/src/execute-tool-call.ts b/packages/agentlayer-core/src/execute-tool-call.ts index 26800e2..78809e7 100644 --- a/packages/agentlayer-core/src/execute-tool-call.ts +++ b/packages/agentlayer-core/src/execute-tool-call.ts @@ -1,10 +1,12 @@ import type { ModelMessage } from 'ai' +import type { RunResult } from './agent' import type { AgentRun } from './agent-run' import type { SubAgentPauseResult, SubAgentResult, SubAgentRunHandle, Tool, ToolContext } from './define-tool' import type { HookStopResult, StopOptions } from './hooks' import { type AgentLayerToolOutput, buildToolResultMessage, isToolResultOutput } from './messages' import { sanitizeTextForModelState, sanitizeToolOutputForModelState } from './sanitize-text' import type { AgentState, TerminalChildMap } from './state' +import type { TokenUsage } from './token-usage' export interface ToolCallRef { toolCallId: string @@ -30,6 +32,8 @@ export interface ExecuteToolCallContext { getContextWindowTokens?: () => number /** Returns the context window token limit for the current model. */ getContextWindowLimit?: () => number | undefined + /** Aggregate one completed child run without relying on forwarded event accounting. */ + onChildTokenUsage?: (usage: TokenUsage) => void /** Capture an isolated caller state and equivalent runtime for a fork child. */ createSubAgentFork?: (toolCallId: string) => { agent: import('./agent').Agent; state: AgentState } createSubAgentForkAgent?: () => import('./agent').Agent @@ -141,7 +145,8 @@ export async function executeToolCall(tc: ToolCallRef, ctx: ExecuteToolCallConte })() // Wait for the child result - const result = await childRun.result + const result = (await childRun.result) as RunResult + ctx.onChildTokenUsage?.(result.tokenUsage) // Wait for forwarding to finish await forwardingPromise diff --git a/packages/agentlayer-core/src/hooks/compaction.ts b/packages/agentlayer-core/src/hooks/compaction.ts new file mode 100644 index 0000000..7c93601 --- /dev/null +++ b/packages/agentlayer-core/src/hooks/compaction.ts @@ -0,0 +1,26 @@ +import type { ModelMessage } from 'ai' +import type { CompactionTrigger } from '../compaction' + +export interface CompactionHookContext { + toolState: Readonly> + replacedMessages: ReadonlyArray + retainedMessages: ReadonlyArray + trigger: CompactionTrigger +} + +export type CompactionHook = ( + context: CompactionHookContext, +) => Record | undefined | Promise | undefined> + +/** Run compaction hooks in order, passing each hook the previous hook's tool-state result. */ +export async function runCompactionHooks( + hooks: ReadonlyArray, + context: CompactionHookContext, +): Promise> { + let toolState = { ...context.toolState } + for (const hook of hooks) { + const next = await hook({ ...context, toolState }) + if (next !== undefined) toolState = next + } + return toolState +} diff --git a/packages/agentlayer-core/src/hooks/index.ts b/packages/agentlayer-core/src/hooks/index.ts index bbb692c..d77292b 100644 --- a/packages/agentlayer-core/src/hooks/index.ts +++ b/packages/agentlayer-core/src/hooks/index.ts @@ -1,4 +1,5 @@ export * from './approval' +export * from './compaction' export * from './deduplicate-reads' export * from './output-truncation' export * from './post-tool-use' diff --git a/packages/agentlayer-core/src/index.ts b/packages/agentlayer-core/src/index.ts index 10fcf27..20d2b7b 100644 --- a/packages/agentlayer-core/src/index.ts +++ b/packages/agentlayer-core/src/index.ts @@ -1,6 +1,7 @@ export { Agent, type AgentConfig, + type CompactOptions, type FinishReason, type ProviderOptions, type ProviderOptionsFactory, @@ -8,6 +9,7 @@ export { type RunResult, } from './agent' export { type AgentEvent, AgentRun } from './agent-run' +export * from './compaction' export { defineTool, defineToolInterface, @@ -100,6 +102,7 @@ export { type AgentState, type ApprovalDecision, type ApprovalHistoryEntry, + type CompactionCheckpoint, getAgentState, getAllPendingApprovals, sanitizeStateForPersistence, diff --git a/packages/agentlayer-core/src/state.ts b/packages/agentlayer-core/src/state.ts index f31061a..f0c9858 100644 --- a/packages/agentlayer-core/src/state.ts +++ b/packages/agentlayer-core/src/state.ts @@ -1,4 +1,5 @@ import type { ModelMessage } from 'ai' +import type { CompactionTrigger } from './compaction' import type { ApprovalRequest, PendingToolCall } from './hooks' import { buildToolResultMessage } from './messages' @@ -57,6 +58,17 @@ export interface TerminalChildRecord { export type TerminalChildMap = Record +/** JSON-safe metadata for the canonical summary currently replacing an older message prefix. */ +export interface CompactionCheckpoint { + version: 1 + summary: string + trigger: CompactionTrigger + replacedMessageCount: number + retainedMessageCount: number + totalReplacedMessageCount: number + priorContextWindowTokens?: number +} + // ── AgentState ──────────────────────────────────────────────────────────────── /** @@ -82,6 +94,8 @@ export interface AgentState { subAgents?: Record /** Terminal child states that can be continued by stable agent ID. */ terminalChildren?: TerminalChildMap + /** Metadata for the synthetic summary message in the active model view. */ + compaction?: CompactionCheckpoint /** Estimated tokens in context window after the most recent streamText call (input + output). */ contextWindowTokens?: number } @@ -400,6 +414,7 @@ export function withApprovals(state: AgentState, decisions: ApprovalDecision[]): ...(state.toolState !== undefined ? { toolState: state.toolState } : {}), ...(newSubAgents !== undefined ? { subAgents: newSubAgents } : {}), ...(newTerminalChildren !== undefined ? { terminalChildren: newTerminalChildren } : {}), + ...(state.compaction !== undefined ? { compaction: state.compaction } : {}), ...(state.contextWindowTokens !== undefined ? { contextWindowTokens: state.contextWindowTokens } : {}), } } diff --git a/packages/agentlayer-core/src/tools/subagent-fork.ts b/packages/agentlayer-core/src/tools/subagent-fork.ts index ea32292..0c0f5b3 100644 --- a/packages/agentlayer-core/src/tools/subagent-fork.ts +++ b/packages/agentlayer-core/src/tools/subagent-fork.ts @@ -1,4 +1,5 @@ import type { ModelMessage } from 'ai' +import { isCompactionSummaryMessage } from '../compaction' import type { AgentState, ApprovalHistoryEntry } from '../state' export type ForkTurns = 'all' | 'none' | number @@ -74,9 +75,15 @@ export function createForkState( const approvalHistory = callerState.approvalHistory ? clone(callerState.approvalHistory) : undefined + const compaction = + callerState.compaction && + messages.some((message) => isCompactionSummaryMessage(message, callerState.compaction!.summary)) + ? clone(callerState.compaction) + : undefined return { messages: [...messages, { role: 'user', content: prompt }], ...(approvalHistory?.length ? { approvalHistory } : {}), + ...(compaction ? { compaction } : {}), } } diff --git a/packages/agentlayer-core/test/compaction-engine.test.ts b/packages/agentlayer-core/test/compaction-engine.test.ts new file mode 100644 index 0000000..333672d --- /dev/null +++ b/packages/agentlayer-core/test/compaction-engine.test.ts @@ -0,0 +1,143 @@ +import { describe, expect, test } from 'bun:test' +import type { ModelMessage } from 'ai' +import { + compactionUsableTokens, + estimateMessageTokens, + findCompactionCut, + hasValidToolCallResultPairs, + isContextOverflowError, + planCompaction, + resolveCompactionMaxOutputTokens, + resolveCompactionPolicy, + resolveCompactionThreshold, + serializeConversation, +} from '../src/compaction' +import { assistantMessage, toolCall, toolResult, userMessage } from '../src/messages' + +describe('compaction policy', () => { + test('uses Fold-compatible budget arithmetic', () => { + expect(compactionUsableTokens({ contextWindow: 200_000 })).toBe(151_616) + expect(compactionUsableTokens({ contextWindow: 100, reserveTokens: 50 })).toBe(63) + expect(compactionUsableTokens({ contextWindow: 1 })).toBe(1) + }) + + test('caps history and turn-prefix summary output against reserved and model budgets', () => { + expect(resolveCompactionMaxOutputTokens({ reserveTokens: 20_000 })).toBe(16_000) + expect(resolveCompactionMaxOutputTokens({ reserveTokens: 20_000, turnPrefix: true })).toBe(10_000) + expect(resolveCompactionMaxOutputTokens({ reserveTokens: 20_000, modelOutputLimit: 3_000 })).toBe(3_000) + expect(resolveCompactionMaxOutputTokens({})).toBe(13_107) + }) + + test('is enabled by default and resolves disablement and overrides', () => { + const defaults = resolveCompactionPolicy() + expect(defaults).toMatchObject({ enabled: true, reserveTokens: 16_384, keepRecentTokens: 20_000 }) + expect(resolveCompactionThreshold(defaults, 200_000)).toBe(151_616) + + const disabled = resolveCompactionPolicy({ enabled: false }) + expect(disabled.enabled).toBe(false) + + const overridden = resolveCompactionPolicy({ + thresholdTokens: 42, + contextWindow: 1_000, + reserveTokens: 10, + keepRecentTokens: 25, + compactionPrompt: 'Initial override.', + compactionUpdatePrompt: 'Incremental override.', + }) + expect(resolveCompactionThreshold(overridden, 200_000)).toBe(42) + expect(overridden).toMatchObject({ + contextWindow: 1_000, + reserveTokens: 10, + keepRecentTokens: 25, + compactionPrompt: 'Initial override.', + compactionUpdatePrompt: 'Incremental override.', + }) + }) +}) + +describe('compaction engine', () => { + test('estimates text with the chars/4 heuristic', () => { + expect(estimateMessageTokens(userMessage('12345678'))).toBe(2) + expect(estimateMessageTokens(userMessage(''))).toBe(1) + }) + + test('selects a bounded native tail at a coherent boundary', () => { + const messages = [ + userMessage('old user'), + assistantMessage('old assistant'), + userMessage('new user'), + assistantMessage('new assistant'), + ] + const cut = findCompactionCut(messages, { keepRecentTokens: 6 }) + expect(cut).toBe(2) + + const plan = planCompaction(messages, { keepRecentTokens: 6 }) + expect(plan?.replacedMessages).toEqual(messages.slice(0, 2)) + expect(plan?.retainedMessages).toEqual(messages.slice(2)) + expect(plan?.conversationText).toContain('[User]: old user') + expect(plan?.conversationText).toContain('[Assistant]: old assistant') + }) + + test('keeps tool results with matching calls and pending calls in the retained region', () => { + const call = toolCall({ toolCallId: 'call-1', toolName: 'read', input: { path: 'a.ts' } }) + const result = toolResult({ toolCallId: 'call-1', toolName: 'read', output: 'contents' }) + const messages: ModelMessage[] = [ + userMessage('old'), + assistantMessage('old answer'), + userMessage('use a tool'), + call, + result, + ] + + const cut = findCompactionCut(messages, { + keepRecentTokens: 1, + requiredToolCallIds: new Set(['call-1']), + }) + expect(cut).toBe(3) + expect(hasValidToolCallResultPairs(messages.slice(cut))).toBe(true) + expect(hasValidToolCallResultPairs([result])).toBe(false) + }) + + test('splits an oversized current turn only at message boundaries', () => { + const call = toolCall({ toolCallId: 'call-1', toolName: 'read', input: { path: 'large.ts' } }) + const result = toolResult({ toolCallId: 'call-1', toolName: 'read', output: 'x'.repeat(400) }) + const messages: ModelMessage[] = [ + userMessage('older completed request'), + assistantMessage('older completed answer'), + userMessage('oversized current request'), + assistantMessage('early progress'), + call, + result, + assistantMessage('ok'), + ] + const plan = planCompaction(messages, { keepRecentTokens: 2 }) + + expect(plan?.isSplitTurn).toBe(true) + expect(plan?.historyMessages).toEqual(messages.slice(0, 2)) + expect(plan?.turnPrefixMessages).toEqual(messages.slice(2, 6)) + expect(plan?.retainedMessages).toEqual(messages.slice(6)) + expect(plan?.replacedMessages).toEqual(messages.slice(0, 6)) + expect(hasValidToolCallResultPairs(plan!.retainedMessages)).toBe(true) + expect(plan?.turnPrefixConversationText).toContain('[Tool result]') + }) + + test('bounds serialized tool output without dropping transcript roles', () => { + const transcript = serializeConversation([ + userMessage('inspect'), + toolCall({ toolCallId: 'call-1', toolName: 'bash', input: { command: 'run' } }), + toolResult({ toolCallId: 'call-1', toolName: 'bash', output: 'x'.repeat(3_000) }), + ]) + expect(transcript).toContain('[User]: inspect') + expect(transcript).toContain('[Assistant tool calls]: bash(') + expect(transcript).toContain('1000 more characters truncated') + expect(transcript.length).toBeLessThan(2_300) + }) + + test('recognizes provider overflow errors and excludes rate limits and quotas', () => { + expect(isContextOverflowError(new Error('context_length_exceeded'))).toBe(true) + expect(isContextOverflowError('Prompt is too long for this model')).toBe(true) + expect(isContextOverflowError('rate limit: request exceeds the context window')).toBe(false) + expect(isContextOverflowError('quota exceeded')).toBe(false) + expect(isContextOverflowError('connection reset')).toBe(false) + }) +}) diff --git a/packages/agentlayer-core/test/compaction-prompts.test.ts b/packages/agentlayer-core/test/compaction-prompts.test.ts new file mode 100644 index 0000000..95d2bd4 --- /dev/null +++ b/packages/agentlayer-core/test/compaction-prompts.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, test } from 'bun:test' +import { + buildCompactionRequestText, + buildTurnPrefixCompactionRequestText, + DEFAULT_COMPACTION_PROMPT, + DEFAULT_COMPACTION_UPDATE_PROMPT, + TURN_PREFIX_COMPACTION_PROMPT, +} from '../src/compaction' + +describe('compaction prompts', () => { + test('frames an initial summary request with every required section', () => { + const request = buildCompactionRequestText({ conversationText: '[User]: implement it' }) + expect(request.startsWith('\n[User]: implement it\n\n\n')).toBe(true) + expect(request).toContain(DEFAULT_COMPACTION_PROMPT) + expect(request).not.toContain('') + for (const heading of [ + '## Goal', + '## Constraints & Preferences', + '## Progress', + '## Key Decisions', + '## Next Steps', + '## Critical Context', + ]) { + expect(request).toContain(heading) + } + expect(request).not.toContain('## Commands & Verification') + expect(request).not.toContain('commands or workflows') + expect(request).not.toContain('verification workflow') + }) + + test('frames incremental updates without command or verification preservation instructions', () => { + const request = buildCompactionRequestText({ + conversationText: '[Assistant]: implementation is complete', + previousSummary: '## Progress\n### In Progress\n- [ ] Implement the change', + }) + expect(request).toContain( + '\n## Progress\n### In Progress\n- [ ] Implement the change\n', + ) + expect(request).toContain(DEFAULT_COMPACTION_UPDATE_PROMPT) + expect(request).not.toContain('## Commands & Verification') + expect(request).not.toContain('PRESERVE still-valid commands') + expect(request).not.toContain('REMOVE stale commands') + expect(request).not.toContain('verification workflows') + }) + + test('appends manual guidance after rather than replacing the fixed template', () => { + const request = buildCompactionRequestText({ + conversationText: '[User]: continue', + additionalInstructions: 'Focus on unresolved blockers.', + }) + expect(request).toContain(DEFAULT_COMPACTION_PROMPT) + expect(request).toEndWith('Additional user guidance for this summary:\nFocus on unresolved blockers.') + expect(request.indexOf('Use this EXACT format:')).toBeLessThan(request.indexOf('Additional user guidance')) + }) + + test('initial and incremental overrides replace only their corresponding default templates', () => { + const initial = buildCompactionRequestText({ + conversationText: 'initial work', + compactionPrompt: 'Use the initial company checkpoint format.', + compactionUpdatePrompt: 'Use the incremental company checkpoint format.', + additionalInstructions: 'Focus on the active blocker.', + }) + expect(initial).toBe( + '\ninitial work\n\n\nUse the initial company checkpoint format.\n\nAdditional user guidance for this summary:\nFocus on the active blocker.', + ) + expect(initial).not.toContain(DEFAULT_COMPACTION_PROMPT) + expect(initial).not.toContain('incremental company') + + const incremental = buildCompactionRequestText({ + conversationText: 'new work', + previousSummary: 'old work', + compactionPrompt: 'Use the initial company checkpoint format.', + compactionUpdatePrompt: 'Use the incremental company checkpoint format.', + additionalInstructions: 'Focus on the active blocker.', + }) + expect(incremental).toBe( + '\nnew work\n\n\n\nold work\n\n\nUse the incremental company checkpoint format.\n\nAdditional user guidance for this summary:\nFocus on the active blocker.', + ) + expect(incremental).not.toContain(DEFAULT_COMPACTION_UPDATE_PROMPT) + expect(incremental).not.toContain('initial company') + }) + + test('frames a concise turn-prefix request without one-shot summary guidance', () => { + const request = buildTurnPrefixCompactionRequestText('[User]: oversized turn') + expect(request).toContain(TURN_PREFIX_COMPACTION_PROMPT) + expect(request).toContain('\n[User]: oversized turn\n') + expect(request).not.toContain('Additional user guidance') + }) +}) diff --git a/packages/agentlayer-core/test/compaction.test.ts b/packages/agentlayer-core/test/compaction.test.ts new file mode 100644 index 0000000..fe2042b --- /dev/null +++ b/packages/agentlayer-core/test/compaction.test.ts @@ -0,0 +1,589 @@ +import { describe, expect, test } from 'bun:test' +import type { + LanguageModelV3, + LanguageModelV3CallOptions, + LanguageModelV3StreamPart, + LanguageModelV3StreamResult, +} from '@ai-sdk/provider' +import { MockLanguageModelV3, simulateReadableStream } from 'ai/test' +import { Agent, type AgentEvent, COMPACTION_SYSTEM_PROMPT, startState, toolCall, toolResult } from '../src' +import { ModelProvider } from '../src/models' +import { createForkState } from '../src/tools/subagent-fork' +import { assistantText, mockModel, userMessage } from './mocks' + +const usage = { + inputTokens: { total: 120, noCache: 100, cacheRead: 20, cacheWrite: 0 }, + outputTokens: { total: 30, text: 25, reasoning: 5 }, +} + +type ScriptedResponse = { text: string; usage?: typeof usage } | { error: Error } + +function scriptedModel(script: ScriptedResponse[], calls: LanguageModelV3CallOptions[]): LanguageModelV3 { + let index = 0 + return new MockLanguageModelV3({ + provider: 'mock', + modelId: 'scripted', + supportedUrls: {}, + doStream: async (options): Promise => { + calls.push(options) + const response = script[index++] + if (!response) throw new Error(`No scripted response for call ${index}`) + const chunks: LanguageModelV3StreamPart[] = + 'error' in response + ? [ + { type: 'stream-start', warnings: [] }, + { type: 'error', error: response.error }, + { type: 'finish', finishReason: { unified: 'error', raw: 'error' }, usage }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: `text-${index}` }, + { type: 'text-delta', id: `text-${index}`, delta: response.text }, + { type: 'text-end', id: `text-${index}` }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: response.usage ?? usage, + }, + ] + return { stream: simulateReadableStream({ chunks, initialDelayInMs: null, chunkDelayInMs: null }) } + }, + }) +} + +function callContains(call: LanguageModelV3CallOptions, text: string): boolean { + return JSON.stringify(call.prompt).includes(text) +} + +function spySummaryModel(calls: LanguageModelV3CallOptions[]): LanguageModelV3 { + return new MockLanguageModelV3({ + provider: 'mock', + modelId: 'summary-model', + supportedUrls: {}, + doStream: async (options) => { + calls.push(options) + const chunks: LanguageModelV3StreamPart[] = [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'summary' }, + { type: 'text-delta', id: 'summary', delta: '## Goal\nShip compaction.' }, + { type: 'text-end', id: 'summary' }, + { type: 'finish', finishReason: { unified: 'stop', raw: 'stop' }, usage }, + ] + return { stream: simulateReadableStream({ chunks, initialDelayInMs: null, chunkDelayInMs: null }) } + }, + }) +} + +describe('Agent.compact()', () => { + test('atomically replaces a prefix, emits inference and metadata, and accounts for usage', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const providerFactoryCalls: Array<{ runId: string; promptCacheKey?: string }> = [] + const agent = new Agent({ + model: spySummaryModel(calls), + tools: {}, + promptCacheKey: 'normal-cache-scope', + providerOptions: (context) => { + providerFactoryCalls.push(context) + return { openai: { promptCacheKey: context.promptCacheKey, reasoningEffort: 'high' } } + }, + autoCompact: { keepRecentTokens: 6 }, + contextWindowLimit: 200_000, + }) + const input = { + ...startState([ + userMessage('old user'), + { role: 'assistant' as const, content: 'old assistant' }, + userMessage('new user'), + { role: 'assistant' as const, content: 'new assistant' }, + ]), + contextWindowTokens: 150_000, + } + const originalJson = JSON.stringify(input) + const run = agent.compact({ state: input, additionalInstructions: 'Focus on unresolved blockers.' }) + const events: AgentEvent[] = [] + for await (const event of run) events.push(event) + const result = await run.result + + expect(result.error).toBeUndefined() + expect(result.finishReason).toBe('complete') + expect(JSON.stringify(input)).toBe(originalJson) + expect(result.state).not.toBe(input) + expect(result.state.contextWindowTokens).toBeUndefined() + expect(result.state.messages).toEqual([ + { role: 'user', content: '\n## Goal\nShip compaction.\n' }, + input.messages[2]!, + input.messages[3]!, + ]) + expect(result.state.compaction).toMatchObject({ + version: 1, + summary: '## Goal\nShip compaction.', + trigger: 'manual', + replacedMessageCount: 2, + retainedMessageCount: 2, + totalReplacedMessageCount: 2, + priorContextWindowTokens: 150_000, + }) + + expect(calls).toHaveLength(1) + expect(calls[0]!.tools).toBeUndefined() + expect(calls[0]!.toolChoice).toBeUndefined() + expect(calls[0]!.maxOutputTokens).toBe(13_107) + expect(calls[0]!.prompt[0] as { role: string; content: string }).toEqual({ + role: 'system', + content: COMPACTION_SYSTEM_PROMPT, + }) + expect(calls[0]!.providerOptions).toEqual({ + openai: { promptCacheKey: 'normal-cache-scope', reasoningEffort: 'high' }, + }) + expect(providerFactoryCalls).toHaveLength(1) + expect(providerFactoryCalls[0]!.promptCacheKey).toBe('normal-cache-scope') + + expect(result.newMessages).toHaveLength(2) + expect(result.newMessages[0]).toMatchObject({ role: 'user' }) + expect(String(result.newMessages[0]!.content)).toContain('Additional user guidance') + expect(result.newMessages[1]).toMatchObject({ + role: 'assistant', + content: [{ type: 'text', text: '## Goal\nShip compaction.' }], + }) + expect(events.map((event) => event.type)).toEqual(['message', 'message', 'compaction']) + const event = events[2] + expect(event).toMatchObject({ + type: 'compaction', + trigger: 'manual', + priorContextWindowTokens: 150_000, + replacedMessageCount: 2, + retainedMessageCount: 2, + summaryUsage: { model: 'mock/summary-model', usage: { inputTokens: 120, outputTokens: 30 } }, + }) + expect(result.tokenUsage.totals).toMatchObject({ inputTokens: 120, outputTokens: 30, reasoningTokens: 5 }) + }) + + test('uses checkpoint metadata for an incremental summary and preserves unrelated state', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const agent = new Agent({ + model: scriptedModel([{ text: 'first summary' }, { text: 'updated summary' }], calls), + tools: {}, + autoCompact: { + keepRecentTokens: 2, + compactionPrompt: 'Initial override template.', + compactionUpdatePrompt: 'Incremental override template.', + }, + }) + const first = await agent.compact({ + state: { + messages: [ + userMessage('a'), + { role: 'assistant', content: 'ok' }, + userMessage('b'), + { role: 'assistant', content: 'ok' }, + ], + toolState: { durable: true }, + }, + }).result + const continued = { + ...first.state, + messages: [...first.state.messages, userMessage('c'), { role: 'assistant' as const, content: 'ok' }], + } + const second = await agent.compact({ state: continued }).result + + expect(second.finishReason).toBe('complete') + expect(callContains(calls[0]!, 'Initial override template.')).toBe(true) + expect(callContains(calls[0]!, 'Incremental override template.')).toBe(false) + expect(callContains(calls[1]!, 'Incremental override template.')).toBe(true) + expect(callContains(calls[1]!, 'Initial override template.')).toBe(false) + expect(String(second.newMessages[0]!.content)).toContain( + '\nfirst summary\n', + ) + expect(second.state.compaction?.summary).toBe('updated summary') + expect(second.state.compaction?.totalReplacedMessageCount).toBeGreaterThan( + first.state.compaction!.totalReplacedMessageCount, + ) + expect(second.state.toolState).toEqual({ durable: true }) + }) + + test('clones checkpoint metadata only when fork projection retains its canonical summary', () => { + const state = { + messages: [ + userMessage('\nsummary\n'), + userMessage('recent work'), + ], + compaction: { + version: 1 as const, + summary: 'summary', + trigger: 'manual' as const, + replacedMessageCount: 4, + retainedMessageCount: 1, + totalReplacedMessageCount: 4, + }, + } + + const fullFork = createForkState(state, 'all', 'missing-invocation', 'delegated task') + expect(fullFork.compaction).toEqual(state.compaction) + expect(fullFork.compaction).not.toBe(state.compaction) + + const emptyFork = createForkState(state, 'none', 'missing-invocation', 'delegated task') + expect(emptyFork.compaction).toBeUndefined() + }) + + test('leaves the exact input state untouched on empty summary and provider failure', async () => { + const state = startState([ + userMessage('old'), + { role: 'assistant', content: 'old answer' }, + userMessage('new'), + { role: 'assistant', content: 'new answer' }, + ]) + const emptyAgent = new Agent({ + model: mockModel([assistantText(' ')]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }) + const empty = await emptyAgent.compact({ state }).result + expect(empty.finishReason).toBe('error') + expect(empty.error?.message).toContain('empty summary') + expect(empty.state).toBe(state) + expect(empty.newMessages).toEqual([]) + expect(empty.tokenUsage.totals).toMatchObject({ inputTokens: 0, outputTokens: 0 }) + + const failingModel = new MockLanguageModelV3({ + provider: 'mock', + modelId: 'failing', + supportedUrls: {}, + doStream: async () => ({ + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { type: 'error', error: new Error('provider unavailable') }, + { type: 'finish', finishReason: { unified: 'error', raw: 'error' }, usage }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }), + }) + const failed = await new Agent({ + model: failingModel, + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }).compact({ + state, + }).result + expect(failed.finishReason).toBe('error') + expect(failed.error?.message).toContain('provider unavailable') + expect(failed.state).toBe(state) + expect(failed.newMessages).toEqual([]) + }) + + test('accounts summarizer usage when empty output or a later compaction hook fails', async () => { + const input = { + ...startState([ + userMessage('old'), + { role: 'assistant' as const, content: 'old answer' }, + userMessage('recent'), + ]), + toolState: { durable: true }, + contextWindowTokens: 90, + } + const inputJson = JSON.stringify(input) + const empty = await new Agent({ + model: mockModel([assistantText(' ', { usage })]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }).compact({ state: input }).result + expect(empty.finishReason).toBe('error') + expect(empty.tokenUsage.totals).toMatchObject({ inputTokens: 120, outputTokens: 30 }) + expect(empty.state).toBe(input) + expect(JSON.stringify(input)).toBe(inputJson) + + const hookFailure = await new Agent({ + model: mockModel([assistantText('valid summary', { usage })]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + hooks: { + compaction: [ + () => { + throw new Error('compaction hook failed') + }, + ], + }, + }).compact({ state: input }).result + expect(hookFailure.finishReason).toBe('error') + expect(hookFailure.error?.message).toContain('compaction hook failed') + expect(hookFailure.tokenUsage.totals).toMatchObject({ inputTokens: 120, outputTokens: 30 }) + expect(hookFailure.state).toBe(input) + expect(JSON.stringify(input)).toBe(inputJson) + }) + + test('summarizes history and an oversized turn prefix into one atomic checkpoint', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const hookInputs: Array<{ replaced: number; retained: number }> = [] + const input = { + ...startState([ + userMessage('older completed request'), + { role: 'assistant' as const, content: 'older completed answer' }, + userMessage('oversized current request'), + { role: 'assistant' as const, content: 'early progress' }, + toolCall({ toolCallId: 'large-call', toolName: 'read', input: { path: 'large.ts' } }), + toolResult({ toolCallId: 'large-call', toolName: 'read', output: 'x'.repeat(400) }), + { role: 'assistant' as const, content: 'ok' }, + ]), + toolState: { durable: true }, + } + const originalJson = JSON.stringify(input) + const modelProvider = new ModelProvider() + modelProvider.getModelLimits = () => ({ context: 200_000, output: 10_000 }) + const result = await new Agent({ + model: scriptedModel([{ text: 'history summary' }, { text: 'turn prefix summary' }], calls), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + modelProvider, + hooks: { + compaction: [ + (ctx) => { + hookInputs.push({ + replaced: ctx.replacedMessages.length, + retained: ctx.retainedMessages.length, + }) + return { ...ctx.toolState, reset: true } + }, + ], + }, + }).compact({ state: input, additionalInstructions: 'Focus on the main history.' }).result + + expect(result.finishReason).toBe('complete') + expect(JSON.stringify(input)).toBe(originalJson) + expect(calls).toHaveLength(2) + expect(calls[0]!.maxOutputTokens).toBe(10_000) + expect(calls[1]!.maxOutputTokens).toBe(8_192) + expect(callContains(calls[0]!, 'Focus on the main history.')).toBe(true) + expect(callContains(calls[1]!, 'Focus on the main history.')).toBe(false) + expect(callContains(calls[1]!, 'PREFIX of a turn that was too large')).toBe(true) + expect(result.state.messages).toEqual([ + userMessage( + '\nhistory summary\n\n---\n\n**Turn Context (split turn):**\n\nturn prefix summary\n', + ), + input.messages[6]!, + ]) + expect(result.newMessages).toHaveLength(4) + expect(result.tokenUsage.totals).toMatchObject({ inputTokens: 240, outputTokens: 60 }) + expect(result.state.compaction).toMatchObject({ replacedMessageCount: 6, retainedMessageCount: 1 }) + expect(hookInputs).toEqual([{ replaced: 6, retained: 1 }]) + expect(result.state.toolState).toEqual({ durable: true, reset: true }) + }) + + test('keeps split-turn state atomic when the second summary is empty while accounting for both calls', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const input = startState([ + userMessage('older completed request'), + { role: 'assistant', content: 'older completed answer' }, + userMessage('oversized current request'), + { role: 'assistant', content: 'x'.repeat(400) }, + { role: 'assistant', content: 'ok' }, + ]) + const result = await new Agent({ + model: scriptedModel([{ text: 'history summary' }, { text: ' ' }], calls), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }).compact({ state: input }).result + + expect(result.finishReason).toBe('error') + expect(result.state).toBe(input) + expect(result.newMessages).toEqual([]) + expect(result.tokenUsage.totals).toMatchObject({ inputTokens: 240, outputTokens: 60 }) + expect(calls).toHaveLength(2) + }) + + test('uses the same configured recent-tail target for programmatic and command compaction', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const agent = new Agent({ + model: scriptedModel( + [{ text: 'programmatic summary' }, { text: 'command summary' }, { text: 'normal answer' }], + calls, + ), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }) + const messages = [ + userMessage('old'), + { role: 'assistant' as const, content: 'old answer' }, + userMessage('b'), + { role: 'assistant' as const, content: 'ok' }, + ] + const explicit = await agent.compact({ state: startState(messages) }).result + const command = await agent.run({ state: startState([...messages, userMessage('/compact focus')]) }).result + + expect(explicit.finishReason).toBe('complete') + expect(command.finishReason).toBe('complete') + expect(explicit.state.compaction?.retainedMessageCount).toBe(command.state.compaction?.retainedMessageCount) + expect(explicit.state.messages.slice(1)).toEqual(command.state.messages.slice(1, -1)) + expect(callContains(calls[1]!, 'focus')).toBe(true) + }) +}) + +describe('automatic loop compaction', () => { + test('compacts over-threshold state before hooks and aggregates summary usage', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const hookViews: Array<{ messages: string; contextWindowTokens: number }> = [] + const agent = new Agent({ + model: scriptedModel([{ text: 'threshold summary' }, { text: 'normal answer' }], calls), + tools: {}, + contextWindowLimit: 100, + autoCompact: { thresholdTokens: 10, keepRecentTokens: 2 }, + hooks: { + preRequest: [ + (ctx) => { + hookViews.push({ + messages: JSON.stringify(ctx.messages), + contextWindowTokens: ctx.contextWindowTokens, + }) + return ctx.next() + }, + ], + }, + }) + const result = await agent.run({ + state: { + messages: [ + userMessage('old user'), + { role: 'assistant', content: 'old answer' }, + userMessage('recent user'), + { role: 'assistant', content: 'recent answer' }, + userMessage('continue'), + ], + contextWindowTokens: 12, + }, + }).result + + expect(result.error).toBeUndefined() + expect(result.finishReason).toBe('complete') + expect(calls).toHaveLength(2) + expect(callContains(calls[0]!, 'old user')).toBe(true) + expect(callContains(calls[1]!, '')).toBe(true) + expect(callContains(calls[1]!, 'old user')).toBe(false) + expect(hookViews).toHaveLength(1) + expect(hookViews[0]!.messages).toContain('') + expect(hookViews[0]!.contextWindowTokens).toBe(0) + expect(result.state.compaction?.trigger).toBe('threshold') + expect(result.tokenUsage.totals).toMatchObject({ inputTokens: 240, outputTokens: 60 }) + }) + + test('does not compact disabled or stale post-checkpoint usage', async () => { + const disabledCalls: LanguageModelV3CallOptions[] = [] + const disabledState = { + messages: [userMessage('old'), { role: 'assistant' as const, content: 'answer' }, userMessage('next')], + contextWindowTokens: 99, + } + const disabled = await new Agent({ + model: scriptedModel([{ text: 'normal' }], disabledCalls), + tools: {}, + autoCompact: { enabled: false }, + contextWindowLimit: 100, + }).run({ state: disabledState }).result + expect(disabled.finishReason).toBe('complete') + expect(disabledCalls).toHaveLength(1) + expect(disabled.state.compaction).toBeUndefined() + + const staleCalls: LanguageModelV3CallOptions[] = [] + const stale = await new Agent({ + model: scriptedModel([{ text: 'normal' }], staleCalls), + tools: {}, + autoCompact: { thresholdTokens: 1 }, + }).run({ + state: { + messages: [ + userMessage('\nprior\n'), + userMessage('recent'), + ], + compaction: { + version: 1, + summary: 'prior', + trigger: 'threshold', + replacedMessageCount: 2, + retainedMessageCount: 1, + totalReplacedMessageCount: 2, + }, + }, + }).result + expect(stale.finishReason).toBe('complete') + expect(staleCalls).toHaveLength(1) + expect(stale.state.compaction?.summary).toBe('prior') + }) + + test('consumes bare and instructed compact commands without sending them to the normal model', async () => { + for (const command of ['/compact', '/compact Focus on unresolved blockers.']) { + const calls: LanguageModelV3CallOptions[] = [] + const result = await new Agent({ + model: scriptedModel([{ text: 'manual summary' }, { text: 'normal answer' }], calls), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }).run({ + state: startState([ + userMessage('old user'), + { role: 'assistant', content: 'old answer' }, + userMessage('recent user'), + userMessage(command), + ]), + }).result + + expect(result.finishReason).toBe('complete') + expect(result.state.compaction?.trigger).toBe('manual') + expect(callContains(calls[1]!, '/compact')).toBe(false) + expect(JSON.stringify(result.state.messages)).not.toContain('/compact') + expect(callContains(calls[0]!, 'Focus on unresolved blockers.')).toBe(command.includes('Focus')) + } + }) + + test('compacts and retries exactly once after context overflow', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const events: AgentEvent[] = [] + const run = new Agent({ + model: scriptedModel( + [ + { error: new Error('context_length_exceeded') }, + { text: 'overflow summary' }, + { error: new Error('maximum context length exceeded again') }, + ], + calls, + ), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }).run({ + state: startState([ + userMessage('old'), + { role: 'assistant', content: 'old answer' }, + userMessage('recent'), + ]), + stream: true, + }) + for await (const event of run) events.push(event) + const result = await run.result + + expect(result.finishReason).toBe('error') + expect(result.error?.message).toContain('maximum context length') + expect(calls).toHaveLength(3) + expect(events.filter((event) => event.type === 'compaction')).toHaveLength(1) + expect(result.state.compaction?.trigger).toBe('overflow') + }) + + test('keeps the pre-compaction model state when automatic summarization fails', async () => { + const calls: LanguageModelV3CallOptions[] = [] + const input = { + messages: [ + userMessage('old'), + { role: 'assistant' as const, content: 'old answer' }, + userMessage('recent'), + ], + contextWindowTokens: 20, + } + const result = await new Agent({ + model: scriptedModel([{ text: ' ' }], calls), + tools: {}, + autoCompact: { thresholdTokens: 10, keepRecentTokens: 2 }, + }).run({ state: input }).result + + expect(result.finishReason).toBe('error') + expect(result.state.messages).toEqual(input.messages) + expect(result.state.contextWindowTokens).toBe(20) + expect(result.state.compaction).toBeUndefined() + expect(result.tokenUsage.totals).toMatchObject({ inputTokens: 120, outputTokens: 30 }) + }) +}) diff --git a/packages/agentlayer-core/test/state.test.ts b/packages/agentlayer-core/test/state.test.ts index acf59ec..9c37f0a 100644 --- a/packages/agentlayer-core/test/state.test.ts +++ b/packages/agentlayer-core/test/state.test.ts @@ -521,6 +521,27 @@ describe('AgentState JSON round-trip', () => { const restored = JSON.parse(JSON.stringify(state)) as AgentState expect(restored.terminalChildren?.['child-1']).toEqual(state.terminalChildren?.['child-1']) }) + + test('compaction checkpoint survives round-trip and approval transforms', () => { + const state: AgentState = { + messages: [userMessage('\nsummary\n'), userMessage('recent')], + pendingToolCalls: [makeApprovalPending('call-checkpoint', 'deploy')], + compaction: { + version: 1, + summary: 'summary', + trigger: 'manual', + replacedMessageCount: 8, + retainedMessageCount: 1, + totalReplacedMessageCount: 8, + priorContextWindowTokens: 120_000, + }, + } + + const restored = JSON.parse(JSON.stringify(state)) as AgentState + expect(restored.compaction).toEqual(state.compaction) + const approved = withApprovals(restored, [{ toolCallId: 'call-checkpoint', approved: true }]) + expect(approved.compaction).toEqual(state.compaction) + }) }) // ─── sanitizeStateForPersistence() ──────────────────────────────────────────── diff --git a/packages/agentlayer-core/test/sub-agent-streaming-events.test.ts b/packages/agentlayer-core/test/sub-agent-streaming-events.test.ts index 901b4ed..ff0b878 100644 --- a/packages/agentlayer-core/test/sub-agent-streaming-events.test.ts +++ b/packages/agentlayer-core/test/sub-agent-streaming-events.test.ts @@ -2,7 +2,7 @@ import { describe, expect, test } from 'bun:test' import type { ModelMessage } from 'ai' import { z } from 'zod' import { Agent, type AgentEvent, defineTool, getAllPendingApprovals, startState, withApprovals } from '../src' -import { createSubagentsTool } from '../src/tools' +import { createForkingSubagentsTool, createSubagentsTool } from '../src/tools' import { assistantText, assistantWithToolCall, @@ -575,4 +575,162 @@ describe('sub-agent streaming events', () => { expect(subagentResults[0]).toContain('Child done after approval.') expect(JSON.stringify(resumedResult.state.messages)).not.toContain('Grandchild approved.') }) + + test('tags child compaction events and aggregates summary usage exactly once', async () => { + const child = new Agent({ + model: mockStreamingModel([ + assistantWithToolCall('dangerous', { value: 'compact' }, { usage: mockUsage(20, 5) }), + assistantText('child summary', { usage: mockUsage(7, 2) }), + assistantText('child complete', { usage: mockUsage(8, 3) }), + ]), + tools: { dangerous: dangerousTool }, + contextWindowLimit: 100, + autoCompact: { thresholdTokens: 10, keepRecentTokens: 1 }, + hooks: { + approval: [(ctx) => (ctx.toolName === 'dangerous' ? ctx.ask({ message: 'Approve?' }) : ctx.next())], + }, + }) + const subagent = createSubagentsTool({ + agents: [{ name: 'worker', description: 'Worker', agent: child }], + }) + const parent = new Agent({ + model: mockStreamingModel([ + assistantWithToolCall( + 'subagent', + { description: 'delegate', prompt: 'approved work', subagent_type: 'worker' }, + { usage: mockUsage(100, 20) }, + ), + assistantText('parent complete', { usage: mockUsage(200, 30) }), + ]), + tools: { subagent }, + }) + const paused = await parent.run({ state: startState([userMessage('go')]), stream: true }).result + const pending = getAllPendingApprovals(paused.state) + const approved = withApprovals(paused.state, [{ toolCallId: pending[0]!.pending.toolCallId, approved: true }]) + const resumedRun = parent.run({ state: approved, stream: true }) + const events: AgentEvent[] = [] + for await (const event of resumedRun) events.push(event) + const result = await resumedRun.result + const compactions = events.filter( + (event): event is Extract => event.type === 'compaction', + ) + + expect(compactions).toHaveLength(1) + expect(compactions[0]).toMatchObject({ + trigger: 'threshold', + parentToolCallId: expect.any(String), + agentId: expect.any(String), + summaryUsage: { usage: { inputTokens: 7, outputTokens: 2 } }, + }) + expect(result.tokenUsage.totals.inputTokens).toBe(215) + expect(result.tokenUsage.totals.outputTokens).toBe(35) + }) + + test('aggregates failed child summarizer usage once without committing a checkpoint', async () => { + const model = mockStreamingModel([ + assistantWithToolCall('subagent', { prompt: '/compact' }, { usage: mockUsage(100, 20) }), + assistantText(' ', { usage: mockUsage(7, 2) }), + assistantText('parent complete', { usage: mockUsage(200, 30) }), + ]) + const subagent = createForkingSubagentsTool({ agents: [] }) + const parent = new Agent({ model, tools: { subagent }, autoCompact: { keepRecentTokens: 2 } }) + const run = parent.run({ + state: startState([ + userMessage('old context'), + { role: 'assistant', content: 'settled answer' }, + userMessage('delegate compaction'), + ]), + stream: true, + }) + const events: AgentEvent[] = [] + for await (const event of run) events.push(event) + const result = await run.result + + expect(result.finishReason).toBe('complete') + expect(result.state.compaction).toBeUndefined() + expect(result.tokenUsage.totals.inputTokens).toBe(307) + expect(result.tokenUsage.totals.outputTokens).toBe(52) + expect(events.filter((event) => event.type === 'compaction')).toHaveLength(0) + expect(getSubagentResultTexts(result.state.messages)[0]).toContain( + 'Compaction summarizer returned an empty summary', + ) + }) + + test('aggregates both split-turn child summaries exactly once', async () => { + const childId = 'split-child' + const child = new Agent({ + model: mockStreamingModel([ + assistantText('child history summary', { usage: mockUsage(7, 2) }), + assistantText('child turn prefix summary', { usage: mockUsage(8, 3) }), + assistantText('child complete', { usage: mockUsage(9, 4) }), + ]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }) + const subagent = createForkingSubagentsTool({ + agents: [{ name: 'worker', description: 'Worker', agent: child, resumable: true }], + }) + const parent = new Agent({ + model: mockStreamingModel([ + assistantWithToolCall( + 'subagent', + { prompt: '/compact focus on child history', agent_id: childId }, + { usage: mockUsage(100, 20) }, + ), + assistantText('parent complete', { usage: mockUsage(200, 30) }), + ]), + tools: { subagent }, + }) + const run = parent.run({ + state: { + messages: [userMessage('resume child')], + terminalChildren: { + [childId]: { + state: startState([ + userMessage('older child request'), + { role: 'assistant', content: 'older child answer' }, + userMessage('oversized child turn'), + { role: 'assistant', content: 'early child work' }, + { + role: 'assistant', + content: [{ type: 'tool-call', toolCallId: 'child-read', toolName: 'read', input: {} }], + }, + { + role: 'tool', + content: [ + { + type: 'tool-result', + toolCallId: 'child-read', + toolName: 'read', + output: { type: 'text', value: 'x'.repeat(400) }, + }, + ], + }, + { role: 'assistant', content: 'ok' }, + ]), + lastOutcome: 'complete', + completedTurns: 1, + runtime: { type: 'specialist', subagentType: 'worker' }, + }, + }, + }, + stream: true, + }) + const events: AgentEvent[] = [] + for await (const event of run) events.push(event) + const result = await run.result + const compaction = events.find( + (event): event is Extract => event.type === 'compaction', + ) + + expect(compaction).toMatchObject({ + agentId: childId, + summaryUsage: { usage: { inputTokens: 15, outputTokens: 5 } }, + }) + expect(result.tokenUsage.totals.inputTokens).toBe(324) + expect(result.tokenUsage.totals.outputTokens).toBe(59) + expect(result.state.terminalChildren?.[childId]?.state.compaction?.summary).toContain( + 'child turn prefix summary', + ) + }) }) diff --git a/packages/agentlayer-core/test/subagent-tool.test.ts b/packages/agentlayer-core/test/subagent-tool.test.ts index cc42f3a..88396a0 100644 --- a/packages/agentlayer-core/test/subagent-tool.test.ts +++ b/packages/agentlayer-core/test/subagent-tool.test.ts @@ -1103,4 +1103,172 @@ describe('forking subagent tool', () => { expect(new Set(childCacheKeys)).toHaveLength(1) }) } + + test('compacts fork children without mutating the parent snapshot', async () => { + const usage = { + inputTokens: { total: 12, noCache: 12, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 3, text: 3, reasoning: 0 }, + } + const model = mockModel([ + assistantWithToolCall('subagent', { prompt: '/compact Preserve child verification.' }), + assistantText('fork-only summary', { usage }), + assistantText('fork child complete', { usage }), + assistantText('parent complete', { usage }), + ]) + const subagent = createForkingSubagentsTool({ agents: [] }) + const parent = new Agent({ model, tools: { subagent }, autoCompact: { keepRecentTokens: 2 } }) + const input = startState([ + userMessage('old parent context'), + { role: 'assistant', content: 'settled parent answer' }, + userMessage('delegate and keep my state intact'), + ]) + const inputJson = JSON.stringify(input) + const result = await parent.run({ state: input }).result + + expect(JSON.stringify(input)).toBe(inputJson) + expect(result.state.compaction).toBeUndefined() + const records = Object.values(result.state.terminalChildren ?? {}) + expect(records).toHaveLength(1) + expect(records[0]!.state.compaction).toMatchObject({ trigger: 'manual' }) + expect(records[0]!.state.compaction?.summary).toContain('fork-only summary') + expect(JSON.stringify(records[0]!.state.messages)).not.toContain('/compact') + expect(JSON.stringify(result.state.messages)).toContain('old parent context') + }) + + test('compacts a paused child on approval resume without compacting its parent', async () => { + const highUsage = { + inputTokens: { total: 20, noCache: 20, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 5, text: 5, reasoning: 0 }, + } + const child = new Agent({ + model: mockModel([ + assistantWithToolCall('dangerous', { value: 'resume' }, { usage: highUsage }), + assistantText('paused-child summary'), + assistantText('resumed child complete'), + ]), + tools: { dangerous: approvedTool }, + contextWindowLimit: 100, + autoCompact: { thresholdTokens: 10, keepRecentTokens: 1 }, + hooks: { + approval: [(ctx) => (ctx.toolName === 'dangerous' ? ctx.ask({ message: 'Approve?' }) : ctx.next())], + }, + }) + const subagent = createForkingSubagentsTool({ + agents: [{ name: 'worker', description: 'Worker', agent: child, resumable: true }], + }) + const parent = new Agent({ + model: mockModel([ + assistantWithToolCall('subagent', { + prompt: 'perform approved work', + subagent_type: 'worker', + }), + assistantText('parent complete'), + ]), + tools: { subagent }, + }) + const paused = await parent.run({ state: startState([userMessage('parent context')]) }).result + expect(paused.finishReason).toBe('approvalRequired') + const pending = getAllPendingApprovals(paused.state) + const approved = withApprovals(paused.state, [{ toolCallId: pending[0]!.pending.toolCallId, approved: true }]) + const resumed = await parent.run({ state: approved }).result + + expect(resumed.finishReason).toBe('complete') + expect(resumed.state.compaction).toBeUndefined() + const records = Object.values(resumed.state.terminalChildren ?? {}) + expect(records).toHaveLength(1) + expect(records[0]!.state.compaction).toMatchObject({ + trigger: 'threshold', + }) + expect(records[0]!.state.compaction?.summary).toContain('paused-child summary') + }) + + test('keeps sibling and terminal-resumed child checkpoints isolated', async () => { + const firstId = 'first-child' + const secondId = 'second-child' + const firstChild = new Agent({ + model: mockModel([assistantText('first summary'), assistantText('first done')]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }) + const secondChild = new Agent({ + model: mockModel([assistantText('second summary'), assistantText('second done')]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }) + const subagent = createForkingSubagentsTool({ + agents: [ + { name: 'first', description: 'First', agent: firstChild, resumable: true }, + { name: 'second', description: 'Second', agent: secondChild, resumable: true }, + ], + }) + const parent = new Agent({ + model: mockModel([ + assistantWithToolCalls( + { toolName: 'subagent', input: { prompt: '/compact', agent_id: firstId } }, + { toolName: 'subagent', input: { prompt: '/compact', agent_id: secondId } }, + ), + assistantText('siblings complete'), + ]), + tools: { subagent }, + }) + const firstRun = await parent.run({ + state: { + messages: [userMessage('run siblings')], + terminalChildren: { + [firstId]: { + state: startState([ + userMessage('first old context'), + { role: 'assistant', content: 'first old answer' }, + ]), + lastOutcome: 'complete', + completedTurns: 1, + runtime: { type: 'specialist', subagentType: 'first' }, + }, + [secondId]: { + state: startState([ + userMessage('second old context'), + { role: 'assistant', content: 'second old answer' }, + ]), + lastOutcome: 'complete', + completedTurns: 1, + runtime: { type: 'specialist', subagentType: 'second' }, + }, + }, + }, + }).result + const entries = Object.entries(firstRun.state.terminalChildren ?? {}) + expect(entries).toHaveLength(2) + const siblingSummaries = entries.map(([, record]) => record.state.compaction?.summary ?? '') + expect(siblingSummaries.some((summary) => summary.includes('first summary'))).toBe(true) + expect(siblingSummaries.some((summary) => summary.includes('second summary'))).toBe(true) + expect(entries[0]![1].state).not.toBe(entries[1]![1].state) + + const [resumeId, resumeRecord] = entries[0]! + const resumedAgent = resumeRecord.runtime.type === 'specialist' ? resumeRecord.runtime.subagentType : 'first' + const resumeChild = new Agent({ + model: mockModel([assistantText('updated resumed summary'), assistantText('resumed child done')]), + tools: {}, + autoCompact: { keepRecentTokens: 2 }, + }) + const resumedTool = createForkingSubagentsTool({ + agents: [{ name: resumedAgent, description: 'Resumed', agent: resumeChild, resumable: true }], + }) + const resumedParent = new Agent({ + model: mockModel([ + assistantWithToolCall('subagent', { prompt: '/compact Update only this child.', agent_id: resumeId }), + assistantText('resume complete'), + ]), + tools: { subagent: resumedTool }, + }) + const resumed = await resumedParent.run({ + state: { ...firstRun.state, messages: [...firstRun.state.messages, userMessage('resume one child')] }, + }).result + const untouchedEntry = entries.find(([id]) => id !== resumeId)! + + expect(resumed.state.terminalChildren?.[resumeId]?.state.compaction?.summary).not.toBe( + resumeRecord.state.compaction?.summary, + ) + expect(resumed.state.terminalChildren?.[untouchedEntry[0]]).toEqual(untouchedEntry[1]) + expect(resumed.state.compaction).toBeUndefined() + }) }) diff --git a/packages/agentlayer-filesystem/src/coding-agent.ts b/packages/agentlayer-filesystem/src/coding-agent.ts index 6003e5a..70817cb 100644 --- a/packages/agentlayer-filesystem/src/coding-agent.ts +++ b/packages/agentlayer-filesystem/src/coding-agent.ts @@ -1,6 +1,7 @@ import { isAbsolute, resolve } from 'node:path' import type { AgentEvent } from '@humanlayer/agentlayer-core' import { + type CompactionHook, createSubagentsTool, createWebFetchTool, type PostToolUseHook, @@ -10,7 +11,12 @@ import { type Tool, } from '@humanlayer/agentlayer-core' import type { ReadToolModalities, Skill } from '@humanlayer/agentlayer-core/interfaces' -import { createFileStateTrackingHook, createReadBeforeWriteHook, createWastedReadHook } from './hooks/file-state' +import { + createFileStateCompactionHook, + createFileStateTrackingHook, + createReadBeforeWriteHook, + createWastedReadHook, +} from './hooks/file-state' import { createBashOutputTruncationHook, createGlobOutputTruncationHook, @@ -42,6 +48,7 @@ export function createAgentFilesystemHooks(opts: CreateAgentFilesystemHooksOptio preToolUse: readonly PreToolUseHook[] postToolUse: readonly PostToolUseHook[] preRequest: readonly PreRequestHook[] + compaction: readonly CompactionHook[] } { const sharedOutputTruncation = opts.outputTruncation ? { @@ -70,6 +77,7 @@ export function createAgentFilesystemHooks(opts: CreateAgentFilesystemHooksOptio createFileStateTrackingHook({ cwd: opts.cwd }), ], preRequest: [], + compaction: [createFileStateCompactionHook()], } as const } diff --git a/packages/agentlayer-filesystem/src/hooks/file-state.ts b/packages/agentlayer-filesystem/src/hooks/file-state.ts index c297611..a817fe1 100644 --- a/packages/agentlayer-filesystem/src/hooks/file-state.ts +++ b/packages/agentlayer-filesystem/src/hooks/file-state.ts @@ -1,6 +1,6 @@ import { createHash } from 'node:crypto' import { readFile, stat } from 'node:fs/promises' -import type { PostToolUseHook, PreToolUseHook } from '@humanlayer/agentlayer-core' +import type { CompactionHook, PostToolUseHook, PreToolUseHook } from '@humanlayer/agentlayer-core' import { createPostToolUseHook, createPreToolUseHook, sanitizeTextForModelState } from '@humanlayer/agentlayer-core' import { ApplyPatchTool, EditTool, ReadTool, WriteTool } from '@humanlayer/agentlayer-core/interfaces' import { type PatchOperation, parsePatch } from '@humanlayer/agentlayer-core/utils' @@ -31,6 +31,16 @@ export type FileVerificationStateMap = Record { + const next = { ...toolState } + delete next[FILE_READ_STATE_KEY] + delete next[FILE_VERIFICATION_STATE_KEY] + return next + } +} + export interface FileStateHookOptions { cwd?: string } diff --git a/packages/agentlayer-filesystem/test/file-state.test.ts b/packages/agentlayer-filesystem/test/file-state.test.ts index 4547e37..ad4039d 100644 --- a/packages/agentlayer-filesystem/test/file-state.test.ts +++ b/packages/agentlayer-filesystem/test/file-state.test.ts @@ -12,6 +12,7 @@ import { } from '@humanlayer/agentlayer-core' import { z } from 'zod' import { + createFileStateCompactionHook, createFileStateTrackingHook, createReadBeforeWriteHook, createReadBeforeWriteHooks, @@ -156,6 +157,59 @@ function createDirectHookHarness(cwd?: string): DirectHookHarness { } describe('file-state hooks', () => { + test('compaction clears stale file evidence and permits a required reread', async () => { + const dir = await mkdtemp(join(tmpdir(), 'file-state-hook-test-')) + try { + const filePath = join(dir, 'compacted-read.txt') + const content = 'important context\n' + await writeFile(filePath, content) + const tracked = createDirectHookHarness() + await tracked.recordReadObservation({ file_path: filePath }, content) + let readExecuted = 0 + const readTool = defineTool({ + name: 'read', + description: 'Read file', + input: z.object({ file_path: z.string() }), + output: z.string(), + execute: async () => { + readExecuted += 1 + return content + }, + }) + const result = await new Agent({ + model: mockModel([ + assistantText('summary'), + assistantWithToolCall('read', { file_path: filePath }), + assistantText('done'), + ]), + tools: { read: readTool }, + autoCompact: { thresholdTokens: 1, keepRecentTokens: 2 }, + hooks: { + compaction: [createFileStateCompactionHook()], + preToolUse: [createWastedReadHook()], + postToolUse: [createFileStateTrackingHook()], + }, + }).run({ + state: { + messages: [ + userMessage('old read request'), + { role: 'assistant', content: 'old read complete' }, + userMessage('continue'), + ], + toolState: { ...tracked.state, durable: 'keep' }, + contextWindowTokens: 2, + }, + }).result + + expect(result.finishReason).toBe('complete') + expect(readExecuted).toBe(1) + expect(result.state.toolState?.durable).toBe('keep') + expect(result.state.toolState?.[FILE_READ_STATE_KEY]).toBeDefined() + } finally { + await rm(dir, { recursive: true }) + } + }) + test('direct hook harness records read state and gates mutations', async () => { const dir = await mkdtemp(join(tmpdir(), 'file-state-hook-test-')) try {