feat(compact): recover context overflow (PR3 phase 2)

packages/compact/compact-basic/tests/compact-basic.spec.ts

@@ -6,9 +6,10 @@ import BasicCompactService, {
 } from '@deepseek-ai/dsh-compact-basic'
 import type { BasicCompactConfig } from '@deepseek-ai/dsh-compact-basic'
 import { selectCompactableRange } from '@deepseek-ai/dsh-compact-basic/src/region.ts'
+import { toolPairingBalancedAfter, toolPairingBalancedBefore } from '@deepseek-ai/dsh-compact'
 import type { CompactionResult } from '@deepseek-ai/dsh-compact'
-import LlmService, { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
-import type { ContentBlock, GenerateOptions, Message, StreamChunk } from '@deepseek-ai/dsh-llm'
+import LlmService, { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, LlmAdapter } from '@deepseek-ai/dsh-llm'
+import type { ContentBlock, GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
 import { Session, SessionId } from '@deepseek-ai/dsh-session'
 import TokenMeterService, {
   TOKEN_METER_MODEL_UNCONFIGURED,
@@ -43,6 +44,12 @@ function conversation(turns = 4, text = 'fixture'): Session {
       source: { kind: 'user' },
     }, { surfaceOp: 'append' })
     session.append('step/start', { turn, step: 1 })
+    if (turn === 1) {
+      session.append('request/header', {
+        header: { config: { model: MODEL } },
+        reason: 'initial',
+      })
+    }
     session.append('assistant/message', {
       turn,
       step: 1,
@@ -68,6 +75,12 @@ function toolConversation(): Session {
       source: { kind: 'user' },
     }, { surfaceOp: 'append' })
     session.append('step/start', { turn, step: 1 })
+    if (turn === 1) {
+      session.append('request/header', {
+        header: { config: { model: MODEL } },
+        reason: 'initial',
+      })
+    }
     session.append('assistant/message', {
       turn,
       step: 1,
@@ -120,11 +133,10 @@ function service(
 async function compactIfNeeded(
   compact: BasicCompactService,
   session: Session,
+  trigger: 'pressure' | 'context-overflow' = 'pressure',
   model: string | undefined = MODEL,
-  system = '',
-  prefix: readonly Message[] = [],
 ): Promise<CompactionResult | null> {
-  return compact.compactIfNeeded(agent(session, model), system, prefix, SIGNAL)
+  return compact.compactIfNeeded(agent(session, model), trigger, SIGNAL)
 }
 
 describe('compact configuration and defaults', () => {
@@ -140,6 +152,7 @@ describe('compact configuration and defaults', () => {
       summarizationModel: '',
       maxTokens: 8192,
       compactionRetries: 1,
+      maxOverflowRetries: 1,
       auto: true,
     })
     expect(resolveModelConfig(resolved, ctx.tokenMeter.resolve(MODEL))).toEqual({
@@ -176,6 +189,7 @@ describe('compact configuration and defaults', () => {
     const bad = [
       [{ maxTokens: 0 }, /maxTokens/],
       [{ compactionRetries: -1 }, /compactionRetries/],
+      [{ maxOverflowRetries: -1 }, /maxOverflowRetries/],
       [{ auto: 'yes' }, /auto must be a boolean/],
       [{ summarizationModel: 1 }, /summarizationModel must be a string/],
       [{ models: null }, /models must be an object/],
@@ -207,19 +221,57 @@ describe('pressure measurement and retention', () => {
     models: { [MODEL]: { thresholdRatio: 0.5, retainTokens: 18 } },
   }
 
-  it('skips the provisional check only when no routed or fallback model exists', async () => {
+  it('skips when no durable routed model exists instead of using AgentOptions fallback', async () => {
     const compact = service(compactConfig)
-    const session = conversation()
-    expect(await compact.compactIfNeeded(agent(session), '', [], SIGNAL)).toBeNull()
+    const session = new Session(SessionId('headerless'))
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    await expect(compact.compactIfNeeded(agent(session, MODEL), 'pressure', SIGNAL))
+      .resolves.toBeNull()
     expect(compact.calls).toHaveLength(0)
   })
 
   it('throws for a named unconfigured model instead of swallowing it', async () => {
     const compact = service(compactConfig)
-    await expect(compactIfNeeded(compact, conversation(), 'missing'))
+    const session = conversation()
+    session.append('request/header', {
+      header: { config: { model: 'missing' } },
+      reason: 'resume',
+    })
+    await expect(compactIfNeeded(compact, session))
       .rejects.toMatchObject({ code: TOKEN_METER_MODEL_UNCONFIGURED, model: 'missing' })
   })
 
+  it('declines forced overflow when the whole surface is one indivisible tool pair', async () => {
+    const compact = service(compactConfig)
+    const session = new Session(SessionId('single-tool-pair'))
+    const callId = CallId('single-call')
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    session.append('step/start', { turn: 1, step: 1 })
+    session.append('request/header', {
+      header: { config: { model: MODEL } },
+      reason: 'initial',
+    })
+    session.append('assistant/message', {
+      turn: 1,
+      step: 1,
+      content: [{ type: 'tool-call', id: callId, name: 'read', arguments: '{}' }],
+    }, { surfaceOp: 'append' })
+    session.append('tool/call', { turn: 1, step: 1, callId, name: 'read', arguments: '{}' })
+    session.append('tool/result', {
+      turn: 1,
+      step: 1,
+      callId,
+      content: [{ type: 'text', text: 'result' }],
+      isError: false,
+    }, { surfaceOp: 'append' })
+    session.append('step/end', { turn: 1, step: 1 })
+    const generation = session.surface.replaceGeneration
+
+    await expect(compactIfNeeded(compact, session, 'context-overflow')).resolves.toBeNull()
+    expect(session.surface.replaceGeneration).toBe(generation)
+    expect(session.events.some(event => event.type === 'compact/start')).toBe(false)
+  })
+
   it('does nothing below threshold and compacts a priced head above threshold', async () => {
     const compact = service(compactConfig)
     expect(await compactIfNeeded(compact, conversation(2))).toBeNull()
@@ -231,7 +283,7 @@ describe('pressure measurement and retention', () => {
     expect(session.surface.nodes.length).toBeLessThan(8)
   })
 
-  it('counts the current prompt and request prefix without putting either on the surface', async () => {
+  it('counts the durable routed request envelope without putting its prefix on the surface', async () => {
     const compact = service({
       auto: false,
       models: { [MODEL]: { thresholdRatio: 0.7, retainTokens: 9 } },
@@ -239,11 +291,16 @@ describe('pressure measurement and retention', () => {
     const session = conversation(2, 'x'.repeat(2_000))
     expect(await compactIfNeeded(compact, session)).toBeNull()
 
-    const prefix: Message[] = [{
-      role: 'user',
-      content: [{ type: 'text', text: 'p'.repeat(10_000) }],
-    }]
-    const result = await compactIfNeeded(compact, session, MODEL, 's'.repeat(5_000), prefix)
+    const prefix = [{ role: 'user' as const, content: [{ type: 'text' as const, text: 'p'.repeat(10_000) }] }]
+    session.append('request/header', {
+      header: {
+        config: { model: MODEL },
+        system: 's'.repeat(5_000),
+        messagePrefix: prefix,
+      },
+      reason: 'resume',
+    })
+    const result = await compactIfNeeded(compact, session)
     expect(result).not.toBeNull()
     expect(prefix).toHaveLength(1)
     expect(session.events.some(event => event.type === 'context/message')).toBe(false)
@@ -264,17 +321,26 @@ describe('pressure measurement and retention', () => {
       reason: 'initial',
     })
 
-    const result = await compactIfNeeded(compact, session, 'fallback')
+    const result = await compactIfNeeded(compact, session, 'pressure', 'fallback')
     expect(result).not.toBeNull()
   })
 
   it('declines when envelope pressure is high but the surface has no compactable range', async () => {
     const compact = service(compactConfig)
     const empty = new Session(SessionId('empty'))
-    expect(await compactIfNeeded(compact, empty, MODEL, 'x'.repeat(100_000))).toBeNull()
+    empty.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    empty.append('request/header', {
+      header: { config: { model: MODEL }, system: 'x'.repeat(100_000) },
+      reason: 'initial',
+    })
+    expect(await compactIfNeeded(compact, empty)).toBeNull()
 
     const retained = conversation(1)
-    expect(await compactIfNeeded(compact, retained, MODEL, 'x'.repeat(100_000))).toBeNull()
+    retained.append('request/header', {
+      header: { config: { model: MODEL }, system: 'x'.repeat(100_000) },
+      reason: 'resume',
+    })
+    expect(await compactIfNeeded(compact, retained)).toBeNull()
   })
 
   it('detects scalar/surface revision disagreement', async () => {
@@ -587,7 +653,19 @@ describe('compaction region transaction', () => {
 
   it('requires a conversation model for pricing', async () => {
     const compact = service()
-    const session = conversation(1)
+    const session = new Session(SessionId('model-less-region'))
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    session.append('user/message', {
+      content: [{ type: 'text', text: 'history' }],
+      source: { kind: 'user' },
+    }, { surfaceOp: 'append' })
+    session.append('step/start', { turn: 1, step: 1 })
+    session.append('assistant/message', {
+      turn: 1,
+      step: 1,
+      content: [{ type: 'text', text: 'answer' }],
+    }, { surfaceOp: 'append' })
+    session.append('step/end', { turn: 1, step: 1 })
     const nodes = session.surface.nodes
     await expect(compact.compactRegion(
       session,
@@ -679,6 +757,25 @@ describe('default one-shot summarizer', () => {
     expect(adapter.lastOptions?.model).toBe('routed')
   })
 
+  it('records the model actually dispatched after one-shot stream routing', async () => {
+    const { ctx, compact } = await summarizerHarness([{ type: 'text', text: 'unused' }])
+    const routedAdapter = new ScriptedAdapter([{ type: 'text', text: 'routed summary' }])
+    ctx.llm.registerAdapter(['routed-summary-model'], routedAdapter)
+    ctx.on('llm/stream', (options, next) => {
+      options.model = 'routed-summary-model'
+      return next()
+    })
+
+    const session = conversation(3, 'large history '.repeat(500))
+    const nodes = session.surface.nodes
+    await compact.compactRegion(session, nodes[0]!.seq, nodes[3]!.seq, agent(session, MODEL), SIGNAL)
+    expect(session.events.findLast(event => event.type === 'compact/summary')?.data).toMatchObject({
+      summary: [{ type: 'text', text: 'routed summary' }],
+      model: 'routed-summary-model',
+    })
+    expect(routedAdapter.lastOptions?.model).toBe('routed-summary-model')
+  })
+
   it('fails clearly when no summarization model can be resolved', async () => {
     const ctx = new Context()
     await ctx.plugin(LlmService)
@@ -717,21 +814,36 @@ describe('default one-shot summarizer', () => {
 })
 
 describe('automatic listener and loader composition', () => {
-  function preStep(ctx: Context, owner: Agent): Promise<unknown> {
-    return ctx.serial('agent/pre-step', owner, 1, 1, '', [], SIGNAL)
+  function postStep(ctx: Context, owner: Agent, signal = SIGNAL): Promise<unknown> {
+    return ctx.serial('agent/post-step', owner, 1, 1, signal)
   }
 
-  it('compacts above threshold and remains idle below it', async () => {
+  function recover(
+    ctx: Context,
+    owner: Agent,
+    error: Error & { code?: string },
+    retryAttempt = 0,
+    signal = SIGNAL,
+    next: () => Promise<{ action: 'fail' | 'retry' }> = () => Promise.resolve({ action: 'fail' }),
+  ): Promise<{ action: 'fail' | 'retry' }> {
+    return ctx.waterfall('agent/request-error', owner, 1, 1, error, retryAttempt, signal, next)
+  }
+
+  function overflow(message = 'provider overflow'): Error & { code: string } {
+    return Object.assign(new Error(message), { code: CONTEXT_WINDOW_EXCEEDED_CODE })
+  }
+
+  it('compacts post-step above threshold using the durable routed model and remains idle below it', async () => {
     const ctx = createContext()
     const compact = new TestCompactService(ctx, {
       models: { [MODEL]: { thresholdRatio: 0.5, retainTokens: 18 } },
     })
     const pressured = conversation(4)
-    await preStep(ctx, agent(pressured, MODEL))
+    await postStep(ctx, agent(pressured, 'unconfigured-agent-fallback'))
     expect(pressured.events.some(event => event.type === 'compact/summary')).toBe(true)
 
     const small = conversation(1)
-    await preStep(ctx, agent(small, MODEL))
+    await postStep(ctx, agent(small, MODEL))
     expect(small.events.some(event => event.type === 'compact/start')).toBe(false)
     expect(compact.calls).toHaveLength(1)
   })
@@ -746,7 +858,7 @@ describe('automatic listener and loader composition', () => {
     compact.error = 'temporary failure'
     const session = conversation(4)
 
-    await expect(preStep(ctx, agent(session, MODEL))).resolves.toBeUndefined()
+    await expect(postStep(ctx, agent(session, MODEL))).resolves.toBeUndefined()
     expect(warnings).toContainEqual(expect.stringContaining('temporary failure'))
     expect(session.events.some(event => event.type === 'compact/summary')).toBe(false)
   })
@@ -754,21 +866,210 @@ describe('automatic listener and loader composition', () => {
   it('propagates a named unknown-model configuration failure', async () => {
     const ctx = createContext()
     void new TestCompactService(ctx)
-    await expect(preStep(ctx, agent(conversation(4), 'missing'))).rejects.toMatchObject({
+    const session = conversation(4)
+    session.append('request/header', {
+      header: { config: { model: 'missing' } },
+      reason: 'resume',
+    })
+    await expect(postStep(ctx, agent(session, MODEL))).rejects.toMatchObject({
       code: TOKEN_METER_MODEL_UNCONFIGURED,
       model: 'missing',
     })
   })
 
-  it('auto:false installs no listener', async () => {
+  it('force-compacts below normal pressure for canonical overflow and retries only after replacement', async () => {
+    const ctx = createContext()
+    void new TestCompactService(ctx, {
+      models: { [MODEL]: { thresholdRatio: 1, retainTokens: 90 } },
+    })
+    const session = conversation(3)
+    const beforeGeneration = session.surface.replaceGeneration
+    const retainedSeq = session.surface.nodes.at(-1)!.seq
+    const threshold = 100
+    expect(ctx.tokenMeter.resolve(MODEL).measure(session).totalTokens).toBeLessThan(threshold)
+    const decision = await recover(ctx, agent(session, 'unconfigured-agent-fallback'), overflow())
+
+    expect(decision).toEqual({ action: 'retry' })
+    expect(session.surface.replaceGeneration).toBe(beforeGeneration + 1)
+    expect(session.events.some(event => event.type === 'compact/summary')).toBe(true)
+    expect(session.surface.nodes.some(node => node.seq === retainedSeq)).toBe(true)
+  })
+
+  it('preserves the newest whole tool-call/result pair during forced overflow compaction', async () => {
+    const ctx = createContext()
+    void new TestCompactService(ctx, {
+      models: { [MODEL]: { thresholdRatio: 1, retainTokens: 90 } },
+    })
+    const session = toolConversation()
+    const newestAssistant = session.surface.nodes.at(-2)!
+    const newestResult = session.surface.nodes.at(-1)!
+
+    expect(await recover(ctx, agent(session, MODEL), overflow())).toEqual({ action: 'retry' })
+    const currentAssistant = session.surface.nodes.find(node => node.seq === newestAssistant.seq)
+    const currentResult = session.surface.nodes.find(node => node.seq === newestResult.seq)
+    expect(currentAssistant).toBeDefined()
+    expect(currentResult).toBeDefined()
+    expect(toolPairingBalancedBefore(session, currentAssistant!)).toBe(true)
+    expect(toolPairingBalancedAfter(session, currentResult!)).toBe(true)
+  })
+
+  it('does not retry when a backend reports success without replacing the surface', async () => {
+    const ctx = createContext()
+    const compact = new TestCompactService(ctx)
+    const session = conversation(2)
+    const fakeResult: CompactionResult = {
+      startSeq: 1,
+      summarySeq: 2,
+      endSeq: 3,
+      summary: [{ type: 'text', text: 'fake' }],
+      shadowedRange: { start: 1, end: 2 },
+      shadowedSeqs: [1, 2],
+      shadowedTokenCount: 10,
+    }
+    vi.spyOn(compact, 'compactIfNeeded').mockResolvedValue(fakeResult)
+
+    expect(await recover(ctx, agent(session, MODEL), overflow())).toEqual({ action: 'fail' })
+    expect(session.surface.replaceGeneration).toBe(0)
+  })
+
+  it('delegates downstream exactly once when no replacement is available', async () => {
+    const ctx = createContext()
+    const compact = new TestCompactService(ctx)
+    vi.spyOn(compact, 'compactIfNeeded').mockResolvedValue(null)
+    const downstream = new Error('downstream recovery failed')
+    let calls = 0
+
+    await expect(recover(
+      ctx,
+      agent(conversation(2), MODEL),
+      overflow(),
+      0,
+      SIGNAL,
+      () => {
+        calls += 1
+        return Promise.reject(downstream)
+      },
+    )).rejects.toBe(downstream)
+    expect(calls).toBe(1)
+  })
+
+  it('preserves the original provider error when recovery throws', async () => {
+    const ctx = createContext()
+    const warnings: string[] = []
+    ctx.logger.warn = ((message: string) => void warnings.push(message)) as typeof ctx.logger.warn
+    const compact = new TestCompactService(ctx)
+    compact.error = new Error('summary unavailable')
+    const original = overflow('original provider overflow')
+
+    expect(await recover(ctx, agent(conversation(3), MODEL), original)).toEqual({ action: 'fail' })
+    expect(original).toMatchObject({
+      message: 'original provider overflow',
+      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+    })
+    expect(warnings).toContainEqual(expect.stringContaining('preserving the original request error'))
+  })
+
+  it('delegates once when overflow recovery throws a non-Error value', async () => {
+    const ctx = createContext()
+    const warnings: string[] = []
+    ctx.logger.warn = ((message: string) => void warnings.push(message)) as typeof ctx.logger.warn
+    const compact = new TestCompactService(ctx)
+    compact.error = 'non-error recovery failure'
+    const session = conversation(3)
+    const generation = session.surface.replaceGeneration
+    const original = overflow('original provider failure')
+    let delegations = 0
+
+    const decision = await recover(ctx, agent(session, MODEL), original, 0, SIGNAL, () => {
+      delegations += 1
+      return Promise.resolve({ action: 'fail' })
+    })
+
+    expect(decision).toEqual({ action: 'fail' })
+    expect(delegations).toBe(1)
+    expect(session.surface.replaceGeneration).toBe(generation)
+    expect(original).toMatchObject({
+      message: 'original provider failure',
+      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+    })
+    expect(warnings).toContainEqual(expect.stringContaining('non-error recovery failure'))
+  })
+
+  it('delegates once and preserves the original overflow for an unknown routed meter model', async () => {
+    const ctx = createContext()
+    void new TestCompactService(ctx)
+    const session = conversation(2)
+    session.append('request/header', {
+      header: { config: { model: 'unknown-routed-model' } },
+      reason: 'resume',
+    })
+    const original = overflow('original unknown-model overflow')
+    let delegations = 0
+
+    const decision = await recover(ctx, agent(session, MODEL), original, 0, SIGNAL, () => {
+      delegations += 1
+      return Promise.resolve({ action: 'fail' })
+    })
+    expect(decision).toEqual({ action: 'fail' })
+    expect(delegations).toBe(1)
+    expect(original).toMatchObject({
+      message: 'original unknown-model overflow',
+      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+    })
+  })
+
+  it('honors retry caps, non-context failures, and cancellation', async () => {
+    const ctx = createContext()
+    const compact = new TestCompactService(ctx, { maxOverflowRetries: 1 })
+    const compactSpy = vi.spyOn(compact, 'compactIfNeeded')
+    const owner = agent(conversation(3), MODEL)
+    expect(await recover(ctx, owner, Object.assign(new Error('rate limit'), { code: 'RATE_LIMIT' })))
+      .toEqual({ action: 'fail' })
+    expect(await recover(ctx, owner, overflow(), 1)).toEqual({ action: 'fail' })
+
+    const controller = new AbortController()
+    controller.abort('cancelled')
+    expect(await recover(ctx, owner, overflow(), 0, controller.signal)).toEqual({ action: 'fail' })
+    expect(compactSpy).not.toHaveBeenCalled()
+  })
+
+  it('does not retry when cancellation lands during an awaited compaction', async () => {
+    const ctx = createContext()
+    const compact = new TestCompactService(ctx)
+    const controller = new AbortController()
+    compact.mutateDuringSummary = () => { controller.abort('cancelled during summary') }
+    const session = conversation(3)
+    const generation = session.surface.replaceGeneration
+
+    expect(await recover(ctx, agent(session, MODEL), overflow(), 0, controller.signal))
+      .toEqual({ action: 'fail' })
+    expect(session.surface.replaceGeneration).toBe(generation + 1)
+  })
+
+  it('maxOverflowRetries:0 disables recovery without disabling post-step pressure', async () => {
+    const ctx = createContext()
+    void new TestCompactService(ctx, {
+      maxOverflowRetries: 0,
+      models: { [MODEL]: { thresholdRatio: 0.5, retainTokens: 18 } },
+    })
+    const session = conversation(4)
+    await postStep(ctx, agent(session, MODEL))
+    const summaries = session.events.filter(event => event.type === 'compact/summary').length
+    expect(summaries).toBe(1)
+    expect(await recover(ctx, agent(session, MODEL), overflow())).toEqual({ action: 'fail' })
+    expect(session.events.filter(event => event.type === 'compact/summary')).toHaveLength(summaries)
+  })
+
+  it('auto:false installs neither automatic listener', async () => {
     const ctx = createContext()
     void new TestCompactService(ctx, {
       auto: false,
       models: { [MODEL]: { thresholdRatio: 0.5, retainTokens: 18 } },
     })
     const session = conversation(4)
-    await preStep(ctx, agent(session, MODEL))
+    await postStep(ctx, agent(session, MODEL))
     expect(session.events.some(event => event.type === 'compact/start')).toBe(false)
+    expect(await recover(ctx, agent(session, MODEL), overflow())).toEqual({ action: 'fail' })
   })
 
   it('loads and disposes the real zero-config service stack', async () => {
@@ -797,8 +1098,9 @@ describe('automatic listener and loader composition', () => {
     await fiber.dispose()
 
     const session = conversation(4)
-    await preStep(ctx, agent(session, MODEL))
+    await postStep(ctx, agent(session, MODEL))
     expect(session.events.some(event => event.type === 'compact/start')).toBe(false)
+    expect(await recover(ctx, agent(session, MODEL), overflow())).toEqual({ action: 'fail' })
   })
 })