Add two DeepSeek LLM adapters: dsh-llm-deepseek and dsh-llm-pi-ai
The first real LlmAdapter implementations, shipped as a deliberate pair:
same models and wire protocol, completely different internals, so the
StreamChunk protocol is verified across independent implementations.
- dsh-llm-deepseek: hand-rolled fetch + SSE parser + chunk-translation
state machine against the official chat-completions format (thinking
mode via top-level thinking/reasoning_effort; the empty-string
reasoning_content first chunk; usage attached to the finish chunk or
trailing; reasoning_content passback on tool-call turns; disjoint
cache-token accounting).
- dsh-llm-pi-ai: the same endpoint through @earendil-works/pi-ai,
mapping its event vocabulary (parsed tool arguments, in-stream error
events, folded reasoning tokens) onto the same chunks.
The agent loop now honors the in-band error path: an adapter that ends
its stream with finish {kind:error|aborted} (the only option for
adapters that can't throw mid-stream, like pi-ai) is translated into a
step error, so the turn ends error/aborted with a logged error event
instead of a normal completed assistant message. This makes the
StreamChunk error contract real for both adapters; docs/architecture.md
and the StreamChunk doc are updated accordingly.
New yarn test:e2e (vitest.e2e.config.ts, *.e2e.ts) runs key-gated
real-API matrices for both adapters across V4 Flash/Pro and all
thinking/effort levels; it self-skips without DEEPSEEK_API_KEY. Unit
suites run against local node:http mock SSE servers at 100% per-file
coverage.
This commit is contained in:
Tianyi Cui
130
packages/llm-pi-ai/tests/adapter.e2e.ts
Normal file
130
packages/llm-pi-ai/tests/adapter.e2e.ts
Normal file
@@ -0,0 +1,130 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateResult, Message, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
import type { Config } from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
|
||||
/**
|
||||
* Real-API e2e for the pi-ai-backed adapter: V4 Flash + V4 Pro across all
|
||||
* reasoning levels the adapter exposes (off / high / xhigh→wire 'max').
|
||||
* Mirrors the llm-deepseek matrix so the two independent implementations
|
||||
* verify the same StreamChunk contract. Key-gated.
|
||||
*/
|
||||
|
||||
const FLASH = 'deepseek-v4-flash'
|
||||
const PRO = 'deepseek-v4-pro'
|
||||
|
||||
async function harness(model: string, config: Partial<Config> = {}) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(LlmPiAi, { models: [model], ...config })
|
||||
return ctx
|
||||
}
|
||||
|
||||
function ask(text: string): Message[] {
|
||||
return [{ role: 'user', content: [{ type: 'text', text }] }]
|
||||
}
|
||||
|
||||
function textOf(result: GenerateResult): string {
|
||||
return result.message.content
|
||||
.filter(block => block.type === 'text')
|
||||
.map(block => block.text)
|
||||
.join('')
|
||||
}
|
||||
|
||||
function blockKinds(result: GenerateResult): string[] {
|
||||
return result.message.content.map(block => block.type)
|
||||
}
|
||||
|
||||
const weatherTool: ToolSchema = {
|
||||
name: 'get_weather',
|
||||
description: 'Get the current weather for a city.',
|
||||
parameters: {
|
||||
type: 'object',
|
||||
properties: { city: { type: 'string', description: 'City name' } },
|
||||
required: ['city'],
|
||||
},
|
||||
}
|
||||
|
||||
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () => {
|
||||
it.each([FLASH, PRO])('%s + reasoning off: plain text generation', async (model) => {
|
||||
const ctx = await harness(model, { reasoning: 'off' })
|
||||
const result = await ctx.llm.generate({
|
||||
model,
|
||||
messages: ask('Reply with exactly the word: pong'),
|
||||
maxTokens: 50,
|
||||
})
|
||||
expect(result.finish.kind).toBe('stop')
|
||||
expect(textOf(result).toLowerCase()).toContain('pong')
|
||||
expect(result.message.content.some(block => block.type === 'reasoning')).toBe(false)
|
||||
})
|
||||
|
||||
it.each([FLASH, PRO])('%s + reasoning high: reasoning blocks present', async (model) => {
|
||||
const ctx = await harness(model, { reasoning: 'high' })
|
||||
const result = await ctx.llm.generate({
|
||||
model,
|
||||
messages: ask('Which is larger, 9.11 or 9.8? Answer with just the number.'),
|
||||
maxTokens: 2000,
|
||||
})
|
||||
expect(result.finish.kind).toBe('stop')
|
||||
expect(result.message.content.some(block => block.type === 'reasoning')).toBe(true)
|
||||
expect(textOf(result)).toContain('9.8')
|
||||
})
|
||||
|
||||
it('pro + reasoning xhigh (wire max): tool-call round trip', async () => {
|
||||
const ctx = await harness(PRO, { reasoning: 'xhigh' })
|
||||
|
||||
const first = await ctx.llm.generate({
|
||||
model: PRO,
|
||||
messages: ask('What is the weather in Paris right now? Use the get_weather tool.'),
|
||||
tools: [weatherTool],
|
||||
maxTokens: 2000,
|
||||
})
|
||||
expect(first.finish.kind).toBe('tool-calls')
|
||||
const call = first.message.content.find(block => block.type === 'tool-call')
|
||||
expect(call).toBeDefined()
|
||||
expect(call!.name).toBe('get_weather')
|
||||
expect(JSON.parse(call!.arguments)).toMatchObject({ city: expect.stringMatching(/paris/i) as string })
|
||||
|
||||
const second = await ctx.llm.generate({
|
||||
model: PRO,
|
||||
messages: [
|
||||
...ask('What is the weather in Paris right now? Use the get_weather tool.'),
|
||||
{ role: 'assistant', content: first.message.content },
|
||||
{
|
||||
role: 'user',
|
||||
content: [{
|
||||
type: 'tool-result',
|
||||
toolCallId: CallId(call!.id),
|
||||
content: [{ type: 'text', text: 'Sunny, 22°C' }],
|
||||
}],
|
||||
},
|
||||
],
|
||||
tools: [weatherTool],
|
||||
maxTokens: 2000,
|
||||
})
|
||||
expect(second.finish.kind).toBe('stop')
|
||||
expect(textOf(second).toLowerCase()).toMatch(/sunny|22/)
|
||||
})
|
||||
|
||||
it('produces the same block structure as llm-deepseek for the same prompt', async () => {
|
||||
// Loose structural equivalence between the two independent adapters:
|
||||
// same block KINDS in the same order for a deterministic prompt — the
|
||||
// cross-implementation check that the StreamChunk design holds.
|
||||
const deepseekCtx = new Context()
|
||||
await deepseekCtx.plugin(LlmService)
|
||||
await deepseekCtx.plugin(LlmDeepSeek, { models: [FLASH], thinking: 'disabled' })
|
||||
|
||||
const piCtx = await harness(FLASH, { reasoning: 'off' })
|
||||
|
||||
const prompt = ask('Reply with exactly the word: pong')
|
||||
const [fromDeepSeek, fromPiAi] = await Promise.all([
|
||||
deepseekCtx.llm.generate({ model: FLASH, messages: prompt, maxTokens: 50 }),
|
||||
piCtx.llm.generate({ model: FLASH, messages: prompt, maxTokens: 50 }),
|
||||
])
|
||||
expect(blockKinds(fromPiAi)).toEqual(blockKinds(fromDeepSeek))
|
||||
expect(fromPiAi.finish.kind).toBe(fromDeepSeek.finish.kind)
|
||||
})
|
||||
})
|
||||
354
packages/llm-pi-ai/tests/adapter.spec.ts
Normal file
354
packages/llm-pi-ai/tests/adapter.spec.ts
Normal file
@@ -0,0 +1,354 @@
|
||||
import { createServer } from 'node:http'
|
||||
import type { IncomingMessage, Server, ServerResponse } from 'node:http'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
import { buildModel, PiAiAdapter } from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
|
||||
/** Scripted SSE responses, one per request (OpenAI chat-completions shape). */
|
||||
interface MockServer {
|
||||
url: string
|
||||
requests: unknown[]
|
||||
close(): Promise<void>
|
||||
}
|
||||
|
||||
const servers: Server[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve))))
|
||||
})
|
||||
|
||||
async function mockServer(script: { status?: number; events?: string[]; body?: string }[]): Promise<MockServer> {
|
||||
const requests: unknown[] = []
|
||||
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
|
||||
let body = ''
|
||||
request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') })
|
||||
request.on('end', () => {
|
||||
requests.push(JSON.parse(body))
|
||||
const behavior = script.shift() ?? { status: 500, body: 'script exhausted' }
|
||||
if (behavior.status !== undefined && behavior.status !== 200) {
|
||||
response.writeHead(behavior.status, { 'content-type': 'application/json' })
|
||||
response.end(behavior.body ?? '{}')
|
||||
return
|
||||
}
|
||||
response.writeHead(200, { 'content-type': 'text/event-stream' })
|
||||
for (const event of behavior.events ?? []) response.write(`data: ${event}\n\n`)
|
||||
response.end()
|
||||
})
|
||||
})
|
||||
servers.push(server)
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (address === null || typeof address === 'string') throw new Error('no port')
|
||||
return {
|
||||
url: `http://127.0.0.1:${address.port}`,
|
||||
requests,
|
||||
close: () => new Promise(resolve => server.close(() => { resolve() })),
|
||||
}
|
||||
}
|
||||
|
||||
const textEvents = [
|
||||
'{"choices":[{"delta":{"role":"assistant","content":""},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{"content":"hello"},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{},"index":0,"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}',
|
||||
'[DONE]',
|
||||
]
|
||||
|
||||
const toolEvents = [
|
||||
'{"choices":[{"delta":{"role":"assistant","content":null},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"get_weather","arguments":""}}]},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"city\\":\\"Paris\\"}"}}]},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{},"index":0,"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":20,"completion_tokens":6}}',
|
||||
'[DONE]',
|
||||
]
|
||||
|
||||
const thinkingEvents = [
|
||||
'{"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{"reasoning_content":"pondering"},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{"content":"answer","reasoning_content":null},"index":0,"finish_reason":null}]}',
|
||||
'{"choices":[{"delta":{},"index":0,"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":9}}',
|
||||
'[DONE]',
|
||||
]
|
||||
|
||||
async function harness(baseURL: string, config: object = {}) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(LlmPiAi, { apiKey: 'test-key', baseURL, models: ['deepseek-v4-flash'], ...config })
|
||||
return ctx
|
||||
}
|
||||
|
||||
describe('PiAiAdapter against a mock server', () => {
|
||||
it('streams a text generation through ctx.llm.generate', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url)
|
||||
|
||||
const result = await ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
|
||||
})
|
||||
expect(result.message.content).toEqual([{ type: 'text', text: 'hello' }])
|
||||
expect(result.finish).toEqual({ kind: 'stop' })
|
||||
expect(result.usage).toMatchObject({ inputTokens: 3, outputTokens: 1 })
|
||||
})
|
||||
|
||||
it('streams tool calls with re-stringified arguments', async () => {
|
||||
const server = await mockServer([{ events: toolEvents }])
|
||||
const ctx = await harness(server.url)
|
||||
|
||||
const result = await ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'weather?' }] }],
|
||||
tools: [{
|
||||
name: 'get_weather',
|
||||
description: 'Get weather',
|
||||
parameters: { type: 'object', properties: { city: { type: 'string' } } },
|
||||
}],
|
||||
})
|
||||
expect(result.finish).toEqual({ kind: 'tool-calls' })
|
||||
const call = result.message.content.find(block => block.type === 'tool-call')
|
||||
expect(call).toMatchObject({ name: 'get_weather', arguments: '{"city":"Paris"}' })
|
||||
})
|
||||
|
||||
it('maps reasoning_content streams to reasoning blocks', async () => {
|
||||
const server = await mockServer([{ events: thinkingEvents }])
|
||||
const ctx = await harness(server.url, { reasoning: 'high' })
|
||||
|
||||
const result = await ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'think' }] }],
|
||||
})
|
||||
expect(result.message.content).toEqual([
|
||||
{ type: 'reasoning', text: 'pondering' },
|
||||
{ type: 'text', text: 'answer' },
|
||||
])
|
||||
})
|
||||
|
||||
it('sends DeepSeek thinking fields when reasoning is configured', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url, { reasoning: 'xhigh' })
|
||||
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
|
||||
expect(server.requests[0]).toMatchObject({
|
||||
thinking: { type: 'enabled' },
|
||||
reasoning_effort: 'max', // xhigh maps to max via thinkingLevelMap
|
||||
})
|
||||
})
|
||||
|
||||
it('disables thinking for reasoning: off', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url, { reasoning: 'off' })
|
||||
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
|
||||
expect(server.requests[0]).toMatchObject({ thinking: { type: 'disabled' } })
|
||||
})
|
||||
|
||||
it('injects stop sequences through onPayload', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url)
|
||||
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [], stop: ['END'] })
|
||||
expect(server.requests[0]).toMatchObject({ stop: ['END'] })
|
||||
})
|
||||
|
||||
it('maps HTTP errors to error finish chunks (pi-ai in-stream style)', async () => {
|
||||
const server = await mockServer([{
|
||||
status: 401,
|
||||
body: JSON.stringify({ error: { message: 'bad key' } }),
|
||||
}])
|
||||
const ctx = await harness(server.url)
|
||||
const result = await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
|
||||
expect(result.finish.kind).toBe('error')
|
||||
expect((result.finish as { message: string }).message).toMatch(/bad key|401/)
|
||||
})
|
||||
|
||||
it('rejects prefill with UNSUPPORTED', async () => {
|
||||
const ctx = await harness('http://127.0.0.1:1')
|
||||
await expect(ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [],
|
||||
prefill: [{ type: 'text', text: 'Sure' }],
|
||||
})).rejects.toThrow(LlmError)
|
||||
})
|
||||
|
||||
it('registers/unregisters models on the llm service (HMR safety)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
const fiber = await ctx.plugin(LlmPiAi, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' })
|
||||
expect(ctx.llm.models().sort()).toEqual(['deepseek-v4-flash', 'deepseek-v4-pro'])
|
||||
await fiber.dispose()
|
||||
expect(ctx.llm.models()).toEqual([])
|
||||
})
|
||||
|
||||
it('throws a clear error when no API key is available', async () => {
|
||||
const previous = process.env.DEEPSEEK_API_KEY
|
||||
delete process.env.DEEPSEEK_API_KEY
|
||||
try {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await expect(ctx.plugin(LlmPiAi, {})).rejects.toThrow(/an API key is required/)
|
||||
} finally {
|
||||
if (previous !== undefined) process.env.DEEPSEEK_API_KEY = previous
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('option spreads and env fallbacks', () => {
|
||||
it('forwards temperature, maxTokens, and signal', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url)
|
||||
const controller = new AbortController()
|
||||
await ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [],
|
||||
temperature: 0.5,
|
||||
maxTokens: 40,
|
||||
signal: controller.signal,
|
||||
})
|
||||
expect(server.requests[0]).toMatchObject({ temperature: 0.5, max_tokens: 40 })
|
||||
})
|
||||
|
||||
it('falls back to DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL env vars', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'env-key')
|
||||
vi.stubEnv('DEEPSEEK_BASE_URL', server.url)
|
||||
try {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(LlmPiAi, { models: ['deepseek-v4-flash'] })
|
||||
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
|
||||
expect(server.requests).toHaveLength(1)
|
||||
} finally {
|
||||
vi.unstubAllEnvs()
|
||||
}
|
||||
})
|
||||
|
||||
it('defaults to the public base URL without config or env', async () => {
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'k')
|
||||
vi.stubEnv('DEEPSEEK_BASE_URL', undefined)
|
||||
try {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(LlmPiAi, {})
|
||||
expect(ctx.llm.models().length).toBeGreaterThan(0)
|
||||
} finally {
|
||||
vi.unstubAllEnvs()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('buildModel', () => {
|
||||
it('builds a DeepSeek-compat openai-completions model descriptor', () => {
|
||||
const model = buildModel('deepseek-v4-pro', { apiKey: 'k', baseURL: 'http://x', reasoning: 'high' })
|
||||
expect(model).toMatchObject({
|
||||
id: 'deepseek-v4-pro',
|
||||
api: 'openai-completions',
|
||||
provider: 'deepseek',
|
||||
baseUrl: 'http://x',
|
||||
reasoning: true,
|
||||
compat: { thinkingFormat: 'deepseek', requiresReasoningContentOnAssistantMessages: true },
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps reasoning true even for off (pi-ai gates the thinking field on it)', () => {
|
||||
// 'off' yields {thinking: {type: 'disabled'}} on the wire — pi-ai only
|
||||
// emits the field at all when model.reasoning is true.
|
||||
expect(buildModel('m', { apiKey: 'k', baseURL: 'http://x', reasoning: 'off' }).reasoning).toBe(true)
|
||||
})
|
||||
|
||||
it('adapter is constructible directly for embedding', () => {
|
||||
expect(new PiAiAdapter({ apiKey: 'k', baseURL: 'http://x' })).toBeInstanceOf(PiAiAdapter)
|
||||
})
|
||||
})
|
||||
|
||||
describe('review fixes', () => {
|
||||
it('defaults omitted reasoning config to thinking ENABLED (provider default)', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url) // no reasoning key at all
|
||||
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
|
||||
expect(server.requests[0]).toMatchObject({
|
||||
thinking: { type: 'enabled' },
|
||||
reasoning_effort: 'high',
|
||||
})
|
||||
})
|
||||
|
||||
it('replays reasoning_content on assistant tool-call turns (passback rule)', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url)
|
||||
await ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [
|
||||
{ role: 'user', content: [{ type: 'text', text: 'weather?' }] },
|
||||
{
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'reasoning', text: 'I should check.' },
|
||||
{ type: 'tool-call', id: CallId('c1'), name: 'get_weather', arguments: '{"city":"Paris"}' },
|
||||
],
|
||||
},
|
||||
{
|
||||
role: 'user',
|
||||
content: [{ type: 'tool-result', toolCallId: CallId('c1'), content: [{ type: 'text', text: 'Sunny' }] }],
|
||||
},
|
||||
],
|
||||
})
|
||||
const request = server.requests[0] as { messages: { role: string; reasoning_content?: string }[] }
|
||||
const assistant = request.messages.find(message => message.role === 'assistant')
|
||||
expect(assistant?.reasoning_content).toBe('I should check.')
|
||||
})
|
||||
|
||||
it('aborts the upstream request when the consumer stops streaming early', async () => {
|
||||
// Slow server: write one chunk, then hold the connection open and record
|
||||
// whether the socket closes (the adapter must cancel on early break).
|
||||
let socketClosed = false
|
||||
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
|
||||
request.on('data', () => undefined)
|
||||
request.on('end', () => {
|
||||
response.writeHead(200, { 'content-type': 'text/event-stream' })
|
||||
response.write(`data: ${textEvents[0]}\n\n`)
|
||||
response.write(`data: ${textEvents[1]}\n\n`)
|
||||
// never finish; rely on client abort
|
||||
request.socket.on('close', () => { socketClosed = true })
|
||||
})
|
||||
})
|
||||
servers.push(server)
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (address === null || typeof address === 'string') throw new Error('no port')
|
||||
const ctx = await harness(`http://127.0.0.1:${address.port}`)
|
||||
|
||||
for await (const chunk of ctx.llm.stream({ model: 'deepseek-v4-flash', messages: [] })) {
|
||||
if (chunk.type === 'text-delta') break // stop early mid-stream
|
||||
}
|
||||
// The finally-abort must reach the server as a closed socket.
|
||||
await vi.waitFor(() => { expect(socketClosed).toBe(true) }, { timeout: 5_000 })
|
||||
})
|
||||
})
|
||||
|
||||
describe('review fixes: abort wiring', () => {
|
||||
it('honors a pre-aborted caller signal', async () => {
|
||||
const ctx = await harness('http://127.0.0.1:1')
|
||||
const controller = new AbortController()
|
||||
controller.abort('already cancelled')
|
||||
// pi-ai surfaces the abort as an in-stream error event → aborted finish.
|
||||
const result = await ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [],
|
||||
signal: controller.signal,
|
||||
})
|
||||
expect(result.finish.kind).toBe('aborted')
|
||||
})
|
||||
|
||||
it('propagates a mid-stream caller abort to the upstream request', async () => {
|
||||
const server = await mockServer([{ events: textEvents }])
|
||||
const ctx = await harness(server.url)
|
||||
const controller = new AbortController()
|
||||
const pending = ctx.llm.generate({
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [],
|
||||
signal: controller.signal,
|
||||
})
|
||||
controller.abort()
|
||||
const result = await pending
|
||||
// Either the abort lands before any chunk (aborted) or after the tiny
|
||||
// mock stream finished (stop) — both are valid races; never a hang.
|
||||
expect(['aborted', 'stop']).toContain(result.finish.kind)
|
||||
})
|
||||
})
|
||||
322
packages/llm-pi-ai/tests/convert.spec.ts
Normal file
322
packages/llm-pi-ai/tests/convert.spec.ts
Normal file
@@ -0,0 +1,322 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import type { StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import type { AssistantMessage, AssistantMessageEvent, Usage } from '@earendil-works/pi-ai'
|
||||
import { mapStopReason, mapUsage, toPiContext, toStreamChunks } from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
|
||||
function usage(input = 0, output = 0, cacheRead = 0, cacheWrite = 0): Usage {
|
||||
return {
|
||||
input,
|
||||
output,
|
||||
cacheRead,
|
||||
cacheWrite,
|
||||
totalTokens: input + output + cacheRead + cacheWrite,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
}
|
||||
}
|
||||
|
||||
function assistant(overrides: Partial<AssistantMessage> = {}): AssistantMessage {
|
||||
return {
|
||||
role: 'assistant',
|
||||
content: [],
|
||||
api: 'openai-completions',
|
||||
provider: 'deepseek',
|
||||
model: 'deepseek-v4-flash',
|
||||
usage: usage(),
|
||||
stopReason: 'stop',
|
||||
timestamp: 0,
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
async function* feed(...events: AssistantMessageEvent[]): AsyncGenerator<AssistantMessageEvent> {
|
||||
for (const event of events) yield event
|
||||
}
|
||||
|
||||
async function collect(stream: AsyncIterable<StreamChunk>): Promise<StreamChunk[]> {
|
||||
const out: StreamChunk[] = []
|
||||
for await (const chunk of stream) out.push(chunk)
|
||||
return out
|
||||
}
|
||||
|
||||
describe('toPiContext', () => {
|
||||
it('maps system prompt, user text, and tools', () => {
|
||||
const context = toPiContext({
|
||||
model: 'deepseek-v4-flash',
|
||||
system: 'be helpful',
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
|
||||
tools: [{ name: 'f', description: 'F', parameters: { type: 'object', properties: {} } }],
|
||||
})
|
||||
expect(context.systemPrompt).toBe('be helpful')
|
||||
expect(context.messages).toEqual([{ role: 'user', content: 'hi', timestamp: 0 }])
|
||||
expect(context.tools).toEqual([
|
||||
{ name: 'f', description: 'F', parameters: { type: 'object', properties: {} } },
|
||||
])
|
||||
})
|
||||
|
||||
it('omits empty tools and absent system prompt', () => {
|
||||
const context = toPiContext({ model: 'm', messages: [], tools: [] })
|
||||
expect(context.systemPrompt).toBeUndefined()
|
||||
expect(context.tools).toBeUndefined()
|
||||
})
|
||||
|
||||
it('maps assistant text/reasoning/tool-call blocks', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [{
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'reasoning', text: 'hmm' },
|
||||
{ type: 'text', text: 'calling' },
|
||||
{ type: 'tool-call', id: CallId('c1'), name: 'f', arguments: '{"a":1}' },
|
||||
],
|
||||
}],
|
||||
})
|
||||
const message = context.messages[0] as AssistantMessage
|
||||
expect(message.role).toBe('assistant')
|
||||
expect(message.stopReason).toBe('toolUse')
|
||||
expect(message.content).toEqual([
|
||||
// thinkingSignature names the replay field — DeepSeek's passback rule.
|
||||
{ type: 'thinking', thinking: 'hmm', thinkingSignature: 'reasoning_content' },
|
||||
{ type: 'text', text: 'calling' },
|
||||
{ type: 'toolCall', id: 'c1', name: 'f', arguments: { a: 1 } },
|
||||
])
|
||||
})
|
||||
|
||||
it('marks tool-call-free assistant messages with stopReason stop', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [{ role: 'assistant', content: [{ type: 'text', text: 'done' }] }],
|
||||
})
|
||||
expect((context.messages[0] as AssistantMessage).stopReason).toBe('stop')
|
||||
})
|
||||
|
||||
it('parses malformed tool-call arguments to {}', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [{
|
||||
role: 'assistant',
|
||||
content: [{ type: 'tool-call', id: CallId('c1'), name: 'f', arguments: '{broken' }],
|
||||
}],
|
||||
})
|
||||
const message = context.messages[0] as AssistantMessage
|
||||
expect(message.content[0]).toEqual({ type: 'toolCall', id: 'c1', name: 'f', arguments: {} })
|
||||
})
|
||||
|
||||
it('parses non-object argument JSON (arrays, scalars) to {}', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [{
|
||||
role: 'assistant',
|
||||
content: [{ type: 'tool-call', id: CallId('c1'), name: 'f', arguments: '[1,2]' }],
|
||||
}],
|
||||
})
|
||||
expect((context.messages[0] as AssistantMessage).content[0]).toMatchObject({ arguments: {} })
|
||||
})
|
||||
|
||||
it('recovers toolName for tool results from the preceding assistant call', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [
|
||||
{
|
||||
role: 'assistant',
|
||||
content: [{ type: 'tool-call', id: CallId('c1'), name: 'get_weather', arguments: '{}' }],
|
||||
},
|
||||
{
|
||||
role: 'user',
|
||||
content: [{ type: 'tool-result', toolCallId: CallId('c1'), content: [{ type: 'text', text: 'Sunny' }] }],
|
||||
},
|
||||
],
|
||||
})
|
||||
expect(context.messages[1]).toEqual({
|
||||
role: 'toolResult',
|
||||
toolCallId: 'c1',
|
||||
toolName: 'get_weather',
|
||||
content: [{ type: 'text', text: 'Sunny' }],
|
||||
isError: false,
|
||||
timestamp: 0,
|
||||
})
|
||||
})
|
||||
|
||||
it('labels unmatched tool results with toolName unknown and keeps isError', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [{
|
||||
role: 'user',
|
||||
content: [{ type: 'tool-result', toolCallId: CallId('zz'), content: [], isError: true }],
|
||||
}],
|
||||
})
|
||||
expect(context.messages[0]).toMatchObject({
|
||||
role: 'toolResult',
|
||||
toolName: 'unknown',
|
||||
isError: true,
|
||||
content: [{ type: 'text', text: '(no output)' }],
|
||||
})
|
||||
})
|
||||
|
||||
it('splits mixed user text + tool results and folds history system messages', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [
|
||||
{ role: 'system', content: [{ type: 'text', text: 'rule' }] },
|
||||
{
|
||||
role: 'user',
|
||||
content: [
|
||||
{ type: 'text', text: 'note' },
|
||||
{ type: 'tool-result', toolCallId: CallId('c1'), content: [{ type: 'text', text: 'ok' }] },
|
||||
],
|
||||
},
|
||||
],
|
||||
})
|
||||
expect(context.messages.map(message => message.role)).toEqual(['user', 'user', 'toolResult'])
|
||||
})
|
||||
|
||||
it('skips image and unknown blocks in assistant content', () => {
|
||||
const context = toPiContext({
|
||||
model: 'm',
|
||||
messages: [{
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'image', url: 'data:,x' },
|
||||
{ type: 'text', text: 'visible' },
|
||||
],
|
||||
}],
|
||||
})
|
||||
expect((context.messages[0] as AssistantMessage).content).toEqual([{ type: 'text', text: 'visible' }])
|
||||
})
|
||||
})
|
||||
|
||||
describe('toStreamChunks', () => {
|
||||
const partialWithToolCall = assistant({
|
||||
content: [{ type: 'toolCall', id: 'call-1', name: 'f', arguments: {} }],
|
||||
})
|
||||
|
||||
it('maps text events to text blocks', async () => {
|
||||
const done = assistant({ content: [{ type: 'text', text: 'hi' }], usage: usage(3, 2) })
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'start', partial: assistant() },
|
||||
{ type: 'text_start', contentIndex: 0, partial: assistant() },
|
||||
{ type: 'text_delta', contentIndex: 0, delta: 'hi', partial: assistant() },
|
||||
{ type: 'text_end', contentIndex: 0, content: 'hi', partial: assistant() },
|
||||
{ type: 'done', reason: 'stop', message: done },
|
||||
)))
|
||||
expect(chunks).toEqual([
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
{ type: 'text-delta', index: 0, text: 'hi' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text: 'hi' } },
|
||||
{ type: 'usage', usage: { inputTokens: 3, outputTokens: 2 } },
|
||||
{ type: 'finish', reason: { kind: 'stop' } },
|
||||
])
|
||||
})
|
||||
|
||||
it('maps thinking events to reasoning blocks', async () => {
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'thinking_start', contentIndex: 0, partial: assistant() },
|
||||
{ type: 'thinking_delta', contentIndex: 0, delta: 'mull', partial: assistant() },
|
||||
{ type: 'thinking_end', contentIndex: 0, content: 'mull', partial: assistant() },
|
||||
{ type: 'done', reason: 'stop', message: assistant() },
|
||||
)))
|
||||
expect(chunks.slice(0, 3)).toEqual([
|
||||
{ type: 'block-start', index: 0, blockType: 'reasoning' },
|
||||
{ type: 'reasoning-delta', index: 0, text: 'mull' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'reasoning', text: 'mull' } },
|
||||
])
|
||||
})
|
||||
|
||||
it('maps tool-call events, re-stringifying parsed arguments', async () => {
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'toolcall_start', contentIndex: 0, partial: partialWithToolCall },
|
||||
{ type: 'toolcall_delta', contentIndex: 0, delta: '{"a"', partial: partialWithToolCall },
|
||||
{ type: 'toolcall_delta', contentIndex: 0, delta: ':1}', partial: partialWithToolCall },
|
||||
{
|
||||
type: 'toolcall_end',
|
||||
contentIndex: 0,
|
||||
toolCall: { type: 'toolCall', id: 'call-1', name: 'f', arguments: { a: 1 } },
|
||||
partial: partialWithToolCall,
|
||||
},
|
||||
{ type: 'done', reason: 'toolUse', message: assistant({ stopReason: 'toolUse' }) },
|
||||
)))
|
||||
expect(chunks).toEqual([
|
||||
{ type: 'block-start', index: 0, blockType: 'tool-call' },
|
||||
{ type: 'tool-call-delta', index: 0, id: 'call-1', name: 'f', argumentsDelta: '{"a"' },
|
||||
{ type: 'tool-call-delta', index: 0, id: 'call-1', name: 'f', argumentsDelta: ':1}' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: 'call-1', name: 'f', arguments: '{"a":1}' } },
|
||||
{ type: 'usage', usage: { inputTokens: 0, outputTokens: 0 } },
|
||||
{ type: 'finish', reason: { kind: 'tool-calls' } },
|
||||
])
|
||||
})
|
||||
|
||||
it('tolerates toolcall_start with a missing partial entry', async () => {
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'toolcall_start', contentIndex: 0, partial: assistant() },
|
||||
{ type: 'toolcall_delta', contentIndex: 0, delta: '{}', partial: assistant() },
|
||||
{ type: 'done', reason: 'stop', message: assistant() },
|
||||
)))
|
||||
expect(chunks[1]).toEqual({ type: 'tool-call-delta', index: 0, id: '', argumentsDelta: '{}' })
|
||||
})
|
||||
|
||||
it('maps error events to error finish chunks (in-stream error style)', async () => {
|
||||
const error = assistant({ stopReason: 'error', errorMessage: 'boom', usage: usage(1, 0) })
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'error', reason: 'error', error },
|
||||
)))
|
||||
expect(chunks).toEqual([
|
||||
{ type: 'usage', usage: { inputTokens: 1, outputTokens: 0 } },
|
||||
{ type: 'finish', reason: { kind: 'error', message: 'boom', code: 'PI_AI_ERROR' } },
|
||||
])
|
||||
})
|
||||
|
||||
it('maps aborted error events to aborted finish', async () => {
|
||||
const error = assistant({ stopReason: 'aborted' })
|
||||
const chunks = await collect(toStreamChunks(feed({ type: 'error', reason: 'aborted', error })))
|
||||
expect(chunks.at(-1)).toEqual({ type: 'finish', reason: { kind: 'aborted' } })
|
||||
})
|
||||
})
|
||||
|
||||
describe('mapStopReason / mapUsage', () => {
|
||||
it.each([
|
||||
['stop', { kind: 'stop' }],
|
||||
['length', { kind: 'max-tokens' }],
|
||||
['toolUse', { kind: 'tool-calls' }],
|
||||
['aborted', { kind: 'aborted' }],
|
||||
] as const)('maps %s', (stopReason, expected) => {
|
||||
expect(mapStopReason(assistant({ stopReason }))).toEqual(expected)
|
||||
})
|
||||
|
||||
it('defaults the error message when pi-ai omits it', () => {
|
||||
expect(mapStopReason(assistant({ stopReason: 'error' })))
|
||||
.toEqual({ kind: 'error', message: 'pi-ai stream error', code: 'PI_AI_ERROR' })
|
||||
})
|
||||
|
||||
it('maps cache fields only when nonzero', () => {
|
||||
expect(mapUsage(usage(10, 5, 8, 2))).toEqual({
|
||||
inputTokens: 10,
|
||||
outputTokens: 5,
|
||||
cacheReadTokens: 8,
|
||||
cacheWriteTokens: 2,
|
||||
})
|
||||
expect(mapUsage(usage(10, 5))).toEqual({ inputTokens: 10, outputTokens: 5 })
|
||||
})
|
||||
})
|
||||
|
||||
describe('toStreamChunks edge branches', () => {
|
||||
it('omits the name field for tool calls whose partial carried an empty name', async () => {
|
||||
const blank = assistant({ content: [{ type: 'toolCall', id: 'x', name: '', arguments: {} }] })
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'toolcall_start', contentIndex: 0, partial: blank },
|
||||
{ type: 'toolcall_delta', contentIndex: 0, delta: '{}', partial: blank },
|
||||
{ type: 'done', reason: 'stop', message: assistant() },
|
||||
)))
|
||||
expect(chunks[1]).toEqual({ type: 'tool-call-delta', index: 0, id: 'x', argumentsDelta: '{}' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('toStreamChunks defensive branches', () => {
|
||||
it('tolerates a toolcall_delta with no preceding toolcall_start', async () => {
|
||||
const chunks = await collect(toStreamChunks(feed(
|
||||
{ type: 'toolcall_delta', contentIndex: 0, delta: '{}', partial: assistant() },
|
||||
{ type: 'done', reason: 'stop', message: assistant() },
|
||||
)))
|
||||
expect(chunks[0]).toEqual({ type: 'tool-call-delta', index: 0, id: '', argumentsDelta: '{}' })
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user