diff --git a/packages/agent-core-v2/src/human/agent/machine.ts b/packages/agent-core-v2/src/human/agent/machine.ts index 3570a17fc6..5cb4f0111f 100644 --- a/packages/agent-core-v2/src/human/agent/machine.ts +++ b/packages/agent-core-v2/src/human/agent/machine.ts @@ -43,6 +43,8 @@ export type AgentEvent = | { type: 'input.steer'; id: string } | { type: 'input.cancel'; id: string } | { type: 'input.abort' } + | { type: 'input.pause' } + | { type: 'input.continue' } | { type: 'turn.spawn_tools'; toolCalls: ToolCall[] } | { type: 'turn.drain' } | { type: 'turn.reminders_consumed'; reminders: HistoryMessage[] } @@ -89,6 +91,7 @@ export interface AgentMachineContext { activeTurnId?: number; branchId: string; drainedId?: string; + paused: boolean; } function completionNotification(toolCall: ToolCall, output: ToolOutput): UserEntry { @@ -169,6 +172,13 @@ function hasPendingWork(context: AgentMachineContext): boolean { return context.notifications.length > 0 || context.queue.length > 0; } +function historyEndsMidToolChain(messages: readonly HistoryMessage[]): boolean { + const last = messages.at(-1); + if (last === undefined) return false; + if (last.message.role === 'tool') return true; + return last.message.role === 'assistant' && last.message.toolCalls.length > 0; +} + function hasBackgroundWork(context: AgentMachineContext): boolean { return Object.keys(context.background).length > 0; } @@ -336,6 +346,7 @@ export function createAgentMachine({ queue: [], turnId: 0, branchId: 'main', + paused: false, }), invoke: [ { @@ -430,6 +441,12 @@ export function createAgentMachine({ target: '.idle', actions: ['abortScope', 'resetMirror', 'emitReset', 'forwardToParent'], }, + 'input.pause': { + actions: assign({ paused: true }), + }, + 'input.continue': { + actions: assign({ paused: false }), + }, 'store.error': { actions: 'forwardToParent', }, @@ -469,7 +486,7 @@ export function createAgentMachine({ idle: { initial: 'ready', always: { - guard: ({ context }) => hasPendingWork(context), + guard: ({ context }) => hasPendingWork(context) && !context.paused, target: 'running', actions: [ sendTo('store', ({ context }) => { @@ -491,6 +508,33 @@ export function createAgentMachine({ assign(({ context }) => drainPendingPatch(context)), ], }, + on: { + 'input.continue': { + guard: ({ context }) => + !hasPendingWork(context) && historyEndsMidToolChain(context.messages), + target: 'running', + actions: [ + assign({ paused: false }), + sendTo('store', ({ context }) => { + const head = context.queue[0]; + return { + type: 'store.append' as const, + event: [ + ...context.notifications.map((entry) => messageAppended({ message: entry })), + ...(head === undefined + ? [] + : [ + messageAppended({ message: createUserEntry(head.message, { source: 'input' }) }), + queueDrained({ id: head.id }), + ]), + ...(context.notifications.length === 0 ? [] : [notificationsDrained({})]), + ], + }; + }), + assign(({ context }) => drainPendingPatch(context)), + ], + }, + }, states: { ready: { always: { @@ -581,6 +625,12 @@ export function createAgentMachine({ 'forwardToParent', ], }, + 'input.pause': { + actions: [assign({ paused: true }), sendTo('turn', { type: 'turn.pause' as const })], + }, + 'input.continue': { + actions: [assign({ paused: false }), sendTo('turn', { type: 'turn.continue' as const })], + }, 'turn.drain': { actions: enqueueActions(({ context, enqueue }) => { const messages = [...context.notifications, ...context.reminders]; diff --git a/packages/agent-core-v2/src/human/agent/turn.ts b/packages/agent-core-v2/src/human/agent/turn.ts index 630e8b81c1..91115534d6 100644 --- a/packages/agent-core-v2/src/human/agent/turn.ts +++ b/packages/agent-core-v2/src/human/agent/turn.ts @@ -1,4 +1,4 @@ -import { assign, raise, setup } from '#/xstate2'; +import { assign, fromPromise, raise, setup } from '#/xstate2'; import { emptyResponseError } from '#/llm/empty-response'; import type { LlmErrorMessage, LlmRemoteErrorMessage } from '#/llm/errors'; @@ -187,6 +187,8 @@ export type TurnEvent = | LlmEvent | TurnToolEvent | { type: 'turn.notify'; messages: HistoryMessage[] } + | { type: 'turn.pause' } + | { type: 'turn.continue' } | { type: 'turn.abort' }; export type TurnLlmEvent = @@ -217,6 +219,7 @@ export interface TurnMachineContext { attempt: number; delayMs: number; appliedRecoveries: LlmRecoveryRecord[]; + paused: boolean; lastError?: LlmRemoteErrorMessage; outcome?: 'done' | 'failed' | 'aborted'; error?: unknown; @@ -343,10 +346,18 @@ function emptyErrorOf(context: TurnMachineContext): LlmErrorMessage<'empty_respo ); } +export interface TurnBeforeStepContext { + messages: readonly HistoryMessage[]; + request: LlmRequestConfig; +} + +export type TurnBeforeStep = (context: TurnBeforeStepContext) => void | Promise; + export interface CreateTurnMachineOptions { readonly recovery?: LlmRecovery; readonly retry?: LlmRetryOptions; readonly abortGraceMs?: number; + readonly onBeforeStep?: TurnBeforeStep; } export function createTurnMachine( @@ -365,6 +376,9 @@ export function createTurnMachine( }, actors: { llmActor, + onBeforeStepActor: fromPromise(async ({ input }) => { + await options?.onBeforeStep?.(input); + }), }, actions: { forwardToParent: ({ self, event }) => { @@ -403,7 +417,7 @@ export function createTurnMachine( }, }).createMachine({ id: 'turn', - initial: 'thinking', + initial: 'gating', context: ({ input }) => { const toolCallIds = new ToolCallIdNormalizer(); toolCallIds.seedFrom(toInputMessages(input.history)); @@ -420,9 +434,36 @@ export function createTurnMachine( attempt: 1, delayMs: 0, appliedRecoveries: [], + paused: false, }; }, + on: { + 'turn.pause': { + actions: assign({ paused: true }), + }, + 'turn.continue': { + actions: assign({ paused: false }), + }, + }, states: { + gating: { + always: [{ guard: () => options?.onBeforeStep === undefined, target: 'thinking' }], + invoke: { + src: 'onBeforeStepActor', + input: ({ context }) => ({ + messages: [...context.input.history, ...context.produced], + request: context.input.request, + }), + onDone: { target: 'thinking' }, + onError: { target: 'done' }, + }, + on: { + 'turn.abort': { + target: 'aborted', + actions: assign({ outcome: 'aborted' as const }), + }, + }, + }, thinking: { entry: [ assign({ @@ -441,20 +482,20 @@ export function createTurnMachine( ], invoke: { src: 'llmActor', - input: ({ context }) => { - const entries = [...context.input.history, ...context.produced]; - return { - config: context.input.request, - signal: context.llmScope.signal, - content: { - messages: attemptMessages(context, recovery), - usedContextTokens: estimateUsedContextTokens(entries, { + input: ({ context }) => ({ + config: context.input.request, + signal: context.llmScope.signal, + content: { + messages: attemptMessages(context, recovery), + usedContextTokens: estimateUsedContextTokens( + [...context.input.history, ...context.produced], + { systemPrompt: context.input.request.systemPrompt, tools: context.input.request.tools, - }), - }, - }; - }, + }, + ), + }, + }), onError: { target: 'failed', actions: assign({ @@ -778,6 +819,16 @@ export function createTurnMachine( }, on: { 'turn.notify': [ + { + guard: ({ context }) => context.paused, + target: 'done', + actions: [ + assign(({ context, event }) => ({ + produced: [...context.produced, ...event.messages], + })), + 'signalRemindersConsumed', + ], + }, { guard: ({ context, event }) => event.messages.length === 0 && maxStepsExceeded(context), @@ -788,7 +839,7 @@ export function createTurnMachine( })), }, { - target: 'thinking', + target: 'gating', actions: [ assign(({ context, event }) => ({ produced: [...context.produced, ...event.messages], diff --git a/packages/agent-core-v2/src/human/compaction/compaction-instruction.md b/packages/agent-core-v2/src/human/compaction/compaction-instruction.md new file mode 100644 index 0000000000..90742b820b --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/compaction-instruction.md @@ -0,0 +1,73 @@ +You are about to run out of context. Create a handoff summary for the +model that will resume this task after the earlier conversation is cleared. + +--- This message is a direct task, not part of the above conversation --- + +Do not impose rigid section headings; let the shape follow the task. Write it +in the same language the conversation has been using — do not switch to English +just because these instructions happen to be in English. + +Make the summary self-sufficient: the next turn will see only the preserved +messages and this summary — every other assistant message, tool call, and tool +result above will be gone. In your own words, preserve what you genuinely need +to continue: + +- What the latest request is actually asking for: your reading of its intent and + any ambiguity you have already resolved — not a re-transcription, since what + fits is kept verbatim in the preserved messages. But those kept messages are + size-capped, so a long request is truncated there: if the latest request is + large (a big paste or file), preserve the parts at risk of being dropped — + above all the actual ask. If several requests are in play, say which one governs + the next move, and re-quote any still-relevant earlier request that may have + scrolled out of the kept messages. +- The instructions and constraints currently in force (user preferences, + project rules, environment and tooling limits) — condensed to what still + matters, keeping decisions you have already settled (what you chose and why) + separate from questions still open, so you neither silently reopen a closed + choice nor treat an undecided point as decided. +- What has actually been done, at high fidelity: keep the exact commands that + were run, the exact file paths touched, and whether each succeeded or failed — + and the results themselves, not just the commands: the concrete values + returned, the key lines or error text, the schema or signature a lookup + revealed, since re-running to recover them may be slow or impossible. Keep only + the final working version of any code; drop intermediate attempts and + already-resolved errors. +- What you still don't know: context the next step depends on that this + conversation never established — files or paths referenced but not yet read, + schemas or APIs assumed but unseen, questions the user has not answered. Name + these gaps so the next turn goes and checks them instead of assuming. +- The forward plan — and this is the moment to invest in it. Right now you + hold more context on this task than you ever will again; the next turn + resumes with less, so the plan you commit here is the one it will follow. + Give the exact next command or tool call, but don't stop at the next step: + set out the remaining sequence to finish, the decisions you have already + made for those upcoming steps (so the next turn doesn't reopen them), the + obstacles or edge cases you can already foresee and how you mean to handle + them, and any work you can commit to now — the exact patch, query, or shape + of the final answer you already know you will produce. Anything you settle + here is one less thing the next turn must rediscover. Include any required + format for the final answer. + +This conversation's event log stays on disk and a recovery pointer is appended below this summary automatically, so you need not reproduce long outputs verbatim — keep exact identifiers, key values and error lines, and name anything the next turn should look up. + +Your TODO list is re-attached automatically below this summary from its live +source, so do not transcribe it — copying it wastes space and can contradict the +live version. What that list cannot hold is the reasoning between tasks — why one +was reordered or dropped, or a decision on one that constrains another — so +record that instead. + +Be honest about uncertainty. If an earlier step claimed something was done but +was never verified (tests "passing", a fix "working", a file "created"), say so +plainly and treat it as unverified rather than fact — re-check before relying +on it. + +Be concise, and keep the summary proportional to the task: a long multi-step +task warrants detail, but a trivial or nearly finished exchange needs only a +sentence or two — do not pad it out. Include the critical data, identifiers, and +references needed to continue, and omit anything that does not change the next +move. + +Respond with text only. Do not call any tools — you already have everything you +need in the conversation history. + +${custom_instruction_block} diff --git a/packages/agent-core-v2/src/human/compaction/compaction-summary-prefix.md b/packages/agent-core-v2/src/human/compaction/compaction-summary-prefix.md new file mode 100644 index 0000000000..3b8345bf34 --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/compaction-summary-prefix.md @@ -0,0 +1 @@ +The conversation so far has been compacted to free up context. What follows is your own working summary of this task — use it to continue your train of thought rather than starting over. Treat it as notes, not proof: where it says a step was done, tests passed, or a fix worked, verify that yourself before relying on it. Any user messages earlier in this context are preserved verbatim from the compacted conversation; where a system-reminder note among them marks an omitted middle section, the user messages it replaced are covered by this summary. The summary records which earlier requests were already addressed. diff --git a/packages/agent-core-v2/src/human/compaction/controller.ts b/packages/agent-core-v2/src/human/compaction/controller.ts new file mode 100644 index 0000000000..b3c58d97f4 --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/controller.ts @@ -0,0 +1,247 @@ +import { estimateUsedContextTokens } from '#/agent/context-usage'; +import type { ExternalEvent } from '#/eventStore/events'; +import type { UserMessage } from '#/llm/message'; +import { + compactionCancelled, + compactionCompleted, + compactionStarted, +} from '#/session/events'; +import type { AgentActorRef } from '#/session/machine'; +import type { SessionStores } from '#/session/stores'; +import type { TurnBeforeStep, TurnBeforeStepContext } from '#/agent/turn'; +import { createActor, waitFor, type ActorRefFrom, type Subscription } from '#/xstate2'; + +import { CompactError, isContextOverflowError } from './errors'; +import { + createCompactionMachine, + type CompactionEvent, + type CompactionMachineOutput, + type CompactionPhase, + type CompactionReason, +} from './machine'; +import type { Summarize } from './summarize'; + +export { + type CompactionCancelCause, + type CompactionEvent, + type CompactionPhase, + type CompactionReason, + type CompactionStats, +} from './machine'; + +export interface CompactionStatus { + phase: CompactionPhase; + reason?: CompactionReason; + startedAt?: number; +} + +export interface CompactionControllerDeps { + agentId: string; + actor: AgentActorRef; + stores: SessionStores; + summarize: Summarize; + budget: { + maxContextTokens(): number; + triggerRatio: number; + }; + continuation?: (reason: CompactionReason) => UserMessage | undefined; + maxAutoAttempts?: number; + todos?: () => string | undefined; + onEvent?: (event: CompactionEvent) => void; + onWillCompact?: (input: { + reason: CompactionReason; + instruction?: string; + signal: AbortSignal; + tokenCount: number; + }) => void | Promise; +} + +export interface CompactionController { + compact(instruction?: string): Promise<{ branchId: string }>; + cancel(): void; + status(): CompactionStatus; + onBeforeStep: TurnBeforeStep; + dispose(): void; +} + +type RunActor = ActorRefFrom>; + +interface ActiveRun { + actor: RunActor; + reason: CompactionReason; + startedAt: number; +} + +const DEFAULT_MAX_AUTO_ATTEMPTS = 3; + +function errorMessageOf(error: unknown): string | undefined { + if (error === undefined) return undefined; + if (error instanceof Error) return error.message; + if (typeof error === 'string') return error; + return JSON.stringify(error) ?? String(typeof error); +} + +export function createCompactionController(deps: CompactionControllerDeps): CompactionController { + const maxAutoAttempts = deps.maxAutoAttempts ?? DEFAULT_MAX_AUTO_ATTEMPTS; + const machine = createCompactionMachine(deps); + let active: ActiveRun | undefined; + let pendingAuto: { reason: 'budget' | 'overflow' } | undefined; + let overflowAttempts = 0; + let lastCompactedTokens: number | undefined; + + const record = (event: ExternalEvent): void => { + void deps.stores + .session() + .then((session) => session.dispatch(event)) + .then( + () => undefined, + () => undefined, + ); + }; + + const budgetExceeded = (used: number): boolean => { + const max = deps.budget.maxContextTokens(); + if (max <= 0 || used < max * deps.budget.triggerRatio) return false; + return lastCompactedTokens === undefined || used > lastCompactedTokens; + }; + + const firePending = (): void => { + const scheduled = pendingAuto; + pendingAuto = undefined; + if (scheduled === undefined) return; + if (scheduled.reason === 'budget') { + const history = deps.stores.get(deps.agentId)?.getState().history; + if (history === undefined || !budgetExceeded(estimateUsedContextTokens(history))) { + return; + } + } + queueMicrotask(() => void run(scheduled.reason)); + }; + + const pipeEvents = (actor: RunActor): Subscription[] => [ + actor.on('compaction.started', (event) => { + deps.onEvent?.(event); + record( + compactionStarted({ + agentId: deps.agentId, + reason: event.reason, + instruction: event.instruction, + }), + ); + }), + actor.on('compaction.blocked', (event) => { + deps.onEvent?.(event); + }), + actor.on('compaction.completed', (event) => { + deps.onEvent?.(event); + record(compactionCompleted({ agentId: deps.agentId, branch: event.branchId })); + lastCompactedTokens = event.stats.tokensBefore; + active = undefined; + firePending(); + }), + actor.on('compaction.cancelled', (event) => { + deps.onEvent?.(event); + record( + compactionCancelled({ + agentId: deps.agentId, + cause: event.cause, + errorMessage: errorMessageOf(event.error), + }), + ); + active = undefined; + firePending(); + }), + ]; + + const run = async ( + reason: CompactionReason, + instruction?: string, + ): Promise<{ branchId: string } | undefined> => { + if (active !== undefined) { + if (reason === 'manual') { + throw new CompactError('busy', 'compaction is already running'); + } + pendingAuto = { reason }; + return undefined; + } + if (deps.stores.get(deps.agentId) === undefined) { + if (reason === 'manual') { + throw new CompactError('unknown-agent', `unknown agent: '${deps.agentId}'`); + } + return undefined; + } + const actor = createActor(machine, { input: { reason, instruction } }); + const current = { actor, reason, startedAt: Date.now() }; + active = current; + const subscriptions = pipeEvents(actor); + await deps.stores.session(); + actor.start(); + try { + const snapshot = await waitFor(actor, (s) => s.status !== 'active'); + const output = snapshot.output as CompactionMachineOutput; + if (output.status === 'completed') { + return { branchId: output.branchId }; + } + if (reason === 'manual') { + throw output.error; + } + return undefined; + } finally { + for (const subscription of subscriptions) { + subscription.unsubscribe(); + } + if (active === current) { + active = undefined; + } + actor.stop(); + } + }; + + const subscriptions: Subscription[] = [ + deps.actor.on('turn.done', () => { + overflowAttempts = 0; + }), + deps.actor.on('turn.failed', (event) => { + if (!isContextOverflowError(event.error) || overflowAttempts >= maxAutoAttempts) { + return; + } + overflowAttempts += 1; + queueMicrotask(() => void run('overflow')); + }), + deps.actor.on('turn.aborting', () => { + active?.actor.send({ type: 'cancel', cause: 'user-abort' }); + }), + ]; + + const onBeforeStep: TurnBeforeStep = async ({ messages, request }: TurnBeforeStepContext) => { + const used = estimateUsedContextTokens(messages, { + systemPrompt: request.systemPrompt, + tools: request.tools, + }); + if (!budgetExceeded(used)) return; + queueMicrotask(() => void run('budget')); + throw new CompactError('budget-blocked', 'context budget exceeded; compacting before next step'); + }; + + return { + compact: (instruction) => run('manual', instruction) as Promise<{ branchId: string }>, + cancel: () => { + active?.actor.send({ type: 'cancel', cause: 'cancelled' }); + }, + status: () => + active === undefined + ? { phase: 'idle' } + : { + phase: active.actor.getSnapshot().value as CompactionPhase, + reason: active.reason, + startedAt: active.startedAt, + }, + onBeforeStep, + dispose: () => { + active?.actor.send({ type: 'cancel', cause: 'cancelled' }); + for (const subscription of subscriptions) { + subscription.unsubscribe(); + } + }, + }; +} diff --git a/packages/agent-core-v2/src/human/compaction/errors.ts b/packages/agent-core-v2/src/human/compaction/errors.ts new file mode 100644 index 0000000000..8a91891477 --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/errors.ts @@ -0,0 +1,31 @@ +export type CompactErrorCode = + | 'busy' + | 'unknown-agent' + | 'insufficient' + | 'drift' + | 'summary-failed' + | 'aborted' + | 'cancelled' + | 'reset-timeout' + | 'budget-blocked'; + +export class CompactError extends Error { + readonly code: CompactErrorCode; + + constructor(code: CompactErrorCode, message: string) { + super(message); + this.name = 'CompactError'; + this.code = code; + } +} + +export function isContextOverflowError(error: unknown): boolean { + if (typeof error !== 'object' || error === null) return false; + return (error as { kind?: unknown }).kind === 'context_overflow'; +} + +export function isShrinkableSummaryError(error: unknown): boolean { + if (typeof error !== 'object' || error === null) return false; + const kind = (error as { kind?: unknown }).kind; + return kind === 'context_overflow' || kind === 'empty_response'; +} diff --git a/packages/agent-core-v2/src/human/compaction/machine.ts b/packages/agent-core-v2/src/human/compaction/machine.ts new file mode 100644 index 0000000000..8a9f7595a2 --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/machine.ts @@ -0,0 +1,498 @@ +import { estimateUsedContextTokens } from '#/agent/context-usage'; +import { + inputCancelled, + inputNotified, + inputReminded, + inputSteered, + inputSubmitted, +} from '#/agent/events'; +import type { QueuedPrompt } from '#/agent/slices'; +import type { HistoryMessage } from '#/agent/turn'; +import type { ExternalEvent } from '#/eventStore/events'; +import type { SystemMessage, UserMessage } from '#/llm/message'; +import type { AgentActorRef } from '#/session/machine'; +import type { SessionStores } from '#/session/stores'; +import { assign, emit, enqueueActions, fromPromise, setup, waitFor } from '#/xstate2'; + +import { CompactError } from './errors'; +import { buildCompactionSeed, compactionContinuationMessage } from './shape'; +import type { Summarize, SummaryOutcome } from './summarize'; + +export type CompactionReason = 'budget' | 'manual' | 'overflow'; + +export type CompactionPhase = + | 'idle' + | 'quiescing' + | 'summarizing' + | 'switching' + | 'resuming' + | 'completed' + | 'cancelled'; + +export type CompactionCancelCause = 'cancelled' | 'user-abort' | 'drift' | 'failed'; + +export type SummaryTelemetry = Omit; + +export interface CompactionStats { + compactedCount: number; + tokensBefore: number; + tokensAfter: number; +} + +export type CompactionEvent = + | { type: 'compaction.started'; reason: CompactionReason; instruction?: string } + | { type: 'compaction.blocked'; turnId?: number } + | { + type: 'compaction.completed'; + reason: CompactionReason; + branchId: string; + stats: CompactionStats; + durationMs: number; + originTurnId?: number; + summary?: SummaryTelemetry; + } + | { + type: 'compaction.cancelled'; + reason: CompactionReason; + cause: CompactionCancelCause; + error?: unknown; + durationMs: number; + originTurnId?: number; + tokensBefore?: number; + }; + +export interface CompactionMachineDeps { + agentId: string; + actor: AgentActorRef; + stores: SessionStores; + summarize: Summarize; + continuation?: (reason: CompactionReason) => UserMessage | undefined; + todos?: () => string | undefined; + onWillCompact?: (input: { + reason: CompactionReason; + instruction?: string; + signal: AbortSignal; + tokenCount: number; + }) => void | Promise; +} + +export interface CompactionMachineInput { + reason: CompactionReason; + instruction?: string; +} + +export type CompactionMachineOutput = + | { status: 'completed'; branchId: string; stats: CompactionStats } + | { status: 'cancelled'; cause: CompactionCancelCause; error: unknown }; + +type CompactionMachineEvent = { type: 'cancel'; cause: 'cancelled' | 'user-abort' }; + +interface QuiesceSnapshot { + history: HistoryMessage[]; + queue: QueuedPrompt[]; + nextTurnId: number; + branch: string; + head: number | null; + tokensBefore: number; +} + +interface SummaryResult { + seedEvents: ExternalEvent[]; + stats: CompactionStats; + telemetry: SummaryTelemetry; +} + +interface CompactionMachineContext { + input: CompactionMachineInput; + startedAt: number; + cause?: CompactionCancelCause; + error?: unknown; + originTurnId?: number; + snap?: QuiesceSnapshot; + seedEvents?: ExternalEvent[]; + stats?: CompactionStats; + summaryTelemetry?: SummaryTelemetry; + branchId?: string; +} + +const PAUSE_TIMEOUT_MS = 300_000; +const RESET_TIMEOUT_MS = 20_000; + +const INPUT_DELTA_TYPES: ReadonlySet = new Set([ + inputSubmitted.type, + inputNotified.type, + inputReminded.type, + inputCancelled.type, + inputSteered.type, +]); + +function aborted(signal: AbortSignal): Promise { + return new Promise((_, reject) => { + if (signal.aborted) { + reject(signal.reason as unknown); + return; + } + signal.addEventListener('abort', () => reject(signal.reason as unknown), { once: true }); + }); +} + +async function forEachDeltaEntry( + stores: SessionStores, + snapBranch: string, + snapHead: number | null, + visit: (type: string, data: Record) => void, +): Promise { + const branch = stores.tree.openBranch(snapBranch); + const head = branch.head; + if (head === null) return; + for (let seq = (snapHead ?? -1) + 1; seq <= head; seq++) { + const entry = branch.entryAt(seq); + if (entry === null) continue; + const data = (await stores.tree.resolve(entry)) as Record | null; + if (data === null) continue; + visit(entry.type, data); + } +} + +async function assertInputOnlyDelta( + stores: SessionStores, + snapBranch: string, + snapHead: number | null, +): Promise { + await forEachDeltaEntry(stores, snapBranch, snapHead, (type) => { + if (!INPUT_DELTA_TYPES.has(type)) { + throw new CompactError('drift', 'history changed during compaction; cancelled'); + } + }); +} + +async function replayInputDelta( + deps: CompactionMachineDeps, + snapBranch: string, + snapHead: number | null, +): Promise { + await forEachDeltaEntry(deps.stores, snapBranch, snapHead, (type, data) => { + if (type === inputSubmitted.type) { + deps.actor.send({ + type: 'input.submit', + id: data['id'] as string | undefined, + message: data['message'] as UserMessage, + }); + } else if (type === inputNotified.type) { + deps.actor.send({ type: 'input.notify', message: data['message'] as UserMessage }); + } else if (type === inputReminded.type) { + deps.actor.send({ + type: 'input.remind', + key: data['key'] as string, + message: data['message'] as UserMessage | SystemMessage, + }); + } else if (type === inputSteered.type) { + deps.actor.send({ type: 'input.steer', id: data['id'] as string }); + } else if (type === inputCancelled.type) { + deps.actor.send({ type: 'input.cancel', id: data['id'] as string }); + } + }); +} + +function waitForResetApplied(deps: CompactionMachineDeps, branchId: string): Promise { + if (deps.actor.getSnapshot().context.branchId === branchId) { + return Promise.resolve(); + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + subscription.unsubscribe(); + reject(new CompactError('reset-timeout', `machine did not apply branch '${branchId}' in time`)); + }, RESET_TIMEOUT_MS); + const subscription = deps.actor.on('context.reset', (event) => { + if (event.branchId !== branchId) return; + clearTimeout(timer); + subscription.unsubscribe(); + resolve(); + }); + }); +} + +export function createCompactionMachine(deps: CompactionMachineDeps) { + return setup({ + types: { + input: {} as CompactionMachineInput, + context: {} as CompactionMachineContext, + events: {} as CompactionMachineEvent, + emitted: {} as CompactionEvent, + output: {} as CompactionMachineOutput, + }, + actors: { + quiesce: fromPromise(async ({ signal }) => { + const store = deps.stores.get(deps.agentId); + if (store === undefined) { + throw new CompactError('unknown-agent', `unknown agent: '${deps.agentId}'`); + } + deps.actor.send({ type: 'input.pause' }); + const waiting = waitFor(deps.actor, (s) => s.matches('idle'), { timeout: PAUSE_TIMEOUT_MS }); + void waiting.catch(() => undefined); + await Promise.race([waiting, aborted(signal)]); + await store.flush(); + const state = store.getState(); + if (state.history.length === 0) { + throw new CompactError('insufficient', 'nothing to compact'); + } + return { + history: state.history, + queue: state.queue, + nextTurnId: state.turnIndex.nextTurnId, + branch: store.ref.branch, + head: deps.stores.tree.openBranch(store.ref.branch).head, + tokensBefore: estimateUsedContextTokens(state.history), + }; + }), + summarize: fromPromise< + SummaryResult, + { snap: QuiesceSnapshot; reason: CompactionReason; instruction?: string } + >(async ({ input, signal }) => { + await deps.onWillCompact?.({ + reason: input.reason, + instruction: input.instruction, + signal, + tokenCount: input.snap.tokensBefore, + }); + const outcome = await deps.summarize({ + history: input.snap.history, + instruction: input.instruction, + signal, + }); + let summary = outcome.text; + const todoText = deps.todos?.(); + if (todoText !== undefined && todoText.length > 0) { + summary = `${summary.trim()}\n\n${todoText}`; + } + if (signal.aborted) throw signal.reason; + const store = deps.stores.get(deps.agentId); + if (store === undefined || store.ref.branch !== input.snap.branch) { + throw new CompactError('drift', 'branch switched during compaction; cancelled'); + } + await assertInputOnlyDelta(deps.stores, input.snap.branch, input.snap.head); + const seed = buildCompactionSeed({ + turnId: input.snap.nextTurnId, + history: input.snap.history, + summary, + queue: input.snap.queue, + }); + return { + seedEvents: seed.events, + stats: { + compactedCount: input.snap.history.length, + tokensBefore: input.snap.tokensBefore, + tokensAfter: seed.tokensAfter, + }, + telemetry: { + usage: outcome.usage, + traceId: outcome.traceId, + attempts: outcome.attempts, + droppedCount: outcome.droppedCount, + }, + }; + }), + switchStore: fromPromise<{ branchId: string }, { seedEvents: ExternalEvent[]; stats: CompactionStats }>( + async ({ input }) => { + const { branchId } = await deps.stores.switchBranch(deps.agentId, { + reason: 'compaction', + stats: { + compactedCount: input.stats.compactedCount, + tokensBefore: input.stats.tokensBefore, + tokensAfter: input.stats.tokensAfter, + }, + seed: input.seedEvents, + }); + await waitForResetApplied(deps, branchId); + return { branchId }; + }, + ), + resume: fromPromise( + async ({ input }) => { + await replayInputDelta(deps, input.branch, input.head); + const continuation = (deps.continuation ?? defaultContinuation)(input.reason); + if (continuation !== undefined) { + deps.actor.send({ type: 'input.submit', message: continuation }); + } + deps.actor.send({ type: 'input.continue' }); + }, + ), + }, + }).createMachine({ + id: 'compaction', + initial: 'quiescing', + context: ({ input }) => ({ input, startedAt: Date.now() }), + on: { + cancel: {}, + }, + states: { + quiescing: { + entry: [ + emit(({ context }) => ({ + type: 'compaction.started' as const, + reason: context.input.reason, + instruction: context.input.instruction, + })), + assign({ + originTurnId: ({ context }) => + context.input.reason === 'manual' + ? undefined + : deps.actor.getSnapshot().context.activeTurnId, + }), + enqueueActions(({ enqueue }) => { + const snapshot = deps.actor.getSnapshot(); + if (!snapshot.matches('idle')) { + enqueue.emit({ + type: 'compaction.blocked', + turnId: snapshot.context.activeTurnId, + }); + } + }), + ], + invoke: { + src: 'quiesce', + onDone: { + target: 'summarizing', + actions: assign({ snap: ({ event }) => event.output }), + }, + onError: { + target: 'cancelled', + actions: assign(({ event }) => ({ + cause: (event.error instanceof CompactError && event.error.code === 'drift' + ? 'drift' + : 'failed') as CompactionCancelCause, + error: event.error, + })), + }, + }, + on: { + cancel: { + target: 'cancelled', + actions: assign(({ event }) => ({ + cause: event.cause as CompactionCancelCause, + error: cancelError(event.cause), + })), + }, + }, + }, + summarizing: { + invoke: { + src: 'summarize', + input: ({ context }) => ({ + snap: context.snap as QuiesceSnapshot, + reason: context.input.reason, + instruction: context.input.instruction, + }), + onDone: { + target: 'switching', + actions: assign({ + seedEvents: ({ event }) => event.output.seedEvents, + stats: ({ event }) => event.output.stats, + summaryTelemetry: ({ event }) => event.output.telemetry, + }), + }, + onError: { + target: 'cancelled', + actions: assign(({ event }) => ({ + cause: (event.error instanceof CompactError && event.error.code === 'drift' + ? 'drift' + : 'failed') as CompactionCancelCause, + error: event.error, + })), + }, + }, + on: { + cancel: { + target: 'cancelled', + actions: assign(({ event }) => ({ + cause: event.cause as CompactionCancelCause, + error: cancelError(event.cause), + })), + }, + }, + }, + switching: { + invoke: { + src: 'switchStore', + input: ({ context }) => ({ + seedEvents: context.seedEvents as ExternalEvent[], + stats: context.stats as CompactionStats, + }), + onDone: { + target: 'resuming', + actions: assign({ branchId: ({ event }) => event.output.branchId }), + }, + onError: { + target: 'cancelled', + actions: assign({ cause: 'failed' as CompactionCancelCause, error: ({ event }) => event.error }), + }, + }, + }, + resuming: { + invoke: { + src: 'resume', + input: ({ context }) => ({ + branch: (context.snap as QuiesceSnapshot).branch, + head: (context.snap as QuiesceSnapshot).head, + reason: context.input.reason, + }), + onDone: { target: 'completed' }, + onError: { + target: 'cancelled', + actions: assign({ cause: 'failed' as CompactionCancelCause, error: ({ event }) => event.error }), + }, + }, + }, + completed: { + type: 'final', + entry: emit(({ context }) => ({ + type: 'compaction.completed' as const, + reason: context.input.reason, + branchId: context.branchId as string, + stats: context.stats as CompactionStats, + durationMs: Date.now() - context.startedAt, + originTurnId: context.originTurnId, + summary: context.summaryTelemetry, + })), + }, + cancelled: { + type: 'final', + entry: [ + ({ context }) => { + if (context.cause !== 'user-abort') { + deps.actor.send({ type: 'input.continue' }); + } + }, + emit(({ context }) => ({ + type: 'compaction.cancelled' as const, + reason: context.input.reason, + cause: context.cause as CompactionCancelCause, + error: context.cause === 'failed' ? context.error : undefined, + durationMs: Date.now() - context.startedAt, + originTurnId: context.originTurnId, + tokensBefore: context.snap?.tokensBefore, + })), + ], + }, + }, + output: ({ context }): CompactionMachineOutput => + context.cause === undefined + ? { + status: 'completed', + branchId: context.branchId as string, + stats: context.stats as CompactionStats, + } + : { status: 'cancelled', cause: context.cause, error: context.error }, + }); +} + +function cancelError(cause: 'cancelled' | 'user-abort'): CompactError { + return cause === 'cancelled' + ? new CompactError('cancelled', 'compaction was cancelled') + : new CompactError('aborted', 'compaction cancelled by user abort'); +} + +function defaultContinuation(reason: CompactionReason): UserMessage | undefined { + if (reason === 'manual') return undefined; + return compactionContinuationMessage(); +} diff --git a/packages/agent-core-v2/src/human/compaction/shape.ts b/packages/agent-core-v2/src/human/compaction/shape.ts new file mode 100644 index 0000000000..28ad91ec69 --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/shape.ts @@ -0,0 +1,205 @@ +import { estimateMessageTokens, estimateUsedContextTokens } from '#/agent/context-usage'; +import { inputSubmitted, messageAppended, turnEnded, turnStarted } from '#/agent/events'; +import type { QueuedPrompt } from '#/agent/slices'; +import { createUserEntry, type HistoryMessage, type UserEntry } from '#/agent/turn'; +import type { ExternalEvent } from '#/eventStore/events'; +import { createUserMessage, type UserMessage } from '#/llm/message'; + +import summaryPrefixTemplate from './compaction-summary-prefix.md?raw'; + +const COMPACTION_SUMMARY_PREFIX = summaryPrefixTemplate.trimEnd(); +const COMPACT_USER_MESSAGE_MAX_TOKENS = 20_000; +const COMPACT_USER_MESSAGE_HEAD_TOKENS = 2_000; + +export interface CompactionSeed { + events: ExternalEvent[]; + tokensAfter: number; + keptUserMessageCount: number; + keptHeadUserMessageCount?: number; +} + +interface CompactionUserSelection { + head: UserEntry[]; + tail: UserEntry[]; + elided: boolean; + omittedTokens: number; +} + +export function buildCompactionSeed(input: { + turnId: number; + history: readonly HistoryMessage[]; + summary: string; + queue: readonly QueuedPrompt[]; +}): CompactionSeed { + const compactable = input.history.filter(isKeptUserEntry); + const selection = selectCompactionUserMessages( + compactable, + COMPACT_USER_MESSAGE_MAX_TOKENS, + COMPACT_USER_MESSAGE_HEAD_TOKENS, + ); + const elision = selection.elided + ? createUserEntry(createUserMessage(elisionText(selection.omittedTokens)), { + source: 'compaction', + key: 'elision', + }) + : undefined; + const summaryEntry = createUserEntry(createUserMessage(summaryText(input.summary)), { + source: 'compaction', + key: 'summary', + }); + const kept: HistoryMessage[] = [ + ...selection.head, + ...(elision === undefined ? [] : [elision]), + ...selection.tail, + ]; + const seeded = [...kept, summaryEntry]; + const events: ExternalEvent[] = [ + turnStarted({ turnId: input.turnId }), + ...seeded.map((message) => messageAppended({ message })), + turnEnded({ turnId: input.turnId, outcome: 'done' }), + ...input.queue.map((item) => inputSubmitted({ id: item.id, message: item.message })), + ]; + return { + events, + tokensAfter: estimateUsedContextTokens(seeded), + keptUserMessageCount: selection.head.length + selection.tail.length, + keptHeadUserMessageCount: selection.elided ? selection.head.length : undefined, + }; +} + +export function compactionContinuationMessage(): UserMessage { + return createUserMessage( + wrapSystemReminder( + 'Context compaction is complete — continue the work that was in progress when it began.', + ), + ); +} + +function summaryText(summary: string): string { + const trimmed = summary.trim(); + return `${COMPACTION_SUMMARY_PREFIX}\n${trimmed.length > 0 ? trimmed : '(no summary available)'}`; +} + +function elisionText(omittedTokens: number): string { + return wrapSystemReminder( + `Some of this conversation's user messages were omitted here during compaction: the messages above this note are the oldest user input, the messages below are the most recent, and roughly ${String(omittedTokens)} tokens in between were dropped. The omitted content is covered by the compaction summary at the end of the conversation.`, + ); +} + +function wrapSystemReminder(content: string): string { + return `\n${content.trim()}\n`; +} + +function isKeptUserEntry(entry: HistoryMessage): entry is UserEntry { + if (entry.message.role !== 'user') return false; + if (entry.meta.source === 'compaction') return false; + return entry.meta.source === undefined || entry.meta.source === 'input'; +} + +function selectCompactionUserMessages( + messages: readonly UserEntry[], + maxTokens: number, + headTokens: number, +): CompactionUserSelection { + let totalTokens = 0; + for (const entry of messages) { + totalTokens += estimateMessageTokens(entry.message); + } + if (totalTokens <= maxTokens) { + return { head: [], tail: [...messages], elided: false, omittedTokens: 0 }; + } + + const headBudget = Math.min(Math.max(headTokens, 0), maxTokens); + const tail: UserEntry[] = []; + let tailRemaining = maxTokens - headBudget; + let headEndExclusive = messages.length; + let tailBoundaryDroppedPrefix: UserEntry | null = null; + for (let i = messages.length - 1; i >= 0 && tailRemaining > 0; i--) { + const entry = messages[i] as UserEntry; + const tokens = estimateMessageTokens(entry.message); + if (tokens <= tailRemaining) { + tail.push(entry); + tailRemaining -= tokens; + headEndExclusive = i; + continue; + } + const fullText = textOf(entry.message); + const keptSuffix = truncateTextToTokensFromEnd(fullText, tailRemaining); + tail.push(replaceEntryText(entry, keptSuffix)); + headEndExclusive = i; + const droppedPrefix = fullText.slice(0, fullText.length - keptSuffix.length); + if (droppedPrefix.length > 0) { + tailBoundaryDroppedPrefix = replaceEntryText(entry, droppedPrefix); + } + break; + } + tail.reverse(); + + const headCandidates = messages.slice(0, headEndExclusive); + if (tailBoundaryDroppedPrefix !== null) { + headCandidates.push(tailBoundaryDroppedPrefix); + } + const head: UserEntry[] = []; + let headRemaining = headBudget; + for (const entry of headCandidates) { + if (headRemaining <= 0) break; + const tokens = estimateMessageTokens(entry.message); + if (tokens <= headRemaining) { + head.push(entry); + headRemaining -= tokens; + continue; + } + head.push(replaceEntryText(entry, truncateTextToTokens(textOf(entry.message), headRemaining))); + break; + } + + let keptTokens = 0; + for (const entry of head) keptTokens += estimateMessageTokens(entry.message); + for (const entry of tail) keptTokens += estimateMessageTokens(entry.message); + return { head, tail, elided: true, omittedTokens: Math.max(0, totalTokens - keptTokens) }; +} + +function textOf(message: UserMessage): string { + let text = ''; + for (const part of message.content) { + if (part.type === 'text') { + text += part.text; + } + } + return text; +} + +function replaceEntryText(entry: UserEntry, text: string): UserEntry { + return { ...entry, message: { ...entry.message, content: [{ type: 'text', text }] } }; +} + +function truncateTextToTokens(text: string, maxTokens: number): string { + if (maxTokens <= 0) return ''; + let asciiCount = 0; + let nonAsciiCount = 0; + let end = 0; + for (const char of text) { + if ((char.codePointAt(0) as number) <= 127) { + asciiCount++; + } else { + nonAsciiCount++; + } + if (Math.ceil(asciiCount / 4) + nonAsciiCount > maxTokens) break; + end += char.length; + } + return text.slice(0, end); +} + +function truncateTextToTokensFromEnd(text: string, maxTokens: number): string { + if (maxTokens <= 0) return ''; + const chars = Array.from(text); + let tokens = 0; + let start = chars.length; + for (let i = chars.length - 1; i >= 0; i--) { + const code = chars[i]?.codePointAt(0) ?? 0; + tokens += code <= 127 ? 0.25 : 1; + if (Math.ceil(tokens) > maxTokens) break; + start = i; + } + return chars.slice(start).join(''); +} diff --git a/packages/agent-core-v2/src/human/compaction/summarize.ts b/packages/agent-core-v2/src/human/compaction/summarize.ts new file mode 100644 index 0000000000..030976977f --- /dev/null +++ b/packages/agent-core-v2/src/human/compaction/summarize.ts @@ -0,0 +1,126 @@ +import { + createUserEntry, + type AssistantEntry, + type createTurnMachine, + type HistoryMessage, + type TurnOutput, +} from '#/agent/turn'; +import { createUserMessage, extractText } from '#/llm/message'; +import type { LlmRequestConfig } from '#/llm/requester/requester'; +import type { TokenUsage } from '#/llm/usage'; +import { createActor, waitFor } from '#/xstate2'; + +import instructionTemplate from './compaction-instruction.md?raw'; +import { CompactError, isShrinkableSummaryError } from './errors'; + +export interface SummaryOutcome { + text: string; + usage?: TokenUsage; + traceId?: string; + attempts: number; + droppedCount: number; +} + +export type Summarize = (input: { + history: readonly HistoryMessage[]; + instruction?: string; + signal: AbortSignal; +}) => Promise; + +export interface CreateSummarizeOptions { + request: LlmRequestConfig; + llm: () => ReturnType; + maxShrinkAttempts?: number; + timeoutMs?: number; +} + +export function createSummarize(options: CreateSummarizeOptions): Summarize { + const maxShrinkAttempts = options.maxShrinkAttempts ?? 3; + return async ({ history, instruction, signal }) => { + const instructionEntry = createUserEntry( + createUserMessage(compactionInstructionText(instruction)), + { source: 'input' }, + ); + let attemptHistory = [...history]; + for (let attempt = 0; ; attempt++) { + if (signal.aborted) { + throw new CompactError('aborted', 'compaction was aborted'); + } + const output = await runSummaryTurn(options, [...attemptHistory, instructionEntry], signal); + if (output.type === 'done') { + const entry = lastAssistantEntry(output.produced); + const text = entry === undefined ? undefined : extractText(entry.message); + if (entry !== undefined && text !== undefined && text.trim().length > 0) { + return { + text, + usage: entry.meta.usage, + traceId: entry.meta.headers?.['x-trace-id'], + attempts: attempt + 1, + droppedCount: history.length - attemptHistory.length, + }; + } + } + if (output.type === 'aborted') { + throw new CompactError('aborted', 'summary turn was aborted'); + } + const error = output.type === 'failed' ? output.error : undefined; + if ( + attempt + 1 >= maxShrinkAttempts || + attemptHistory.length <= 1 || + (error !== undefined && !isShrinkableSummaryError(error)) + ) { + throw new CompactError( + 'summary-failed', + `summary turn failed: ${error instanceof Error ? error.message : String(error)}`, + ); + } + attemptHistory = dropOldestAndLeadingToolResults(attemptHistory); + } + }; +} + +async function runSummaryTurn( + options: CreateSummarizeOptions, + history: readonly HistoryMessage[], + signal: AbortSignal, +): Promise { + const actor = createActor(options.llm(), { + input: { request: options.request, history, parentSignal: signal }, + }); + actor.start(); + try { + const snapshot = await waitFor(actor, (s) => s.status !== 'active', { + timeout: options.timeoutMs ?? 120_000, + }); + return snapshot.output as TurnOutput; + } finally { + actor.stop(); + } +} + +function lastAssistantEntry(produced: readonly HistoryMessage[]): AssistantEntry | undefined { + for (let i = produced.length - 1; i >= 0; i--) { + const entry = produced[i]; + if (entry !== undefined && entry.message.role === 'assistant') { + return entry as AssistantEntry; + } + } + return undefined; +} + +function compactionInstructionText(customInstruction?: string): string { + const custom = customInstruction?.trim() ?? ''; + const block = custom.length > 0 ? `\nOptional user instruction:\n${custom}\n` : ''; + return instructionTemplate.replace('${custom_instruction_block}', () => block).trimEnd(); +} + +function dropOldestAndLeadingToolResults( + history: readonly HistoryMessage[], +): HistoryMessage[] { + const rest = history.slice(1); + let start = 0; + while (start < rest.length && rest[start]?.message.role === 'tool') { + start++; + } + return rest.slice(start); +} diff --git a/packages/agent-core-v2/src/human/index.ts b/packages/agent-core-v2/src/human/index.ts index 79e1b9c636..c05269a2cc 100644 --- a/packages/agent-core-v2/src/human/index.ts +++ b/packages/agent-core-v2/src/human/index.ts @@ -70,6 +70,10 @@ export * from './session/machine'; export * from './session/events'; export * from './session/slices'; export * from './session/stores'; +export * from './compaction/controller'; +export * from './compaction/errors'; +export * from './compaction/shape'; +export * from './compaction/summarize'; export * from './usage/usage'; export * from './usage/machine'; export * from './usage/plugin'; diff --git a/packages/agent-core-v2/src/human/session/events.ts b/packages/agent-core-v2/src/human/session/events.ts index a40131d205..433c4a5e8d 100644 --- a/packages/agent-core-v2/src/human/session/events.ts +++ b/packages/agent-core-v2/src/human/session/events.ts @@ -18,7 +18,12 @@ export type AgentClosed = ReturnType; export const agentSwitched = defineEvent({ type: 'agent.switched', - schema: z.object({ agentId: z.string(), branch: z.string(), reason: z.string().optional() }), + schema: z.object({ + agentId: z.string(), + branch: z.string(), + reason: z.string().optional(), + stats: z.record(z.string(), z.number()).optional(), + }), }); export type AgentSwitched = ReturnType; @@ -27,3 +32,29 @@ export const sessionMetaUpdated = defineEvent({ schema: z.object({ meta: z.unknown() }), }); export type SessionMetaUpdated = ReturnType; + +export const compactionStarted = defineEvent({ + type: 'compaction.started', + schema: z.object({ + agentId: z.string(), + reason: z.string(), + instruction: z.string().optional(), + }), +}); +export type CompactionStarted = ReturnType; + +export const compactionCompleted = defineEvent({ + type: 'compaction.completed', + schema: z.object({ agentId: z.string(), branch: z.string() }), +}); +export type CompactionCompleted = ReturnType; + +export const compactionCancelled = defineEvent({ + type: 'compaction.cancelled', + schema: z.object({ + agentId: z.string(), + cause: z.string(), + errorMessage: z.string().optional(), + }), +}); +export type CompactionCancelled = ReturnType; diff --git a/packages/agent-core-v2/src/human/session/stores.ts b/packages/agent-core-v2/src/human/session/stores.ts index aa3e7da82a..d998d3c5db 100644 --- a/packages/agent-core-v2/src/human/session/stores.ts +++ b/packages/agent-core-v2/src/human/session/stores.ts @@ -1,4 +1,5 @@ import { createEventStore, type EventStore } from '#/eventStore/eventStore'; +import type { ExternalEvent } from '#/eventStore/events'; import { journalFromBranch } from '#/eventStore/journal'; import { agentSlices, type AgentEventStore } from '#/agent/slices'; import type { StoreBackend } from '#/store/backend/backend'; @@ -123,11 +124,11 @@ export class SessionStores { throw new UndoError('insufficient', `cannot undo ${turns} turn(s): not enough turns`); } const from = undoForkRef(this.tree, cut.start); + if (from === undefined) { + throw new UndoError('insufficient', `cannot undo ${turns} turn(s): no earlier history`); + } const branchId = freshBranchName(this.tree, agentId); - const branch = - from === undefined - ? this.tree.createBranch(branchId) - : this.tree.createBranch(branchId, { from }); + const branch = this.tree.createBranch(branchId, { from }); await store.reset(journalFromBranch(branch, this.tree)); await ( await this.session() @@ -135,6 +136,31 @@ export class SessionStores { return { branchId }; } + async switchBranch( + agentId: string, + opts: { reason: string; stats?: Record; seed: readonly ExternalEvent[] }, + ): Promise<{ branchId: string }> { + const store = this.agents.get(agentId); + if (store === undefined) { + throw new StoreError('unknown-agent', `unknown agent '${agentId}'`); + } + const branchId = freshBranchName(this.tree, agentId); + const branch = this.tree.createBranch(branchId); + const journal = journalFromBranch(branch, this.tree); + const seedStore = await createEventStore({ journal, slices: agentSlices }); + try { + await seedStore.dispatch([...opts.seed]); + await seedStore.flush(); + } finally { + await seedStore.close(); + } + await store.reset(journal); + await (await this.session()).dispatch( + agentSwitched({ agentId, branch: branchId, reason: opts.reason, stats: opts.stats }), + ); + return { branchId }; + } + async flush(): Promise { await Promise.all([...this.agents.values()].map((store) => store.flush())); await this.sessionStore?.flush(); diff --git a/packages/agent-core-v2/src/human/test/agent/machine.test.ts b/packages/agent-core-v2/src/human/test/agent/machine.test.ts index f8fa34b7b9..25827260ee 100644 --- a/packages/agent-core-v2/src/human/test/agent/machine.test.ts +++ b/packages/agent-core-v2/src/human/test/agent/machine.test.ts @@ -1114,6 +1114,7 @@ describe('agent machine input.abort', () => { it('aborts running turn tools and completes the transcript with aborted tool messages', async () => { const requester = createStubRequester([ createAssistantMessage([], [toolCall('call-1', 'slow_tool')]), + createAssistantMessage([{ type: 'text', text: 'resumed' }], []), ]); const signals: AbortSignal[] = []; const tools = stubTools(({ signal }) => { @@ -1146,6 +1147,20 @@ describe('agent machine input.abort', () => { 'assistant:', 'tool:aborted', ]); + + actor.send({ type: 'input.continue' }); + await waitFor( + actor, + (s) => s.matches('idle') && store.getState().history.length === 4, + { timeout: 5000 }, + ); + expect(rolesAndTexts(store.getState().history)).toEqual([ + 'user:hi', + 'assistant:', + 'tool:aborted', + 'assistant:resumed', + ]); + expect(store.getState().turnIndex.nextTurnId).toBe(2); }); it('waits for the real outcome of a tool that settles after the abort signal', async () => { @@ -1355,6 +1370,131 @@ describe('agent machine input.abort', () => { }); }); +describe('agent machine input.pause/input.continue', () => { + it('gates queue drain while paused and resumes on continue', async () => { + const requester = createStubRequester([ + createAssistantMessage([{ type: 'text', text: 'hi there' }], []), + ]); + const store = await testStore(); + const actor = createActor(createTestAgentMachine([], requester), { + input: { request: { model }, store }, + }); + actor.start(); + actor.send({ type: 'input.pause' }); + actor.send({ type: 'input.submit', message: createUserMessage('hi') }); + + await vi.waitFor(() => { + expect(store.getState().queue).toHaveLength(1); + }); + expect(actor.getSnapshot().matches('running')).toBe(false); + + actor.send({ type: 'input.continue' }); + await waitFor( + actor, + (s) => s.matches('idle') && store.getState().history.length === 2, + { timeout: 5000 }, + ); + expect(rolesAndTexts(store.getState().history)).toEqual(['user:hi', 'assistant:hi there']); + expect(actor.getSnapshot().context.paused).toBe(false); + actor.stop(); + }); + + it('ends the turn at the acting boundary when paused and resumes with a new turn on continue', async () => { + let calls = 0; + const base = createStubRequester([ + createAssistantMessage([], [toolCall('call-1', 'fast_tool')]), + createAssistantMessage([{ type: 'text', text: 'final' }], []), + ]); + const requester: LlmRequester = { + generate: (config, content, control) => { + calls += 1; + return base.generate(config, content, control); + }, + }; + let releaseTool: (() => void) | undefined; + const tools = stubTools( + () => + new Promise((resolve) => { + releaseTool = () => resolve({ content: [{ type: 'text', text: 'tool result' }] }); + }), + 'fast_tool', + ); + const store = await testStore(); + const actor = createActor(createTestAgentMachine(tools, requester), { + input: { request: { model }, store }, + }); + actor.start(); + actor.send({ type: 'input.submit', message: createUserMessage('hi') }); + + await vi.waitFor(() => { + expect(actor.getSnapshot().context.turnTools['call-1']).toBeDefined(); + }); + actor.send({ type: 'input.pause' }); + (releaseTool as () => void)(); + + await waitFor( + actor, + (s) => s.matches('idle') && store.getState().history.length === 3, + { timeout: 5000 }, + ); + expect(rolesAndTexts(store.getState().history)).toEqual([ + 'user:hi', + 'assistant:', + 'tool:tool result', + ]); + expect(calls).toBe(1); + expect(store.getState().turnIndex.nextTurnId).toBe(1); + + actor.send({ type: 'input.continue' }); + await waitFor( + actor, + (s) => s.matches('idle') && store.getState().history.length === 4, + { timeout: 5000 }, + ); + expect(rolesAndTexts(store.getState().history)).toEqual([ + 'user:hi', + 'assistant:', + 'tool:tool result', + 'assistant:final', + ]); + expect(calls).toBe(2); + expect(store.getState().turnIndex.nextTurnId).toBe(2); + actor.stop(); + }); + + it('does not start a turn on continue when history ends with a plain assistant message', async () => { + let calls = 0; + const base = createStubRequester([ + createAssistantMessage([{ type: 'text', text: 'done' }], []), + ]); + const requester: LlmRequester = { + generate: (config, content, control) => { + calls += 1; + return base.generate(config, content, control); + }, + }; + const store = await testStore(); + const actor = createActor(createTestAgentMachine([], requester), { + input: { request: { model }, store }, + }); + actor.start(); + actor.send({ type: 'input.submit', message: createUserMessage('hi') }); + await waitFor( + actor, + (s) => s.matches('idle') && store.getState().history.length === 2, + { timeout: 5000 }, + ); + + actor.send({ type: 'input.continue' }); + await new Promise((resolve) => setTimeout(resolve, 150)); + + expect(actor.getSnapshot().matches('idle')).toBe(true); + expect(store.getState().history).toHaveLength(2); + expect(calls).toBe(1); + actor.stop(); + }); +}); + describe('agent machine max steps', () => { it('resets the step budget on drained input and fails only on pure tool-call continuation', async () => { let call = 0; diff --git a/packages/agent-core-v2/src/human/test/compaction/controller.test.ts b/packages/agent-core-v2/src/human/test/compaction/controller.test.ts new file mode 100644 index 0000000000..5476f8d2a5 --- /dev/null +++ b/packages/agent-core-v2/src/human/test/compaction/controller.test.ts @@ -0,0 +1,541 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { createAgentMachine, type AgentMachineContext } from '#/agent/machine'; +import { messageAppended } from '#/agent/events'; +import type { AgentEventStore } from '#/agent/slices'; +import { createTurnMachine, createUserEntry, type TurnBeforeStep } from '#/agent/turn'; +import { createCompactionController, type CompactionEvent } from '#/compaction/controller'; +import type { Summarize, SummaryOutcome } from '#/compaction/summarize'; +import { UNKNOWN_CAPABILITY } from '#/llm/capability'; +import { createUserMessage, extractText } from '#/llm/message'; +import type { LlmModel } from '#/llm/model'; +import { createLlmMachine } from '#/llm/requester/machine'; +import type { LlmRequestConfig, LlmRequester } from '#/llm/requester/requester'; +import { SessionStores } from '#/session/stores'; +import { MemoryBackend } from '#/store/backend/memory'; +import { TreeStore } from '#/store/store'; +import type { Tree } from '#/store/tree'; +import { createActor, waitFor, type ActorRefFrom } from '#/xstate2'; + +const model: LlmModel = { provider: 'test', model: 'test-model', capability: UNKNOWN_CAPABILITY }; + +type AgentActor = ActorRefFrom>; + +interface TestEnv { + backend: MemoryBackend; + tree: Tree; + stores: SessionStores; +} + +async function testEnv(): Promise { + const backend = new MemoryBackend(); + const store = await TreeStore.open(backend, {}); + const tree = await store.tree('sess'); + return { backend, tree, stores: new SessionStores(tree, backend) }; +} + +interface BeforeStepHook { + current?: TurnBeforeStep; +} + +function startAgent( + store: AgentEventStore, + requester: LlmRequester, + beforeStep?: BeforeStepHook, + request?: Partial, +): AgentActor { + const actor = createActor( + createAgentMachine({ + tools: [], + turnActor: createTurnMachine(createLlmMachine({ requester }), { + onBeforeStep: (context) => beforeStep?.current?.(context), + }), + }), + { input: { request: { model, ...request }, store } }, + ); + actor.start(); + return actor; +} + +function createEchoRequester(): LlmRequester { + return { + generate: (_config, { messages }, { onEvent }) => { + const last = messages.at(-1); + const text = last !== undefined && last.role === 'user' ? extractText(last) : ''; + onEvent?.({ type: 'llm.streaming.part', part: { type: 'text', text: `echo:${text}` } }); + onEvent?.({ type: 'llm.done' }); + return Promise.resolve(); + }, + }; +} + +function createOverflowRequester(): LlmRequester { + return { + generate: (_config, _content, { onEvent }) => { + onEvent?.({ + type: 'llm.failed.remote', + error: { + kind: 'context_overflow', + message: 'maximum context length exceeded', + statusCode: 400, + requestId: null, + retryAfterMs: null, + headers: null, + }, + }); + return Promise.resolve(); + }, + }; +} + +interface ControllerHarness { + controller: ReturnType; + events: CompactionEvent[]; + summarizeCalls: { historyLength: number; instruction?: string }[]; +} + +function startController( + env: TestEnv, + actor: AgentActor, + overrides?: Partial[0]>, +): ControllerHarness { + const events: CompactionEvent[] = []; + const summarizeCalls: { historyLength: number; instruction?: string }[] = []; + const summarize: Summarize = async ({ history, instruction }) => { + summarizeCalls.push({ historyLength: history.length, instruction }); + return { text: 'SUMMARY TEXT', attempts: 1, droppedCount: 0 }; + }; + const controller = createCompactionController({ + agentId: 'main', + actor, + stores: env.stores, + summarize, + budget: { maxContextTokens: () => 2000, triggerRatio: 0.85 }, + onEvent: (event) => events.push(event), + ...overrides, + }); + return { controller, events, summarizeCalls }; +} + +function historyTexts(store: AgentEventStore): string[] { + return store.getState().history.map((entry) => extractText(entry.message)); +} + +describe('compaction controller manual', () => { + it('compacts an idle agent onto a fresh branch and blocks undo across the switch', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const actor = startAgent(main, createEchoRequester()); + actor.send({ type: 'input.submit', message: createUserMessage('first') }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 2, { + timeout: 5000, + }); + const willCompactInputs: { tokenCount: number }[] = []; + const harness = startController(env, actor, { + onWillCompact: (input) => { + willCompactInputs.push(input); + }, + }); + + const result = await harness.controller.compact(); + + expect(result.branchId).toBe('main~2'); + expect(main.ref.branch).toBe('main~2'); + const texts = historyTexts(main); + expect(texts).toHaveLength(2); + expect(texts[0]).toBe('first'); + expect(texts[1]).toContain('SUMMARY TEXT'); + expect(main.getState().turnIndex.turns).toHaveLength(1); + expect(main.getState().turnIndex.nextTurnId).toBe(2); + expect(env.tree.openBranch('main~2').header.parentBranch).toBeUndefined(); + expect((await env.stores.session()).getState().roster.agents['main']).toBe('main~2'); + await expect(env.stores.undo('main', 1)).rejects.toMatchObject({ reason: 'insufficient' }); + const sessionBranch = env.tree.openBranch('_session'); + const sessionTypes: string[] = []; + for (let seq = 0; seq <= (sessionBranch.head ?? -1); seq++) { + const entry = sessionBranch.entryAt(seq); + if (entry !== null) sessionTypes.push(entry.type); + } + expect(sessionTypes).toEqual([ + 'agent.opened', + 'compaction.started', + 'agent.switched', + 'compaction.completed', + ]); + expect(harness.controller.status()).toEqual({ phase: 'idle' }); + expect(harness.summarizeCalls).toEqual([{ historyLength: 2, instruction: undefined }]); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.completed', + ]); + const completed = harness.events.at(-1); + expect(completed?.type === 'compaction.completed' && completed.reason === 'manual').toBe(true); + expect( + completed?.type === 'compaction.completed' && + completed.originTurnId === undefined && + completed.summary?.attempts === 1 && + completed.summary?.droppedCount === 0, + ).toBe(true); + expect(willCompactInputs).toHaveLength(1); + expect(willCompactInputs[0]?.tokenCount).toBeGreaterThan(0); + expect(actor.getSnapshot().matches('idle')).toBe(true); + expect((actor.getSnapshot().context as AgentMachineContext).messages).toHaveLength(2); + await env.stores.flush(); + expect(main.getState().history).toHaveLength(2); + + harness.controller.dispose(); + actor.stop(); + }); + + it('pauses a running turn at the step boundary and preserves queued inputs through the switch', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + let release: (() => void) | undefined; + let first = true; + const requester: LlmRequester = { + generate: (_config, { messages }, { onEvent }) => { + const last = messages.at(-1); + const text = last !== undefined && last.role === 'user' ? extractText(last) : ''; + const respond = (): void => { + onEvent?.({ type: 'llm.streaming.part', part: { type: 'text', text: `echo:${text}` } }); + onEvent?.({ type: 'llm.done' }); + }; + if (!first) { + respond(); + return Promise.resolve(); + } + first = false; + return new Promise((resolve) => { + release = () => { + respond(); + resolve(); + }; + }); + }, + }; + const actor = startAgent(main, requester); + actor.send({ type: 'input.submit', message: createUserMessage('first') }); + await waitFor(actor, (s) => s.matches('running'), { timeout: 5000 }); + actor.send({ type: 'input.submit', message: createUserMessage('q1') }); + actor.send({ type: 'input.submit', message: createUserMessage('q2') }); + await vi.waitFor(() => expect(main.getState().queue).toHaveLength(2), { timeout: 5000 }); + const harness = startController(env, actor); + + const compactPromise = harness.controller.compact(); + await vi.waitFor(() => expect(actor.getSnapshot().context.paused).toBe(true), { timeout: 5000 }); + (release as () => void)(); + + const result = await compactPromise; + expect(result.branchId).toBe('main~2'); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 6, { + timeout: 5000, + }); + + const texts = historyTexts(main); + expect(texts[0]).toBe('first'); + expect(texts[1]).toContain('SUMMARY TEXT'); + expect(texts.slice(2)).toEqual(['q1', 'echo:q1', 'q2', 'echo:q2']); + expect(main.getState().turnIndex.nextTurnId).toBe(4); + expect(harness.summarizeCalls[0]?.historyLength).toBe(2); + + harness.controller.dispose(); + actor.stop(); + }); + + it('merges inputs submitted and steered during summarization into the new branch', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const actor = startAgent(main, createEchoRequester()); + actor.send({ type: 'input.submit', message: createUserMessage('first') }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 2, { + timeout: 5000, + }); + let resolveSummary: ((outcome: SummaryOutcome) => void) | undefined; + let summaryCalled = false; + const summarize: Summarize = () => { + summaryCalled = true; + return new Promise((resolve) => { + resolveSummary = resolve; + }); + }; + const harness = startController(env, actor, { summarize }); + + const compactPromise = harness.controller.compact(); + await vi.waitFor(() => expect(summaryCalled).toBe(true), { timeout: 5000 }); + expect(harness.controller.status().phase).toBe('summarizing'); + actor.send({ type: 'input.submit', id: 's1', message: createUserMessage('late') }); + actor.send({ type: 'input.submit', message: createUserMessage('queued') }); + await vi.waitFor(() => expect(main.getState().queue).toHaveLength(2), { timeout: 5000 }); + actor.send({ type: 'input.steer', id: 's1' }); + await vi.waitFor(() => expect(main.getState().notifications).toHaveLength(1), { timeout: 5000 }); + (resolveSummary as (outcome: SummaryOutcome) => void)({ + text: 'MERGED SUMMARY', + attempts: 1, + droppedCount: 0, + }); + + const result = await compactPromise; + expect(result.branchId).toBe('main~2'); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 5, { + timeout: 5000, + }); + + expect(historyTexts(main)).toEqual([ + 'first', + expect.stringContaining('MERGED SUMMARY'), + 'late', + 'queued', + 'echo:queued', + ]); + expect(main.getState().turnIndex.nextTurnId).toBe(3); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.completed', + ]); + + harness.controller.dispose(); + actor.stop(); + }); + + it('cancels an in-flight compaction via cancel() and releases the pause', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const actor = startAgent(main, createEchoRequester()); + actor.send({ type: 'input.submit', message: createUserMessage('first') }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 2, { + timeout: 5000, + }); + const summarize: Summarize = ({ signal }) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + }); + const harness = startController(env, actor, { summarize }); + + const compactPromise = harness.controller.compact(); + await vi.waitFor(() => expect(harness.controller.status().phase).toBe('summarizing'), { + timeout: 5000, + }); + actor.send({ type: 'input.submit', message: createUserMessage('while-compacting') }); + harness.controller.cancel(); + + await expect(compactPromise).rejects.toMatchObject({ code: 'cancelled' }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 4, { + timeout: 5000, + }); + expect(main.ref.branch).toBe('main'); + expect(env.tree.has('main~2')).toBe(false); + expect(historyTexts(main)).toEqual([ + 'first', + 'echo:first', + 'while-compacting', + 'echo:while-compacting', + ]); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.cancelled', + ]); + const cancelled = harness.events.at(-1); + expect(cancelled?.type === 'compaction.cancelled' && cancelled.cause === 'cancelled').toBe(true); + expect( + cancelled?.type === 'compaction.cancelled' && + typeof cancelled.tokensBefore === 'number' && + cancelled.tokensBefore > 0, + ).toBe(true); + expect(harness.controller.status()).toEqual({ phase: 'idle' }); + + harness.controller.dispose(); + actor.stop(); + }); + + it('cancels the compaction when the user aborts during quiesce and stays paused until resumed', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + let first = true; + const requester: LlmRequester = { + generate: (_config, { messages }, { onEvent }) => { + const last = messages.at(-1); + const text = last !== undefined && last.role === 'user' ? extractText(last) : ''; + if (!first) { + onEvent?.({ type: 'llm.streaming.part', part: { type: 'text', text: `echo:${text}` } }); + onEvent?.({ type: 'llm.done' }); + return Promise.resolve(); + } + first = false; + return new Promise(() => undefined); + }, + }; + const actor = startAgent(main, requester); + actor.send({ type: 'input.submit', message: createUserMessage('first') }); + await waitFor(actor, (s) => s.matches('running'), { timeout: 5000 }); + const harness = startController(env, actor); + + const compactPromise = harness.controller.compact(); + await vi.waitFor(() => expect(actor.getSnapshot().context.paused).toBe(true), { timeout: 5000 }); + expect(harness.controller.status().phase).toBe('quiescing'); + actor.send({ type: 'input.abort' }); + + await expect(compactPromise).rejects.toMatchObject({ code: 'aborted' }); + await waitFor(actor, (s) => s.matches('idle'), { timeout: 5000 }); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.blocked', + 'compaction.cancelled', + ]); + const cancelled = harness.events.at(-1); + expect(cancelled?.type === 'compaction.cancelled' && cancelled.cause === 'user-abort').toBe(true); + + actor.send({ type: 'input.submit', message: createUserMessage('later') }); + await vi.waitFor(() => expect(main.getState().queue).toHaveLength(1), { timeout: 5000 }); + expect(actor.getSnapshot().matches('idle')).toBe(true); + expect(actor.getSnapshot().context.paused).toBe(true); + expect(historyTexts(main)).toEqual(['first']); + + actor.send({ type: 'input.continue' }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 3, { + timeout: 5000, + }); + expect(historyTexts(main)).toEqual(['first', 'later', 'echo:later']); + expect(harness.controller.status()).toEqual({ phase: 'idle' }); + + harness.controller.dispose(); + actor.stop(); + }); + + it('cancels the compaction when non-input entries land on the branch during summarization', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const actor = startAgent(main, createEchoRequester()); + actor.send({ type: 'input.submit', message: createUserMessage('first') }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 2, { + timeout: 5000, + }); + let resolveSummary: ((outcome: SummaryOutcome) => void) | undefined; + let summaryCalled = false; + const summarize: Summarize = () => { + summaryCalled = true; + return new Promise((resolve) => { + resolveSummary = resolve; + }); + }; + const harness = startController(env, actor, { summarize }); + + const compactPromise = harness.controller.compact(); + await vi.waitFor(() => expect(summaryCalled).toBe(true), { timeout: 5000 }); + await main.dispatch([ + messageAppended({ message: createUserEntry(createUserMessage('foreign'), { source: 'input' }) }), + ]); + (resolveSummary as (outcome: SummaryOutcome) => void)({ text: 'TOO LATE', attempts: 1, droppedCount: 0 }); + + await expect(compactPromise).rejects.toMatchObject({ code: 'drift' }); + expect(main.ref.branch).toBe('main'); + expect(env.tree.has('main~2')).toBe(false); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.cancelled', + ]); + const cancelled = harness.events.at(-1); + expect(cancelled?.type === 'compaction.cancelled' && cancelled.cause === 'drift').toBe(true); + + harness.controller.dispose(); + actor.stop(); + }); +}); + +describe('compaction controller auto', () => { + it('blocks an over-budget step before the request is sent, compacts, then resumes', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const seen: string[] = []; + const requester: LlmRequester = { + generate: (_config, { messages }, { onEvent }) => { + const last = messages.at(-1); + const text = last !== undefined && last.role === 'user' ? extractText(last) : ''; + seen.push(text); + const reply = text === 'big' ? 'R'.repeat(2600) : `echo:${text}`; + onEvent?.({ type: 'llm.streaming.part', part: { type: 'text', text: reply } }); + onEvent?.({ type: 'llm.done' }); + return Promise.resolve(); + }, + }; + const beforeStep: BeforeStepHook = {}; + const actor = startAgent(main, requester, beforeStep, { systemPrompt: 'S'.repeat(4400) }); + const harness = startController(env, actor); + beforeStep.current = harness.controller.onBeforeStep; + + actor.send({ type: 'input.submit', message: createUserMessage('big') }); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 2, { + timeout: 5000, + }); + expect(seen).toEqual(['big']); + + actor.send({ type: 'input.submit', message: createUserMessage('next') }); + await vi.waitFor( + () => { + expect(harness.events.filter((event) => event.type === 'compaction.completed')).toHaveLength(1); + }, + { timeout: 5000 }, + ); + await waitFor(actor, (s) => s.matches('idle') && main.getState().history.length === 5, { + timeout: 5000, + }); + + expect(seen).toHaveLength(2); + expect(seen[1]).toContain('Context compaction is complete'); + expect(main.ref.branch).toBe('main~2'); + const texts = historyTexts(main); + expect(texts[0]).toBe('big'); + expect(texts[1]).toBe('next'); + expect(texts[2]).toContain('SUMMARY TEXT'); + expect(texts[3]).toContain('Context compaction is complete'); + expect(texts[4]).toContain('echo:'); + expect(main.getState().turnIndex.nextTurnId).toBe(4); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.blocked', + 'compaction.completed', + ]); + const completed = harness.events.at(-1); + expect(completed?.type === 'compaction.completed' && completed.originTurnId === 1).toBe(true); + await env.stores.flush(); + expect(main.getState().history).toHaveLength(5); + expect(harness.events.filter((event) => event.type === 'compaction.started')).toHaveLength(1); + + harness.controller.dispose(); + actor.stop(); + }); + + it('recovers from overflow turns up to the attempt cap, then surfaces the failure', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const beforeStep: BeforeStepHook = {}; + const actor = startAgent(main, createOverflowRequester(), beforeStep); + let failedCount = 0; + actor.on('turn.failed', () => { + failedCount += 1; + }); + const harness = startController(env, actor, { maxAutoAttempts: 2 }); + beforeStep.current = harness.controller.onBeforeStep; + + actor.send({ type: 'input.submit', message: createUserMessage('go') }); + await vi.waitFor(() => expect(failedCount).toBe(3), { timeout: 5000 }); + await vi.waitFor( + () => { + expect(harness.events.filter((event) => event.type === 'compaction.completed')).toHaveLength(2); + }, + { timeout: 5000 }, + ); + await waitFor(actor, (s) => s.matches('idle'), { timeout: 5000 }); + + expect(harness.summarizeCalls).toHaveLength(2); + expect(main.ref.branch).toBe('main~3'); + expect(harness.events.map((event) => event.type)).toEqual([ + 'compaction.started', + 'compaction.completed', + 'compaction.started', + 'compaction.completed', + ]); + + harness.controller.dispose(); + actor.stop(); + }); +}); diff --git a/packages/agent-core-v2/src/human/test/session/stores.test.ts b/packages/agent-core-v2/src/human/test/session/stores.test.ts index 35c7674caf..d25dbb0a48 100644 --- a/packages/agent-core-v2/src/human/test/session/stores.test.ts +++ b/packages/agent-core-v2/src/human/test/session/stores.test.ts @@ -7,7 +7,8 @@ import type { LlmModel } from '#/llm/model'; import { createLlmMachine } from '#/llm/requester/machine'; import type { LlmRequester } from '#/llm/requester/requester'; import { createAgentMachine } from '#/agent/machine'; -import { createTurnMachine } from '#/agent/turn'; +import { inputSubmitted, messageAppended, turnEnded, turnStarted } from '#/agent/events'; +import { createTurnMachine, createUserEntry } from '#/agent/turn'; import type { AgentEventStore } from '#/agent/slices'; import { SessionStores } from '#/session/stores'; import { MemoryBackend } from '#/store/backend/memory'; @@ -221,3 +222,59 @@ describe('SessionStores reopen', () => { actor2.stop(); }); }); + +describe('SessionStores switchBranch', () => { + it('seeds a fresh branch, resets the store, and blocks undo across the switch', async () => { + const env = await testEnv(); + const main = await env.stores.open('main'); + const actor = startAgent(main); + await runTurn(actor, main, 'first', 2); + actor.stop(); + const switched: { branch: string; reason?: string; stats?: Record }[] = []; + (await env.stores.session()).subscribe((_state, cause) => { + if (cause.kind === 'event' && cause.event.type === 'agent.switched') { + const event = cause.event as { branch: string; reason?: string; stats?: Record }; + switched.push({ branch: event.branch, reason: event.reason, stats: event.stats }); + } + }); + + const result = await env.stores.switchBranch('main', { + reason: 'compaction', + stats: { compactedCount: 2, tokensBefore: 10, tokensAfter: 5 }, + seed: [ + turnStarted({ turnId: 1 }), + messageAppended({ message: createUserEntry(createUserMessage('seed-user')) }), + messageAppended({ message: createUserEntry(createUserMessage('seed-summary')) }), + turnEnded({ turnId: 1, outcome: 'done' }), + inputSubmitted({ message: createUserMessage('queued') }), + ], + }); + + expect(result.branchId).toBe('main~2'); + expect(main.ref.branch).toBe('main~2'); + expect(historyTexts(main)).toEqual(['seed-user', 'seed-summary']); + expect(main.getState().turnIndex).toEqual({ + turns: [{ turnId: 1, start: { branch: 'main~2', seq: 0 }, end: { branch: 'main~2', seq: 3 } }], + nextTurnId: 2, + }); + expect(main.getState().queue).toEqual([{ id: undefined, message: createUserMessage('queued') }]); + expect(env.tree.openBranch('main~2').header.parentBranch).toBeUndefined(); + expect(switched).toEqual([ + { + branch: 'main~2', + reason: 'compaction', + stats: { compactedCount: 2, tokensBefore: 10, tokensAfter: 5 }, + }, + ]); + await expect(env.stores.undo('main', 1)).rejects.toMatchObject({ reason: 'insufficient' }); + + const actor2 = startAgent(main); + await waitFor(actor2, (s) => s.matches('idle') && main.getState().history.length === 4, { + timeout: 5000, + }); + expect(historyTexts(main)).toEqual(['seed-user', 'seed-summary', 'queued', 'echo:queued']); + expect(main.getState().turnIndex.nextTurnId).toBe(3); + + actor2.stop(); + }); +});