diff --git a/packages/compact/compact-basic/src/index.ts b/packages/compact/compact-basic/src/index.ts index 9d0849b19a..4c7bce00bc 100644 --- a/packages/compact/compact-basic/src/index.ts +++ b/packages/compact/compact-basic/src/index.ts @@ -455,10 +455,11 @@ export class BasicCompactService extends CompactService { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion shadowedTokenCount += this.estimateEventTokens(session.events[seq]!) } - const summaryTokenCount = this.estimateContentTokens(summary) - if (summaryTokenCount >= shadowedTokenCount) { + const framedSummary = this._frameSummary(summary) + const framedSummaryTokenCount = this.estimateContentTokens(framedSummary) + if (framedSummaryTokenCount >= shadowedTokenCount) { throw new Error( - `summary is not smaller than the shadowed content (${summaryTokenCount} estimated tokens >= ${shadowedTokenCount})`, + `summary is not smaller than the shadowed content (${framedSummaryTokenCount} estimated framed tokens >= ${shadowedTokenCount})`, ) } // --- Provenance record (log-only) --- @@ -477,7 +478,7 @@ export class BasicCompactService extends CompactService { // The landed content is FRAMED (checkpoint preamble + tag-wrapped summary); // the compact/summary provenance event above holds the raw model output. session.append('user/message', { - content: this._frameSummary(summary), + content: framedSummary, source: { kind: 'plugin', plugin: 'compact' }, }, { surfaceOp: { op: 'replace', start, end }, diff --git a/packages/compact/compact-basic/tests/compact-basic.spec.ts b/packages/compact/compact-basic/tests/compact-basic.spec.ts index 906841c39f..1134862793 100644 --- a/packages/compact/compact-basic/tests/compact-basic.spec.ts +++ b/packages/compact/compact-basic/tests/compact-basic.spec.ts @@ -12,12 +12,17 @@ import type { Agent } from '@deepseek-ai/dsh-agent' /** A never-aborted signal for the required `compactIfNeeded`/listener arg. */ const SIGNAL = new AbortController().signal +/** Long enough that the real checkpoint preamble is smaller than two fixture messages. */ +const LONG_FIXTURE_TEXT = ' Detailed fixture context that makes framed checkpoint compaction genuinely shrinking.'.repeat(20) + /** * A BasicCompactService with summarize() stubbed (no real model call) and a * predictable token estimate, for deterministic unit tests of the algorithm. */ class TestCompactService extends BasicCompactService { private readonly summaryOutputs = new WeakSet() + /** Boundary/unit tests use tiny fixtures; keep framing from dominating them unless a test opts out. */ + estimateFramedSummariesCheaply = true /** Track calls to summarize for test assertions. */ summarizeCalls: { text: string; model: string }[] = [] /** The fixed summary to return. */ @@ -29,6 +34,7 @@ class TestCompactService extends BasicCompactService { override estimateContentTokens(blocks: readonly ContentBlock[]): number { if (this.summaryOutputs.has(blocks)) return blocks.length * 2 + if (this.estimateFramedSummariesCheaply && isFramedCheckpoint(blocks)) return blocks.length * 2 // 10 tokens per block — predictable for retention/threshold math. return blocks.length * 10 } @@ -43,6 +49,15 @@ class TestCompactService extends BasicCompactService { } } +function isFramedCheckpoint(blocks: readonly ContentBlock[]): boolean { + const first = blocks[0] + const last = blocks[blocks.length - 1] + return first?.type === 'text' + && first.text.includes('') + && last?.type === 'text' + && last.text === '' +} + /** Create a test service with a throwaway context (auto disabled — no model). */ function createTestService(config: BasicCompactConfig = {}): TestCompactService { return new TestCompactService(new Context(), { auto: false, ...config }) @@ -65,12 +80,12 @@ function multiTurnSession(turns: number, messagesPerTurn: number = 2, opts: { le s.append('step/start', { turn: t, step: 1 }) for (let m = 0; m < messagesPerTurn; m++) { s.append('user/message', { - content: [{ type: 'text', text: `turn ${t} user message ${m + 1}` }], + content: [{ type: 'text', text: `turn ${t} user message ${m + 1}.${LONG_FIXTURE_TEXT}` }], source: { kind: 'user' }, }, { surfaceOp: 'append' }) s.append('assistant/message', { turn: t, step: 1, - content: [{ type: 'text', text: `turn ${t} assistant response ${m + 1}` }], + content: [{ type: 'text', text: `turn ${t} assistant response ${m + 1}.${LONG_FIXTURE_TEXT}` }], }, { surfaceOp: 'append' }) } s.append('step/end', { turn: t, step: 1 }) @@ -649,6 +664,7 @@ describe('BasicCompactService.compactIfNeeded', () => { retainTokens: 10, compactionRetries: 2, }) + svc.estimateFramedSummariesCheaply = false svc.mockSummaryQueue = [ Array.from({ length: 4 }, (_, index) => ({ type: 'text', text: `first ${index}` })), [{ type: 'text', text: 'second' }], @@ -670,6 +686,7 @@ describe('BasicCompactService.compactIfNeeded', () => { retainTokens: 10, compactionRetries: 1, }) + svc.estimateFramedSummariesCheaply = false svc.mockSummaryQueue = [ Array.from({ length: 4 }, (_, index) => ({ type: 'text', text: `first ${index}` })), Array.from({ length: 3 }, (_, index) => ({ type: 'text', text: `second ${index}` })), @@ -1067,6 +1084,26 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => { .rejects.toThrow(/summary is not smaller than the shadowed content/) expect(session.events.some(e => e.type === 'compact/summary')).toBe(false) }) + + it('rejects when the framed checkpoint is not smaller than the shadowed content', async () => { + const svc = createTestService({ auto: false }) + svc.estimateFramedSummariesCheaply = false + const session = new Session(SessionId('framed-nonshrinking')) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('user/message', { content: [{ type: 'text', text: 'tiny user' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'tiny assistant' }] }, { surfaceOp: 'append' }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) + const before = [...session.surface.nodes] + const nodes = session.surface.nodes + + await expect(compactRegion(svc, session, nodes[0]!.seq, nodes[1]!.seq, 'm')) + .rejects.toThrow(/summary is not smaller than the shadowed content/) + expect(session.events.some(e => e.type === 'compact/summary')).toBe(false) + expect(session.surface.nodes).toEqual(before) + }) }) describe('BasicCompactService auto-compaction (agent/pre-step listener)', () => { @@ -1606,8 +1643,8 @@ describe('BasicCompactService under the real invariants plugin', () => { function closedTurn(session: Session, turn: number): void { session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('step/start', { turn, step: 1 }) - session.append('user/message', { content: [{ type: 'text', text: `turn ${turn} user` }], source: { kind: 'user' } }, { surfaceOp: 'append' }) - session.append('assistant/message', { turn, step: 1, content: [{ type: 'text', text: `turn ${turn} assistant` }] }, { surfaceOp: 'append' }) + session.append('user/message', { content: [{ type: 'text', text: `turn ${turn} user.${LONG_FIXTURE_TEXT}` }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + session.append('assistant/message', { turn, step: 1, content: [{ type: 'text', text: `turn ${turn} assistant.${LONG_FIXTURE_TEXT}` }] }, { surfaceOp: 'append' }) session.append('step/end', { turn, step: 1 }) session.append('turn/end', { turn, reason: { kind: 'completed' } }) } diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index 2982a60fc8..200115af9a 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -432,16 +432,24 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, // pre-step plugin ends the turn, not the loop. await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, abort.signal) + // Interruption landing during the pre-step seam: do not open an empty + // step. `agent/step-start` listeners get their own check below because + // they necessarily run after step/start is appended/emitted. + if (handle.isCancelled() || handle.isDisposed()) { + handle.setAbort(undefined) + reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() } + break + } + session.append('step/start', { turn, step }) stepOpen = true ctx.emit('agent/step-start', agent, turn, step) - // Cancel landing in the seam / step-start window: a `cancel()` during the - // pre-step seam (it aborted `abort.signal` above) OR a synchronous - // `agent/step-start` listener that cancels. And disposal, which the earlier - // assembly check may have missed if it only checked isCancelled. Check - // AFTER step/start append + emit and before `runStep`: drop the step, end - // the turn accordingly. closeStep balances the already-appended step/start. + // Cancel landing in the step-start window: a synchronous + // `agent/step-start` listener can cancel after the step is already open. + // Check AFTER step/start append + emit and before `runStep`: drop the + // step, end the turn accordingly. closeStep balances the already-appended + // step/start. if (handle.isCancelled() || handle.isDisposed()) { handle.setAbort(undefined) reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() } diff --git a/packages/core/agent-loop/tests/review-fixes.spec.ts b/packages/core/agent-loop/tests/review-fixes.spec.ts index 5b02ca60c1..346d70e2f7 100644 --- a/packages/core/agent-loop/tests/review-fixes.spec.ts +++ b/packages/core/agent-loop/tests/review-fixes.spec.ts @@ -1216,6 +1216,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => { const turnEnd = e.findLast(x => x.type === 'turn/end') // Disposal wins the post-seam check — reason is `disposed`. expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' }) + expect(e.some(x => x.type === 'step/start')).toBe(false) expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) // agent/turn-end may not fire when disposal happens during pre-step: the // fiber's disposer runs before closeTurn(true)'s emit. The durable turn/end @@ -1265,6 +1266,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => { expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) const turnEnd = e.findLast(x => x.type === 'turn/end') expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: 'user cancelled' }) + expect(e.some(x => x.type === 'step/start')).toBe(false) expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled' }]) }) diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 0edc787f9c..fc412f0e0e 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -179,7 +179,7 @@ declare module 'cordis' { */ 'agent/step-end'(agent: Agent, turn: number, step: number): void - // ---- interception seams (waterfall) ---- + // ---- step/request extension seams (serial + waterfall) ---- /** * Awaited pre-step surface-mutation checkpoint, fired once per step AFTER * `turn/start` (and after the prior step closed) but BEFORE this step's