fix pre-step cancellation and compaction convergence

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 },

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<readonly ContentBlock[]>()
+  /** 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('<compacted-summary>')
+    && last?.type === 'text'
+    && last.text === '</compacted-summary>'
+}
+
 /** 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' } })
   }

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() }

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' }])
   })

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