mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
a0268d2509
Automatic compaction fires mid-conversation, right after the loop warmed the provider's KV cache with the last routed request. The default summarizer then issued a separate request whose prefix shared nothing with that warm request — a bespoke summarizer system prompt followed by the older history flattened to one rendered transcript string — so a differing first token invalidated the entire cached prefix and every compaction re-processed the whole replayed history twice. Move the compaction directive from the FRONT (a fresh system prompt) to the END (a trailing user message), and replay the last routed request's own system prompt, tools, message prefix, and shadowed-region messages verbatim via session.requestHeader() + deriveEventMessage. The auxiliary call is now a genuine prefix-extension of the warm request, so the provider reuses the cached tokens up to the trailing instruction. SummarizationInput carries the replayed prefix instead of a flat string; the now-unused renderTranscript/renderContentBlocks path is removed with its spec. Cache reuse is best-effort (head compaction guarantees a hit; a mid-range compaction or a differently-routed summarizer forgoes it), correctness is not.
364 lines
14 KiB
TypeScript
364 lines
14 KiB
TypeScript
import { describe, expect, it } from 'vitest'
|
|
import { Context } from 'cordis'
|
|
import { toolPairingBalancedAfter, toolPairingBalancedBefore } from '@deepseek-ai/dsh-compact'
|
|
import { CONTEXT_WINDOW_EXCEEDED_CODE, LlmError } from '@deepseek-ai/dsh-llm'
|
|
import type { ContentBlock, GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
|
import { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
|
|
import { defineTool } from '@deepseek-ai/dsh-tools'
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
|
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
|
import * as Invariants from '@deepseek-ai/dsh-invariants'
|
|
import { BasicCompactService } from '@deepseek-ai/dsh-compact-basic'
|
|
import TokenMeterService from '@deepseek-ai/dsh-token-meter'
|
|
import * as LlmRetry from '@deepseek-ai/dsh-llm-retry'
|
|
import { SessionId, type SurfaceEvent } from '@deepseek-ai/dsh-session'
|
|
|
|
/**
|
|
* CBR-001 regression through the real loop. A replacement checkpoint has a high
|
|
* log seq at the surface head and carries no tool pair, so both adjacent cuts
|
|
* must be safe and re-compacting that checkpoint alone must succeed. This pins
|
|
* surface-position semantics rather than raw-log scanning.
|
|
*/
|
|
|
|
class ReproCompactService extends BasicCompactService {
|
|
override async summarize(): Promise<{ summary: ContentBlock[]; provider: string; model: string }> {
|
|
return {
|
|
summary: [{ type: 'text', text: 'CHECKPOINT SUMMARY' }],
|
|
provider: 'mock',
|
|
model: 'stub',
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Each call emits one tool-call until exhausted, then a final text answer. */
|
|
class StepwiseToolAdapter extends LlmAdapter {
|
|
calls = 0
|
|
constructor(private toolSteps: number) {
|
|
super()
|
|
}
|
|
|
|
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
|
const n = this.calls
|
|
this.calls += 1
|
|
if (n < this.toolSteps) {
|
|
const id = CallId(`c${n}`)
|
|
const args = `{"i":${n}}`
|
|
yield { type: 'block-start', index: 0, blockType: 'text' }
|
|
yield { type: 'block-end', index: 0, block: { type: 'text', text: `step ${n}` } }
|
|
yield { type: 'block-start', index: 1, blockType: 'tool-call' }
|
|
yield { type: 'block-end', index: 1, block: { type: 'tool-call', id, name: 'work', arguments: args } }
|
|
yield { type: 'finish', reason: { kind: 'tool-calls' } }
|
|
return
|
|
}
|
|
yield { type: 'block-start', index: 0, blockType: 'text' }
|
|
yield { type: 'block-end', index: 0, block: { type: 'text', text: 'all done' } }
|
|
yield { type: 'finish', reason: { kind: 'stop' } }
|
|
}
|
|
}
|
|
|
|
/** First conversation request overflows, then the rebuilt retry succeeds. */
|
|
class OverflowRecoveryAdapter extends LlmAdapter {
|
|
readonly conversationRequests: GenerateOptions[] = []
|
|
readonly summaryRequests: GenerateOptions[] = []
|
|
|
|
constructor(
|
|
private readonly delivery: 'thrown' | 'in-band',
|
|
private readonly transientAfterOverflow = false,
|
|
) {
|
|
super()
|
|
}
|
|
|
|
override async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
|
// The cache-reusing summarizer replays the conversation prefix and marks
|
|
// its call only by the compaction instruction in the trailing user message.
|
|
const trailing = options.messages.at(-1)?.content
|
|
.map(block => (block.type === 'text' ? block.text : ''))
|
|
.join('') ?? ''
|
|
if (trailing.includes('acting as a compaction engine')) {
|
|
this.summaryRequests.push(options)
|
|
yield { type: 'block-start', index: 0, blockType: 'text' }
|
|
yield { type: 'block-end', index: 0, block: { type: 'text', text: 'RECOVERY CHECKPOINT' } }
|
|
yield { type: 'finish', reason: { kind: 'stop' } }
|
|
return
|
|
}
|
|
|
|
this.conversationRequests.push(options)
|
|
if (this.conversationRequests.length === 1) {
|
|
if (this.delivery === 'thrown') {
|
|
throw new LlmError('request too large for model context', CONTEXT_WINDOW_EXCEEDED_CODE)
|
|
}
|
|
yield {
|
|
type: 'finish',
|
|
reason: {
|
|
kind: 'error',
|
|
failure: {
|
|
message: 'request too large for model context',
|
|
code: CONTEXT_WINDOW_EXCEEDED_CODE,
|
|
},
|
|
},
|
|
}
|
|
return
|
|
}
|
|
if (this.transientAfterOverflow && this.conversationRequests.length === 2) {
|
|
throw new LlmError('temporary provider outage', 'SERVER')
|
|
}
|
|
yield { type: 'block-start', index: 0, blockType: 'text' }
|
|
yield { type: 'block-end', index: 0, block: { type: 'text', text: 'recovered' } }
|
|
yield { type: 'finish', reason: { kind: 'stop' } }
|
|
}
|
|
}
|
|
|
|
async function harness(toolSteps: number): Promise<{ ctx: Context; compact: ReproCompactService }> {
|
|
const ctx = new Context()
|
|
await mountAgentLoopTestDependencies(ctx)
|
|
await ctx.plugin(Invariants)
|
|
await ctx.plugin(AgentLoop, { agents: [] })
|
|
await ctx.plugin(TokenMeterService, { contextWindow: 400 })
|
|
ctx.llm.registerAdapter(['mock'], new StepwiseToolAdapter(toolSteps))
|
|
ctx.tools.register(defineTool({
|
|
name: 'work',
|
|
description: 'does work',
|
|
parameters: { i: { type: 'number' } },
|
|
async execute() {
|
|
return [{ type: 'text', text: 'work result' }]
|
|
},
|
|
}))
|
|
// Small window so several tool steps cross the threshold and compaction
|
|
// fires within the runaway turn after enough history can shrink.
|
|
const compact = new ReproCompactService(ctx, {
|
|
auto: true,
|
|
thresholdRatio: 0.5,
|
|
retainTokens: 50,
|
|
summarizationModel: '',
|
|
maxTokens: 8192,
|
|
compactionRetries: 1,
|
|
})
|
|
return { ctx, compact }
|
|
}
|
|
|
|
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
|
|
return new Promise((resolve) => {
|
|
const dispose = ctx.on('agent/status', (subject, status) => {
|
|
if (subject === agent && status === 'idle') {
|
|
dispose()
|
|
resolve()
|
|
}
|
|
})
|
|
})
|
|
}
|
|
|
|
function seedOverflowHistory(agent: Agent): void {
|
|
for (let turn = 1; turn <= 2; turn += 1) {
|
|
const sentinel = turn === 1 ? 'OLD HISTORY SENTINEL' : 'RECENT HISTORY'
|
|
agent.session.append('turn/start', {
|
|
turn,
|
|
trigger: { kind: 'message', source: { kind: 'user' } },
|
|
})
|
|
agent.session.append('user/message', {
|
|
content: [{ type: 'text', text: `${sentinel} ${'old context '.repeat(200)}` }],
|
|
source: { kind: 'user' },
|
|
}, { surfaceOp: 'append' })
|
|
agent.session.append('step/start', { turn, step: 1 })
|
|
agent.session.append('assistant/message', {
|
|
provenance: { provider: 'mock', model: 'mock' },
|
|
turn,
|
|
step: 1,
|
|
content: [{ type: 'text', text: `historical response ${turn} ${'detail '.repeat(200)}` }],
|
|
}, { surfaceOp: 'append' })
|
|
agent.session.append('step/end', { turn, step: 1 })
|
|
agent.session.append('turn/end', { turn, reason: { kind: 'completed' } })
|
|
}
|
|
}
|
|
|
|
describe('CBR-001: a real-loop checkpoint is a valid boundary on both sides', () => {
|
|
it('uses the model actually routed by agent/request for post-step pressure', async () => {
|
|
const { ctx } = await harness(8)
|
|
ctx.on('agent/request', async (_agent, _turn, _step, config) => ({ ...config, provider: 'mock', model: 'mock' }))
|
|
try {
|
|
const agent = ctx.agentLoop.create(SessionId('routed-pressure'), {
|
|
provider: 'unconfigured-agent-fallback',
|
|
model: 'unconfigured-agent-fallback',
|
|
})
|
|
agent.send([{ type: 'text', text: 'do a routed multi-step task' }])
|
|
await waitForIdle(ctx, agent)
|
|
|
|
expect(agent.session.requestHeader()?.config.model).toBe('mock')
|
|
expect(agent.session.events.some(event => event.type === 'compact/summary')).toBe(true)
|
|
expect(agent.session.events.at(-1)).toMatchObject({
|
|
type: 'turn/end',
|
|
data: { reason: { kind: 'completed' } },
|
|
})
|
|
} finally {
|
|
await ctx.fiber.dispose()
|
|
}
|
|
})
|
|
|
|
it('runs automatic pressure after the current tool result and before step/end', async () => {
|
|
const { ctx } = await harness(8)
|
|
try {
|
|
const agent = ctx.agentLoop.create(SessionId('post-step-order'), { provider: 'mock', model: 'mock' })
|
|
agent.send([{ type: 'text', text: 'do tool work' }])
|
|
await waitForIdle(ctx, agent)
|
|
|
|
const events = [...agent.session.events]
|
|
const compactStart = events.find(event => event.type === 'compact/start')
|
|
expect(compactStart).toBeDefined()
|
|
const precedingResult = events.findLast(event =>
|
|
event.type === 'tool/result' && event.seq < compactStart!.seq,
|
|
)
|
|
if (precedingResult?.type !== 'tool/result') throw new Error('expected a durable tool result before compaction')
|
|
const stepEnd = events.find(event =>
|
|
event.type === 'step/end'
|
|
&& event.data.step === precedingResult.data.step
|
|
&& event.seq > compactStart!.seq,
|
|
)
|
|
expect(precedingResult.seq).toBeLessThan(compactStart!.seq)
|
|
expect(compactStart!.seq).toBeLessThan(stepEnd!.seq)
|
|
} finally {
|
|
await ctx.fiber.dispose()
|
|
}
|
|
})
|
|
|
|
it('the head checkpoint the loop lands is a balanced cut on both sides', async () => {
|
|
const { ctx } = await harness(8)
|
|
try {
|
|
const agent = ctx.agentLoop.create(SessionId('repro'), { provider: 'mock', model: 'mock' })
|
|
agent.send([{ type: 'text', text: 'do a long multi-step task' }])
|
|
await waitForIdle(ctx, agent)
|
|
|
|
const events = [...agent.session.events]
|
|
// A compaction ran: at least one checkpoint landed on the surface.
|
|
const checkpoints = events.filter(
|
|
(e): e is SurfaceEvent =>
|
|
e.type === 'user/message'
|
|
&& typeof (e as SurfaceEvent).surfaceOp === 'object',
|
|
)
|
|
expect(checkpoints.length).toBeGreaterThan(0)
|
|
|
|
// High log position does not make a text-only checkpoint mid-step; both
|
|
// its start and end cuts are balanced in surface order.
|
|
const nodes = agent.session.surface.nodes
|
|
for (const cp of checkpoints) {
|
|
const index = nodes.indexOf(cp.seq)
|
|
if (index === -1) continue // shadowed by a later checkpoint — no longer an edge.
|
|
expect(toolPairingBalancedBefore(agent.session, cp.seq),
|
|
`checkpoint seq ${cp.seq} must be a balanced region START`).toBe(true)
|
|
expect(toolPairingBalancedAfter(agent.session, cp.seq),
|
|
`checkpoint seq ${cp.seq} must be a balanced region END`).toBe(true)
|
|
}
|
|
} finally {
|
|
await ctx.fiber.dispose()
|
|
}
|
|
})
|
|
})
|
|
|
|
describe('context-overflow recovery across the real loop and compact-basic', () => {
|
|
it.each(['thrown', 'in-band'] as const)(
|
|
'force-compacts a %s overflow between failed and retry steps',
|
|
async (delivery) => {
|
|
const ctx = new Context()
|
|
const adapter = new OverflowRecoveryAdapter(delivery)
|
|
await mountAgentLoopTestDependencies(ctx)
|
|
await ctx.plugin(Invariants)
|
|
await ctx.plugin(AgentLoop, { agents: [] })
|
|
await ctx.plugin(TokenMeterService, { contextWindow: 128 })
|
|
ctx.llm.registerAdapter(['mock'], adapter)
|
|
ctx.on('agent/request', async (_agent, _turn, _step, config) => ({ ...config, provider: 'mock', model: 'mock' }))
|
|
await ctx.plugin(BasicCompactService, {
|
|
thresholdRatio: 1,
|
|
retainTokens: 100,
|
|
maxTokens: 64,
|
|
compactionRetries: 0,
|
|
maxOverflowRetries: 1,
|
|
})
|
|
|
|
try {
|
|
const agent = ctx.agentLoop.create(SessionId(`overflow-${delivery}`), {
|
|
provider: 'unconfigured-agent-fallback',
|
|
model: 'unconfigured-agent-fallback',
|
|
})
|
|
seedOverflowHistory(agent)
|
|
|
|
agent.send([{ type: 'text', text: 'continue from history' }])
|
|
await agent.whenIdle()
|
|
|
|
expect(adapter.conversationRequests).toHaveLength(2)
|
|
expect(adapter.summaryRequests).toHaveLength(1)
|
|
expect(JSON.stringify(adapter.conversationRequests[0]!.messages)).toContain('OLD HISTORY SENTINEL')
|
|
const retry = JSON.stringify(adapter.conversationRequests[1]!.messages)
|
|
expect(retry).toContain('RECOVERY CHECKPOINT')
|
|
expect(retry).not.toContain('OLD HISTORY SENTINEL')
|
|
|
|
const events = [...agent.session.events]
|
|
const failedEnd = events.find(event =>
|
|
event.type === 'step/end' && event.data.turn === 3 && event.data.step === 1,
|
|
)!
|
|
const retryStart = events.find(event =>
|
|
event.type === 'step/start' && event.data.turn === 3 && event.data.step === 2,
|
|
)!
|
|
const compaction = events.filter(event =>
|
|
event.type === 'compact/start'
|
|
|| event.type === 'compact/summary'
|
|
|| event.type === 'compact/end',
|
|
)
|
|
expect(compaction.map(event => event.type)).toEqual([
|
|
'compact/start',
|
|
'compact/summary',
|
|
'compact/end',
|
|
])
|
|
expect(compaction.every(event => event.seq > failedEnd.seq && event.seq < retryStart.seq)).toBe(true)
|
|
expect(events.at(-1)).toMatchObject({
|
|
type: 'turn/end',
|
|
data: { reason: { kind: 'completed' } },
|
|
})
|
|
} finally {
|
|
await ctx.fiber.dispose()
|
|
}
|
|
},
|
|
)
|
|
|
|
it('keeps context-overflow and transient retry budgets independent in one sequence', async () => {
|
|
const ctx = new Context()
|
|
const adapter = new OverflowRecoveryAdapter('thrown', true)
|
|
await mountAgentLoopTestDependencies(ctx)
|
|
await ctx.plugin(Invariants)
|
|
await ctx.plugin(LlmRetry, {
|
|
maxTransientRetries: 1,
|
|
initialDelayMs: 1,
|
|
maxDelayMs: 1,
|
|
jitterRatio: 0,
|
|
})
|
|
await ctx.plugin(AgentLoop, { agents: [] })
|
|
await ctx.plugin(TokenMeterService, { contextWindow: 128 })
|
|
ctx.llm.registerAdapter(['mock'], adapter)
|
|
await ctx.plugin(BasicCompactService, {
|
|
thresholdRatio: 1,
|
|
retainTokens: 100,
|
|
maxTokens: 64,
|
|
compactionRetries: 0,
|
|
maxOverflowRetries: 1,
|
|
})
|
|
|
|
try {
|
|
const agent = ctx.agentLoop.create(SessionId('alternating-recovery'), { provider: 'mock', model: 'mock' })
|
|
seedOverflowHistory(agent)
|
|
agent.send([{ type: 'text', text: 'continue from history' }])
|
|
await agent.whenIdle()
|
|
|
|
expect(adapter.conversationRequests).toHaveLength(3)
|
|
expect(adapter.summaryRequests).toHaveLength(1)
|
|
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data))
|
|
.toEqual([expect.objectContaining({ step: 2, retry: 1, failure: { message: 'temporary provider outage', code: 'SERVER' } })])
|
|
expect(agent.session.events.filter(event => event.type === 'step/start').slice(-3).map(event => event.data.step))
|
|
.toEqual([1, 2, 3])
|
|
expect(agent.session.events.at(-1)).toMatchObject({
|
|
type: 'turn/end',
|
|
data: { reason: { kind: 'completed' } },
|
|
})
|
|
} finally {
|
|
await ctx.fiber.dispose()
|
|
}
|
|
})
|
|
})
|