Merge remote-tracking branch 'origin/master' into worktree/llm-reasoning-effort
# Conflicts: # docs/architecture.i18n.yaml # docs/config-catalog.md
This commit is contained in:
Yichen Jiang
@@ -41,8 +41,11 @@
|
||||
"@cordisjs/plugin-loader": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent-loop": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent-loop-testkit": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm-deepseek": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm-mock-server": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence-sqlite": "workspace:^",
|
||||
|
||||
234
packages/llm/llm-retry/tests/transport-recovery.spec.ts
Normal file
234
packages/llm/llm-retry/tests/transport-recovery.spec.ts
Normal file
@@ -0,0 +1,234 @@
|
||||
import { createServer } from 'node:http'
|
||||
import type { AddressInfo } from 'node:net'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
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 LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import type { MockLlmBehavior, MockLlmServer } from '@deepseek-ai/dsh-llm-mock-server'
|
||||
import { startMockLlmServer } from '@deepseek-ai/dsh-llm-mock-server'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import * as Retry from '../src/index.ts'
|
||||
|
||||
let context: Context | undefined
|
||||
const servers: MockLlmServer[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await context?.fiber.dispose()
|
||||
context = undefined
|
||||
await Promise.all(servers.splice(0).map(server => server.close()))
|
||||
})
|
||||
|
||||
async function start(
|
||||
sequence: readonly MockLlmBehavior[],
|
||||
options: Omit<Parameters<typeof startMockLlmServer>[0], 'sequence'> = {},
|
||||
): Promise<MockLlmServer> {
|
||||
const server = await startMockLlmServer({ sequence, ...options })
|
||||
servers.push(server)
|
||||
return server
|
||||
}
|
||||
|
||||
async function harness(
|
||||
baseURL: string,
|
||||
options: { streamIdleTimeoutMs?: number; initialDelayMs?: number } = {},
|
||||
): Promise<Context> {
|
||||
const ctx = new Context()
|
||||
await mountAgentLoopTestDependencies(ctx)
|
||||
await ctx.plugin(LlmDeepSeek, {
|
||||
apiKey: 'mock-key',
|
||||
baseURL,
|
||||
streamIdleTimeoutMs: options.streamIdleTimeoutMs ?? 1_000,
|
||||
})
|
||||
await ctx.plugin(Retry, {
|
||||
maxTransientRetries: 2,
|
||||
initialDelayMs: options.initialDelayMs ?? 10,
|
||||
maxDelayMs: options.initialDelayMs ?? 10,
|
||||
jitterRatio: 0,
|
||||
})
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
return ctx
|
||||
}
|
||||
|
||||
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') return
|
||||
dispose()
|
||||
resolve()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function sendAndWait(ctx: Context, agent: Agent): Promise<void> {
|
||||
const idle = waitForIdle(ctx, agent)
|
||||
agent.followup([{ type: 'text', text: 'recover through the provider boundary' }])
|
||||
return idle
|
||||
}
|
||||
|
||||
function finalAssistantText(agent: Agent): string | undefined {
|
||||
const message = agent.session.deriveMessages().at(-1)
|
||||
if (message?.role !== 'assistant') return undefined
|
||||
return message.content
|
||||
.filter(block => block.type === 'text')
|
||||
.map(block => block.text)
|
||||
.join('')
|
||||
}
|
||||
|
||||
async function unusedPort(): Promise<number> {
|
||||
const server = createServer()
|
||||
await new Promise<void>((resolve) => { server.listen(0, '127.0.0.1', resolve) })
|
||||
const port = (server.address() as AddressInfo).port
|
||||
await new Promise<void>((resolve) => { server.close(() => { resolve() }) })
|
||||
return port
|
||||
}
|
||||
|
||||
describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
|
||||
it('recovers from a true refused connection after the endpoint starts during backoff', async () => {
|
||||
const port = await unusedPort()
|
||||
context = await harness(`http://127.0.0.1:${port}`, { initialDelayMs: 100 })
|
||||
const agent = context.agentLoop.create(SessionId('wire-refused'), {
|
||||
provider: 'deepseek',
|
||||
model: 'mock-model',
|
||||
})
|
||||
let recoveryServer: Promise<MockLlmServer> | undefined
|
||||
context.on('session/event', (session, event) => {
|
||||
if (session !== agent.session || event.type !== 'llm/retry' || event.data.retry !== 1) return
|
||||
recoveryServer = start(['success'], { port, apiKey: 'mock-key', successText: 'connected after retry' })
|
||||
})
|
||||
|
||||
await sendAndWait(context, agent)
|
||||
const server = await recoveryServer
|
||||
|
||||
expect(server).toBeDefined()
|
||||
expect(server?.requests).toHaveLength(1)
|
||||
expect(agent.session.events.filter(event => event.type === 'step/start').map(event => event.data.step))
|
||||
.toEqual([1, 2])
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
|
||||
.toEqual(['TRANSPORT'])
|
||||
expect(finalAssistantText(agent)).toBe('connected after retry')
|
||||
})
|
||||
|
||||
it.each([
|
||||
['stream_disconnect', 0] as const,
|
||||
['partial_disconnect', 2] as const,
|
||||
])('retries %s without committing failed chunks', async (behavior, failedChunkCount) => {
|
||||
const server = await start([behavior, 'success'], {
|
||||
apiKey: 'mock-key',
|
||||
partialText: 'discard me',
|
||||
chunkSize: 100,
|
||||
disconnectDelayMs: 20,
|
||||
successText: 'recovered response',
|
||||
})
|
||||
context = await harness(server.baseURL)
|
||||
const agent = context.agentLoop.create(SessionId(`wire-${behavior}`), {
|
||||
provider: 'deepseek',
|
||||
model: 'mock-model',
|
||||
})
|
||||
|
||||
await sendAndWait(context, agent)
|
||||
|
||||
expect(server.requests).toHaveLength(2)
|
||||
expect(server.requests[0]?.body).toEqual(server.requests[1]?.body)
|
||||
expect(agent.session.events.filter(event =>
|
||||
event.type === 'assistant/chunk' && event.data.step === 1,
|
||||
)).toHaveLength(failedChunkCount)
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
|
||||
.toEqual([2])
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
|
||||
.toEqual(['TRANSPORT'])
|
||||
expect(finalAssistantText(agent)).toBe('recovered response')
|
||||
})
|
||||
|
||||
it('retries a wire-valid content-less completion without committing an empty message', async () => {
|
||||
const server = await start(['empty', 'success'], {
|
||||
apiKey: 'mock-key',
|
||||
successText: 'recovered from empty',
|
||||
})
|
||||
context = await harness(server.baseURL)
|
||||
const agent = context.agentLoop.create(SessionId('wire-empty'), {
|
||||
provider: 'deepseek',
|
||||
model: 'mock-model',
|
||||
})
|
||||
|
||||
await sendAndWait(context, agent)
|
||||
|
||||
expect(server.requests).toHaveLength(2)
|
||||
expect(server.requests[0]?.body).toEqual(server.requests[1]?.body)
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
|
||||
.toEqual(['EMPTY_RESPONSE'])
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
|
||||
.toEqual([2])
|
||||
expect(agent.session.events.at(-1)).toMatchObject({
|
||||
type: 'turn/end',
|
||||
data: { reason: { kind: 'completed' } },
|
||||
})
|
||||
expect(finalAssistantText(agent)).toBe('recovered from empty')
|
||||
})
|
||||
|
||||
it('exposes a clean partial EOF as non-default-retryable STREAM_CLOSED', async () => {
|
||||
const server = await start(['partial_eof', 'success'], {
|
||||
apiKey: 'mock-key',
|
||||
partialText: 'discarded clean eof',
|
||||
chunkSize: 100,
|
||||
})
|
||||
context = await harness(server.baseURL)
|
||||
const agent = context.agentLoop.create(SessionId('wire-partial-eof'), {
|
||||
provider: 'deepseek',
|
||||
model: 'mock-model',
|
||||
})
|
||||
|
||||
await sendAndWait(context, agent)
|
||||
|
||||
expect(server.requests).toHaveLength(1)
|
||||
expect(agent.session.events.filter(event =>
|
||||
event.type === 'assistant/chunk' && event.data.step === 1,
|
||||
)).toHaveLength(2)
|
||||
expect(agent.session.events.some(event => event.type === 'assistant/message')).toBe(false)
|
||||
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
expect(agent.session.events.at(-1)).toMatchObject({
|
||||
type: 'turn/end',
|
||||
data: { reason: { kind: 'error', failure: { code: 'STREAM_CLOSED' } } },
|
||||
})
|
||||
})
|
||||
|
||||
it('turns a stalled body into TIMEOUT and succeeds on the next request', async () => {
|
||||
const server = await start(['stall', 'success'], {
|
||||
apiKey: 'mock-key',
|
||||
successText: 'recovered after timeout',
|
||||
})
|
||||
context = await harness(server.baseURL, { streamIdleTimeoutMs: 30 })
|
||||
const agent = context.agentLoop.create(SessionId('wire-stall'), {
|
||||
provider: 'deepseek',
|
||||
model: 'mock-model',
|
||||
})
|
||||
|
||||
await sendAndWait(context, agent)
|
||||
|
||||
expect(server.requests.map(record => record.behavior)).toEqual(['stall', 'success'])
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
|
||||
.toEqual(['TIMEOUT'])
|
||||
expect(finalAssistantText(agent)).toBe('recovered after timeout')
|
||||
})
|
||||
|
||||
it('stops after the configured transport retry budget is exhausted', async () => {
|
||||
const server = await start(['connection_reset', 'connection_reset', 'connection_reset'], {
|
||||
apiKey: 'mock-key',
|
||||
})
|
||||
context = await harness(server.baseURL)
|
||||
const agent = context.agentLoop.create(SessionId('wire-exhausted'), {
|
||||
provider: 'deepseek',
|
||||
model: 'mock-model',
|
||||
})
|
||||
|
||||
await sendAndWait(context, agent)
|
||||
|
||||
expect(server.requests).toHaveLength(3)
|
||||
expect(agent.session.events.filter(event => event.type === 'step/start')).toHaveLength(3)
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(2)
|
||||
expect(agent.session.events.at(-1)).toMatchObject({
|
||||
type: 'turn/end',
|
||||
data: { reason: { kind: 'error', failure: { code: 'TRANSPORT' } } },
|
||||
})
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user