From 6ad97867053e744c3050e5d7e8caa7636357bcbd Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Fri, 11 Sep 2026 11:05:51 +0800 Subject: [PATCH 1/2] fix(agent-core-v2): deliver cron-fired prompts as single opening turn message isDisplayablePromptOrigin now admits cron_job/cron_missed, so the observable turn.started carries the cron-fire prompt text instead of leaving the opening wire user empty. prompt.inject only dispatches turn.steer when the message actually steers into a running turn; a fresh-turn submission is fully expressed by turn.prompt/turn.started, so the duplicate materialization and both opening-steer dedup hacks (live projector and coldFold) go away. --- .../src/agent/loop/turnEvents.ts | 1 + .../src/agent/prompt/promptService.ts | 11 +++++-- .../src/services/history/coldFold.ts | 17 +--------- .../src/services/projection/agentProjector.ts | 31 ------------------- .../kap-server/test/services/history.test.ts | 13 +------- .../test/services/projection.test.ts | 12 ------- 6 files changed, 11 insertions(+), 74 deletions(-) diff --git a/packages/agent-core-v2/src/agent/loop/turnEvents.ts b/packages/agent-core-v2/src/agent/loop/turnEvents.ts index f0163142e58..deb41b8b643 100644 --- a/packages/agent-core-v2/src/agent/loop/turnEvents.ts +++ b/packages/agent-core-v2/src/agent/loop/turnEvents.ts @@ -90,6 +90,7 @@ export function turnPromptAttachments( export function isDisplayablePromptOrigin(origin: PromptOrigin): boolean { if (origin.kind === 'user') return true; + if (origin.kind === 'cron_job' || origin.kind === 'cron_missed') return true; if (origin.kind === 'system_trigger' && origin.name === 'subagent') return true; return ( (origin.kind === 'skill_activation' || origin.kind === 'plugin_command') && diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 3a3a16a421b..100b28aa19e 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -492,7 +492,7 @@ export class AgentPromptService implements IAgentPromptService { id: ownerPromptId, content: gateImageFormatParts(rerouted.content, this.profile.getModelProviderType()), }; - const request = { + const steered = this.loop.steer({ message: gated, promptId: ownerPromptId, onMaterialize: () => { @@ -505,8 +505,13 @@ export class AgentPromptService implements IAgentPromptService { ); this.notifyCaptions(captions, ownerPromptId); }, - }; - return this.loop.steer(request) ?? this.loop.submit(request).turn; + }); + if (steered !== undefined) return steered; + return this.loop.submit({ + message: gated, + promptId: ownerPromptId, + onMaterialize: () => this.notifyCaptions(captions, ownerPromptId), + }).turn; } async retry(): Promise { diff --git a/packages/kap-server/src/services/history/coldFold.ts b/packages/kap-server/src/services/history/coldFold.ts index be6a9361f90..229eca51d20 100644 --- a/packages/kap-server/src/services/history/coldFold.ts +++ b/packages/kap-server/src/services/history/coldFold.ts @@ -182,8 +182,6 @@ interface TurnScratch { currentStep?: number; serverUserSeq: number; attachmentSeq: number; - openingInputKey?: string; - openingSteerDeduped: boolean; } interface GoalState { @@ -271,7 +269,7 @@ export function foldWireHistory( const scratch = (rawId: number): TurnScratch => { let entry = scratchByTurn.get(rawId); if (entry === undefined) { - entry = { serverUserSeq: 0, attachmentSeq: 0, openingSteerDeduped: false }; + entry = { serverUserSeq: 0, attachmentSeq: 0 }; scratchByTurn.set(rawId, entry); } return entry; @@ -500,8 +498,6 @@ export function foldWireHistory( scratchByTurn.set(rawId, { serverUserSeq: carriedUserSeq, attachmentSeq: attachments, - openingInputKey: JSON.stringify(input), - openingSteerDeduped: false, }); if (openingMessageId !== undefined) { const notification = @@ -537,16 +533,6 @@ export function foldWireHistory( if (rawId === undefined || hiddenTurnIds.has(rawId)) return; const input = Array.isArray(record['input']) ? (record['input'] as ContentPart[]) : []; const skipBlocks = kind === 'user' ? (origin?.skillActivations?.length ?? 0) : 0; - const entry = scratch(rawId); - if ( - entry.currentStep === undefined && - !entry.openingSteerDeduped && - entry.openingInputKey !== undefined && - entry.openingInputKey === JSON.stringify(input) - ) { - entry.openingSteerDeduped = true; - return; - } if (kind === 'user') { const recordAtMs = atMs(record); const matchedId = matchQueuedPrompt(input, skipBlocks); @@ -881,7 +867,6 @@ export function foldWireHistory( scratchByTurn.set(rawId, { serverUserSeq: 0, attachmentSeq: 0, - openingSteerDeduped: false, }); } const entry = scratch(rawId); diff --git a/packages/kap-server/src/services/projection/agentProjector.ts b/packages/kap-server/src/services/projection/agentProjector.ts index 9aeaba4ffa4..630c244e20a 100644 --- a/packages/kap-server/src/services/projection/agentProjector.ts +++ b/packages/kap-server/src/services/projection/agentProjector.ts @@ -72,8 +72,6 @@ interface TurnRecord { promptId?: string; userMessageId?: string; attachmentIds?: string[]; - openingKey?: { text: string; attachments: number }; - openingSteerDeduped: boolean; startedAt?: string; endedAt?: string; durationMs?: number; @@ -352,7 +350,6 @@ export class AgentMessageProjector { info.promptId === undefined ? undefined : (info.userMessageId ?? turnUserMessageIdOf(turnId)), - openingSteerDeduped: false, }; this.turns.set(turnId, this.currentTurn); this.timelineIds.push(turnId); @@ -661,8 +658,6 @@ export class AgentMessageProjector { ? (promptRecord?.userMessageId ?? turnUserMessageIdOf(turnId)) : undefined, attachmentIds: attachmentIds.length > 0 ? attachmentIds : undefined, - openingKey: { text: promptText ?? '', attachments: attachmentIds.length }, - openingSteerDeduped: false, startedAt: epochMsToIso(event.time), }; this.currentTurn = turn; @@ -938,7 +933,6 @@ export class AgentMessageProjector { status: 'running', origin: { kind: 'other' }, anchor: false, - openingSteerDeduped: false, startedAt: epochMsToIso(time), }; this.currentTurn = turn; @@ -1466,15 +1460,6 @@ export class AgentMessageProjector { const turn = this.currentTurn; if (turn === undefined || turn.status !== 'running') return ops; const skipBlocks = kind === 'user' ? (origin.skillActivations?.length ?? 0) : 0; - const step = this.currentStep; - const stepStarted = step !== undefined && step.turnId === turn.turnId; - if (!stepStarted && !turn.openingSteerDeduped && turn.openingKey !== undefined) { - const key = steerKeyOf(event.input, skipBlocks); - if (key.text === turn.openingKey.text && key.attachments === turn.openingKey.attachments) { - turn.openingSteerDeduped = true; - return ops; - } - } const matched = kind === 'user' ? this.matchQueuedPrompt(event.input, skipBlocks) : undefined; if (matched !== undefined) { @@ -2306,22 +2291,6 @@ export function promptTextOf(content: readonly ContentPart[]): string { .join(''); } -export function steerKeyOf( - input: readonly ContentPart[], - skipBlocks: number, -): { text: string; attachments: number } { - let text = ''; - let attachments = 0; - for (const part of input.slice(skipBlocks)) { - if (part.type === 'text') { - text += part.text; - continue; - } - if (daemonFileRefFromPart(part) !== undefined) attachments += 1; - } - return { text, attachments }; -} - export function wireContentParts(content: readonly ContentPart[]): WireContentPart[] { const out: WireContentPart[] = []; for (const part of content) { diff --git a/packages/kap-server/test/services/history.test.ts b/packages/kap-server/test/services/history.test.ts index 0a2ea582947..db064c03913 100644 --- a/packages/kap-server/test/services/history.test.ts +++ b/packages/kap-server/test/services/history.test.ts @@ -275,7 +275,7 @@ describe('foldWireHistory origin classification', () => { }); describe('foldWireHistory steer', () => { - it('attaches steers to the running step, buffers between steps, and dedupes the turn-opening steer', () => { + it('attaches steers to the running step and buffers between steps', () => { const messages = fold([ rec('turn.prompt', { input: [{ type: 'text', text: 'do A' }], origin: { kind: 'user' } }), loopEvent({ type: 'step.begin', uuid: 'u1', turnId: '0', step: 1 }, T0 + 1), @@ -300,17 +300,6 @@ describe('foldWireHistory steer', () => { status: 'read', timestamp: T0 + 4, }); - - const deduped = fold([ - rec('turn.prompt', { input: [{ type: 'text', text: 'hello' }], origin: { kind: 'user' } }), - rec('turn.steer', { input: [{ type: 'text', text: 'hello' }], origin: { kind: 'user' } }, T0 + 1), - ]); - const dedupedUsers = ofType(deduped, 'user'); - expect(dedupedUsers).toHaveLength(1); - expect(dedupedUsers[0]).toMatchObject({ - message_id: 't0.u0', - text: [{ type: 'text', text: 'hello', meta: {} }], - }); }); it('emits steers read at their record without synthesizing a step', () => { diff --git a/packages/kap-server/test/services/projection.test.ts b/packages/kap-server/test/services/projection.test.ts index 8f6031547d3..ddcb0213b41 100644 --- a/packages/kap-server/test/services/projection.test.ts +++ b/packages/kap-server/test/services/projection.test.ts @@ -960,18 +960,6 @@ describe('AgentMessageProjector', () => { expect(ofType(messages, 'user')).toHaveLength(0); }); - it('dedupes the turn-opening steer that repeats the prompt input', () => { - const projector = makeProjector(); - const messages = feedAll(projector, [ - ev({ type: 'turn.started', turnId: 1, origin: { kind: 'user' }, prompt: 'hello' }), - ev({ type: 'turn.steer', turnId: 1, input: [{ type: 'text', text: 'hello' }], origin: { kind: 'user' } }), - ev({ type: 'turn.step.started', turnId: 1, step: 1 }), - ]); - const users = ofType(messages, 'user'); - expect(users).toHaveLength(1); - expect(users[0]).toMatchObject({ message_id: 't1.u0', text: [{ type: 'text', text: 'hello', meta: {} }] }); - }); - it('counts undo anchors instead of timeline turns when fromTurnId is missing', () => { const projector = makeProjector(); const messages = feedAll(projector, [ From 4a568713fdf0397ea5eac9e0301dcefd4a31b7bd Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Fri, 11 Sep 2026 12:06:40 +0800 Subject: [PATCH 2/2] test(agent-core-v2): drop stale compaction controller test for removed modules The file imports #/llm/requester/machine, #/llm/message and other modules that no longer exist after the requester-pipeline refactor; it resurfaced in the dev rewrite and fails the suite at import time. --- .../human/test/compaction/controller.test.ts | 541 ------------------ 1 file changed, 541 deletions(-) delete mode 100644 packages/agent-core-v2/src/human/test/compaction/controller.test.ts 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 deleted file mode 100644 index 5476f8d2a5c..00000000000 --- a/packages/agent-core-v2/src/human/test/compaction/controller.test.ts +++ /dev/null @@ -1,541 +0,0 @@ -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(); - }); -});