Merge remote-tracking branch 'origin/master' into worktree/pr628-merge-20260727
# Conflicts: # .agents/notes/implemented/architecture/2026-06-21-bounded-llm-request-recovery.md # docs/architecture.i18n.yaml # docs/config-catalog.md # docs/cordis-catalog/events.md # docs/cordis-catalog/services.md # docs/core-data-structures/llm-streaming.i18n.yaml # docs/core-data-structures/llm-streaming.md # docs/core-data-structures/llm-streaming.zh.md # docs/event-producer-consumer.md # docs/module-graph.md # examples/headless-agent/tests/headless.snapshot.ts # packages/compact/compact-basic/tests/compact-loop-repro.spec.ts # packages/cordis/tool-cordis/src/api-catalog.ts # packages/core/agent-loop/README.md # packages/core/agent-loop/src/loop.ts # packages/examples/agent-spine-demo/README.md # packages/llm/README.md # packages/llm/llm-deepseek/src/adapter.ts # packages/llm/llm-pi-ai/src/adapter.ts # packages/llm/llm-retry/README.md # packages/llm/llm/README.md # packages/llm/llm/src/index.ts # packages/llm/llm/tests/service.spec.ts # packages/support/llm-replay/src/index.ts # packages/support/llm-replay/tests/llm-replay.spec.ts
This commit is contained in:
Tianyi Cui
@@ -1,13 +1,13 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { CallId, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import type { Config } from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import { assemble, type AssembledResult } from './assemble.ts'
|
||||
|
||||
/**
|
||||
* Real-API e2e for the hand-rolled adapter: V4 Flash + V4 Pro across
|
||||
* Real-API e2e for the direct-fetch adapter: V4 Flash + V4 Pro across
|
||||
* thinking modes and both official effort levels. Key-gated — skips
|
||||
* entirely without $DEEPSEEK_API_KEY (see vitest.e2e.config.ts).
|
||||
*/
|
||||
@@ -50,41 +50,40 @@ const weatherTool: ToolSchema = {
|
||||
}
|
||||
|
||||
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', () => {
|
||||
it('flash + thinking disabled: plain text generation', async () => {
|
||||
const ctx = await harness(FLASH, { thinking: 'disabled' })
|
||||
const result = await assemble(ctx,{
|
||||
it('flash dynamically switches from off to high', async () => {
|
||||
const ctx = await harness(FLASH, { reasoningEffort: 'off' })
|
||||
const withoutThinking = await assemble(ctx,{
|
||||
model: FLASH,
|
||||
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)
|
||||
expect(result.usage?.inputTokens).toBeGreaterThan(0)
|
||||
expect(result.usage?.outputTokens).toBeGreaterThan(0)
|
||||
})
|
||||
expect(withoutThinking.finish.kind).toBe('stop')
|
||||
expect(textOf(withoutThinking).toLowerCase()).toContain('pong')
|
||||
expect(withoutThinking.message.content.some(block => block.type === 'reasoning')).toBe(false)
|
||||
expect(withoutThinking.usage?.inputTokens).toBeGreaterThan(0)
|
||||
expect(withoutThinking.usage?.outputTokens).toBeGreaterThan(0)
|
||||
|
||||
it('flash + thinking enabled (effort high): reasoning blocks + reasoning tokens', async () => {
|
||||
const ctx = await harness(FLASH, { thinking: 'enabled', reasoningEffort: 'high' })
|
||||
const result = await assemble(ctx,{
|
||||
const withThinking = await assemble(ctx,{
|
||||
model: FLASH,
|
||||
reasoningEffort: ReasoningEffortId('high'),
|
||||
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')
|
||||
expect(result.usage?.reasoningTokens).toBeGreaterThan(0)
|
||||
expect(withThinking.finish.kind).toBe('stop')
|
||||
expect(withThinking.message.content.some(block => block.type === 'reasoning')).toBe(true)
|
||||
expect(textOf(withThinking)).toContain('9.8')
|
||||
expect(withThinking.usage?.reasoningTokens).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it.each(['high', 'max'] as const)(
|
||||
'pro + thinking enabled (effort %s): tool-call round trip with reasoning passback',
|
||||
async (effort) => {
|
||||
const ctx = await harness(PRO, { thinking: 'enabled', reasoningEffort: effort })
|
||||
const ctx = await harness(PRO, { thinking: 'enabled' })
|
||||
|
||||
// Turn 1: the model must call the tool (and think before it).
|
||||
const first = await assemble(ctx,{
|
||||
model: PRO,
|
||||
reasoningEffort: ReasoningEffortId(effort),
|
||||
messages: ask('What is the weather in Paris right now? Use the get_weather tool.'),
|
||||
tools: [weatherTool],
|
||||
maxTokens: 2000,
|
||||
@@ -99,6 +98,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', ()
|
||||
// block in history (the official thinking+tools passback rule).
|
||||
const second = await assemble(ctx,{
|
||||
model: PRO,
|
||||
reasoningEffort: ReasoningEffortId(effort),
|
||||
messages: [
|
||||
...ask('What is the weather in Paris right now? Use the get_weather tool.'),
|
||||
{ role: 'assistant', content: first.message.content },
|
||||
|
||||
@@ -8,6 +8,7 @@ import LlmService, {
|
||||
LlmError,
|
||||
ProviderRequestId,
|
||||
QUOTA_EXCEEDED_CODE,
|
||||
ReasoningEffortId,
|
||||
userAgent,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
|
||||
@@ -120,6 +121,7 @@ describe('DeepSeekAdapter against a mock server', () => {
|
||||
// The wire request carried the auth header contents we configured.
|
||||
expect(server.requests[0]).toMatchObject({
|
||||
model: 'deepseek-v4-flash',
|
||||
reasoning_effort: 'high',
|
||||
stream: true,
|
||||
stream_options: { include_usage: true },
|
||||
})
|
||||
@@ -173,9 +175,45 @@ describe('DeepSeekAdapter against a mock server', () => {
|
||||
expect(server.headers[0]?.['x-deepseek-harness-compact']).toBe('1')
|
||||
})
|
||||
|
||||
it('forwards thinking config onto the wire', async () => {
|
||||
it('switches dynamically from the configured high default through off to max', async () => {
|
||||
const server = await mockServer([
|
||||
{ kind: 'sse', events: textEvents },
|
||||
{ kind: 'sse', events: textEvents },
|
||||
{ kind: 'sse', events: textEvents },
|
||||
])
|
||||
const ctx = await harness(server.url, { thinking: 'enabled', reasoningEffort: 'high' })
|
||||
|
||||
await assemble(ctx,{
|
||||
model: 'deepseek-v4-flash',
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
|
||||
})
|
||||
await assemble(ctx,{
|
||||
model: 'deepseek-v4-flash',
|
||||
reasoningEffort: ReasoningEffortId('off'),
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi again' }] }],
|
||||
})
|
||||
await assemble(ctx,{
|
||||
model: 'deepseek-v4-flash',
|
||||
reasoningEffort: ReasoningEffortId('max'),
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'one more time' }] }],
|
||||
})
|
||||
expect(server.requests[0]).toMatchObject({
|
||||
thinking: { type: 'enabled' },
|
||||
reasoning_effort: 'high',
|
||||
})
|
||||
expect(server.requests[1]).toMatchObject({
|
||||
thinking: { type: 'disabled' },
|
||||
})
|
||||
expect(server.requests[1]).not.toHaveProperty('reasoning_effort')
|
||||
expect(server.requests[2]).toMatchObject({
|
||||
thinking: { type: 'enabled' },
|
||||
reasoning_effort: 'max',
|
||||
})
|
||||
})
|
||||
|
||||
it('publishes only off and omits the wire effort when thinking is disabled', async () => {
|
||||
const server = await mockServer([{ kind: 'sse', events: textEvents }])
|
||||
const ctx = await harness(server.url, { thinking: 'disabled', reasoningEffort: 'high' })
|
||||
const ctx = await harness(server.url, { thinking: 'disabled' })
|
||||
|
||||
await assemble(ctx,{
|
||||
model: 'deepseek-v4-flash',
|
||||
@@ -183,10 +221,52 @@ describe('DeepSeekAdapter against a mock server', () => {
|
||||
})
|
||||
expect(server.requests[0]).toMatchObject({
|
||||
thinking: { type: 'disabled' },
|
||||
reasoning_effort: 'high',
|
||||
})
|
||||
expect(server.requests[0]).not.toHaveProperty('reasoning_effort')
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'deepseek-v4-flash'))
|
||||
.resolves.toMatchObject({
|
||||
reasoning: {
|
||||
efforts: [{ id: ReasoningEffortId('off'), name: 'Off' }],
|
||||
defaultEffort: ReasoningEffortId('off'),
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects a per-request effort before I/O when thinking is disabled', async () => {
|
||||
const server = await mockServer([])
|
||||
const ctx = await harness(server.url, { thinking: 'disabled' })
|
||||
|
||||
await expect(assemble(ctx, {
|
||||
model: 'deepseek-v4-flash',
|
||||
reasoningEffort: ReasoningEffortId('high'),
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
|
||||
})).rejects.toMatchObject({ code: 'UNSUPPORTED_REASONING_EFFORT' })
|
||||
expect(server.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it.each(['high', 'max'])(
|
||||
'rejects direct adapter effort %s before I/O when thinking is disabled',
|
||||
async (effort) => {
|
||||
const server = await mockServer([])
|
||||
const adapter = new DeepSeekAdapter({
|
||||
apiKey: 'test-key',
|
||||
baseURL: server.url,
|
||||
defaults: { thinking: 'disabled' },
|
||||
})
|
||||
|
||||
const stream = adapter.stream({
|
||||
provider: 'deepseek',
|
||||
model: 'deepseek-v4-flash',
|
||||
reasoningEffort: ReasoningEffortId(effort),
|
||||
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
|
||||
})
|
||||
await expect(async () => {
|
||||
for await (const _chunk of stream) { /* drain */ }
|
||||
}).rejects.toMatchObject({ code: 'UNSUPPORTED_REASONING_EFFORT' })
|
||||
expect(server.requests).toHaveLength(0)
|
||||
},
|
||||
)
|
||||
|
||||
it.each([
|
||||
[401, 'AUTH'],
|
||||
[403, 'AUTH'],
|
||||
@@ -551,8 +631,100 @@ describe('plugin registration and config', () => {
|
||||
{ provider: 'deepseek', id: 'deepseek-v4-flash', name: 'deepseek-v4-flash' },
|
||||
{ provider: 'deepseek', id: 'deepseek-v4-pro', name: 'deepseek-v4-pro' },
|
||||
])
|
||||
await expect(ctx.llm.resolveModelContext('deepseek', 'deepseek-v4-flash'))
|
||||
.resolves.toEqual({ contextWindow: 128_000 })
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'deepseek-v4-flash'))
|
||||
.resolves.toMatchObject({
|
||||
provider: 'deepseek',
|
||||
id: 'deepseek-v4-flash',
|
||||
name: 'deepseek-v4-flash',
|
||||
context: { contextWindow: 128_000 },
|
||||
reasoning: {
|
||||
efforts: [
|
||||
{ id: ReasoningEffortId('off'), name: 'Off' },
|
||||
{ id: ReasoningEffortId('high'), name: 'High' },
|
||||
{ id: ReasoningEffortId('max'), name: 'Max' },
|
||||
],
|
||||
defaultEffort: ReasoningEffortId('high'),
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it.each(['off', 'max'] as const)('uses the configured %s reasoning default', async (effort) => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(LlmDeepSeek, {
|
||||
apiKey: 'k',
|
||||
baseURL: 'http://127.0.0.1:1',
|
||||
reasoningEffort: effort,
|
||||
})
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'unlisted-pass-through'))
|
||||
.resolves.toMatchObject({
|
||||
reasoning: {
|
||||
efforts: [
|
||||
{ id: ReasoningEffortId('off'), name: 'Off' },
|
||||
{ id: ReasoningEffortId('high'), name: 'High' },
|
||||
{ id: ReasoningEffortId('max'), name: 'Max' },
|
||||
],
|
||||
defaultEffort: ReasoningEffortId(effort),
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it('accepts off as the default when thinking is deployment-disabled', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(LlmDeepSeek, {
|
||||
apiKey: 'k',
|
||||
baseURL: 'http://127.0.0.1:1',
|
||||
thinking: 'disabled',
|
||||
reasoningEffort: 'off',
|
||||
})
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'unlisted-pass-through'))
|
||||
.resolves.toMatchObject({
|
||||
reasoning: {
|
||||
efforts: [{ id: ReasoningEffortId('off'), name: 'Off' }],
|
||||
defaultEffort: ReasoningEffortId('off'),
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it.each(['high', 'max'] as const)(
|
||||
'rejects configured reasoning effort %s when thinking is disabled',
|
||||
async (reasoningEffort) => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await expect(ctx.plugin(LlmDeepSeek, {
|
||||
apiKey: 'k',
|
||||
baseURL: 'http://127.0.0.1:1',
|
||||
thinking: 'disabled',
|
||||
reasoningEffort,
|
||||
})).rejects.toThrow(/only reasoningEffort "off"/)
|
||||
expect(ctx.llm.listProviders()).toEqual([])
|
||||
},
|
||||
)
|
||||
|
||||
it.each(['high', 'max'] as const)(
|
||||
'rejects disabled-thinking effort %s at the direct constructor boundary',
|
||||
(reasoningEffort) => {
|
||||
expect(() => new DeepSeekAdapter({
|
||||
apiKey: 'k',
|
||||
baseURL: 'http://127.0.0.1:1',
|
||||
defaults: { thinking: 'disabled', reasoningEffort },
|
||||
})).toThrow(/only reasoningEffort "off"/)
|
||||
},
|
||||
)
|
||||
|
||||
it('accepts disabled thinking with off at the direct constructor boundary', async () => {
|
||||
const adapter = new DeepSeekAdapter({
|
||||
apiKey: 'k',
|
||||
baseURL: 'http://127.0.0.1:1',
|
||||
defaults: { thinking: 'disabled', reasoningEffort: 'off' },
|
||||
})
|
||||
await expect(adapter.resolveModel('deepseek', 'pass-through')).resolves.toMatchObject({
|
||||
reasoning: {
|
||||
efforts: [{ id: ReasoningEffortId('off'), name: 'Off' }],
|
||||
defaultEffort: ReasoningEffortId('off'),
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it('uses the default model catalog when apply is called directly', async () => {
|
||||
@@ -585,10 +757,15 @@ describe('plugin registration and config', () => {
|
||||
{ provider: 'deepseek', id: 'private-fast', name: 'private-fast' },
|
||||
{ provider: 'deepseek', id: 'private-reasoner', name: 'Private Reasoner', description: 'Higher reasoning budget' },
|
||||
])
|
||||
await expect(ctx.llm.resolveModelContext('deepseek', 'private-fast'))
|
||||
.resolves.toEqual({ contextWindow: 32_000 })
|
||||
await expect(ctx.llm.resolveModelContext('deepseek', 'arbitrary-unlisted'))
|
||||
.resolves.toBeUndefined()
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'private-fast'))
|
||||
.resolves.toMatchObject({ context: { contextWindow: 32_000 } })
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'private-reasoner'))
|
||||
.resolves.toMatchObject({
|
||||
name: 'Private Reasoner',
|
||||
description: 'Higher reasoning budget',
|
||||
})
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'arbitrary-unlisted'))
|
||||
.resolves.not.toHaveProperty('context')
|
||||
})
|
||||
|
||||
it('uses exact model capacity before the adapter-wide default', async () => {
|
||||
@@ -604,12 +781,12 @@ describe('plugin registration and config', () => {
|
||||
],
|
||||
})
|
||||
|
||||
await expect(ctx.llm.resolveModelContext('deepseek', 'inherits-default'))
|
||||
.resolves.toEqual({ contextWindow: 256_000 })
|
||||
await expect(ctx.llm.resolveModelContext('deepseek', 'exact-override'))
|
||||
.resolves.toEqual({ contextWindow: 64_000 })
|
||||
await expect(ctx.llm.resolveModelContext('deepseek', 'unlisted-pass-through'))
|
||||
.resolves.toEqual({ contextWindow: 256_000 })
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'inherits-default'))
|
||||
.resolves.toMatchObject({ context: { contextWindow: 256_000 } })
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'exact-override'))
|
||||
.resolves.toMatchObject({ context: { contextWindow: 64_000 } })
|
||||
await expect(ctx.llm.resolveModelInfo('deepseek', 'unlisted-pass-through'))
|
||||
.resolves.toMatchObject({ context: { contextWindow: 256_000 } })
|
||||
})
|
||||
|
||||
it('allows an explicit empty model catalog', async () => {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { serializeMessages, serializeRequest } from '../src/serialize.ts'
|
||||
|
||||
@@ -174,15 +174,47 @@ describe('serializeRequest', () => {
|
||||
expect(wire.tools).toBeUndefined()
|
||||
})
|
||||
|
||||
it('applies adapter defaults for thinking and effort', () => {
|
||||
const wire = serializeRequest(request({ messages: history }), { thinking: 'enabled', reasoningEffort: 'max' })
|
||||
it('maps adapter-default thinking and the request reasoning effort', () => {
|
||||
const wire = serializeRequest(
|
||||
request({ messages: history, reasoningEffort: ReasoningEffortId('max') }),
|
||||
{ thinking: 'enabled', reasoningEffort: 'high' },
|
||||
)
|
||||
expect(wire.thinking).toEqual({ type: 'enabled' })
|
||||
expect(wire.reasoning_effort).toBe('max')
|
||||
})
|
||||
|
||||
it('maps off to disabled thinking without a wire reasoning effort', () => {
|
||||
const wire = serializeRequest(
|
||||
request({ messages: history, reasoningEffort: ReasoningEffortId('off') }),
|
||||
{ thinking: 'enabled', reasoningEffort: 'max' },
|
||||
)
|
||||
expect(wire.thinking).toEqual({ type: 'disabled' })
|
||||
expect(wire.reasoning_effort).toBeUndefined()
|
||||
})
|
||||
|
||||
it('re-enables thinking when max overrides an off default', () => {
|
||||
const wire = serializeRequest(
|
||||
request({ messages: history, reasoningEffort: ReasoningEffortId('max') }),
|
||||
{ reasoningEffort: 'off' },
|
||||
)
|
||||
expect(wire.thinking).toEqual({ type: 'enabled' })
|
||||
expect(wire.reasoning_effort).toBe('max')
|
||||
})
|
||||
|
||||
it('rejects enabling thinking when the deployment is locked to disabled', () => {
|
||||
expect(() => serializeRequest(
|
||||
request({ messages: history, reasoningEffort: ReasoningEffortId('high') }),
|
||||
{ thinking: 'disabled' },
|
||||
)).toThrow(expect.objectContaining({ code: 'UNSUPPORTED_REASONING_EFFORT' }))
|
||||
})
|
||||
|
||||
it('disables thinking for session-title requests without changing adapter defaults', () => {
|
||||
const wire = serializeRequest(
|
||||
request({ messages: history, purpose: 'session-title' }),
|
||||
request({
|
||||
messages: history,
|
||||
purpose: 'session-title',
|
||||
reasoningEffort: ReasoningEffortId('max'),
|
||||
}),
|
||||
{ thinking: 'enabled', reasoningEffort: 'max' },
|
||||
)
|
||||
expect(wire.thinking).toEqual({ type: 'disabled' })
|
||||
@@ -194,6 +226,19 @@ describe('serializeRequest', () => {
|
||||
expect(wire.thinking).toBeUndefined()
|
||||
expect(wire.reasoning_effort).toBeUndefined()
|
||||
})
|
||||
|
||||
it('preserves an explicit enabled default without inventing a wire effort', () => {
|
||||
const wire = serializeRequest(request({ messages: history }), { thinking: 'enabled' })
|
||||
expect(wire.thinking).toEqual({ type: 'enabled' })
|
||||
expect(wire.reasoning_effort).toBeUndefined()
|
||||
})
|
||||
|
||||
it('rejects an effort outside the DeepSeek capability', () => {
|
||||
expect(() => serializeRequest(request({
|
||||
messages: history,
|
||||
reasoningEffort: ReasoningEffortId('medium'),
|
||||
}))).toThrow(expect.objectContaining({ code: 'UNSUPPORTED_REASONING_EFFORT' }))
|
||||
})
|
||||
})
|
||||
|
||||
describe('review fixes: assistant content shapes', () => {
|
||||
|
||||
@@ -2,12 +2,21 @@ import { describe, expect, it } from 'vitest'
|
||||
import { LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import { DONE, parseSse } from '../src/sse.ts'
|
||||
|
||||
/** Build a byte stream from string fragments (fragments = network reads). */
|
||||
async function* bytes(...fragments: (string | Uint8Array)[]): AsyncGenerator<Uint8Array> {
|
||||
/**
|
||||
* DeepSeek protocol contract only: the [DONE] sentinel and STREAM_CLOSED on
|
||||
* EOF without it. SSE framing (chunk splits, CRLF, multi-data joins, comments)
|
||||
* is eventsource-parser's contract, not re-proven here.
|
||||
*/
|
||||
|
||||
/** Build an SSE byte stream from string fragments (fragments = network reads). */
|
||||
function bytes(...fragments: string[]): ReadableStream<Uint8Array<ArrayBuffer>> {
|
||||
const encoder = new TextEncoder()
|
||||
for (const fragment of fragments) {
|
||||
yield typeof fragment === 'string' ? encoder.encode(fragment) : fragment
|
||||
}
|
||||
return new ReadableStream({
|
||||
start(controller) {
|
||||
for (const fragment of fragments) controller.enqueue(encoder.encode(fragment))
|
||||
controller.close()
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
async function collect(stream: AsyncIterable<string>): Promise<string[]> {
|
||||
@@ -17,57 +26,14 @@ async function collect(stream: AsyncIterable<string>): Promise<string[]> {
|
||||
}
|
||||
|
||||
describe('parseSse', () => {
|
||||
it('parses simple events and the DONE sentinel', async () => {
|
||||
it('yields event payloads and the DONE sentinel', async () => {
|
||||
const events = await collect(parseSse(bytes('data: {"a":1}\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('handles events split across reads at arbitrary positions', async () => {
|
||||
const events = await collect(parseSse(bytes('da', 'ta: {"a"', ':1}\n', '\ndata: [DO', 'NE]\n\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('handles multi-byte UTF-8 split across reads', async () => {
|
||||
const encoded = new TextEncoder().encode('data: {"text":"日本語"}\n\ndata: [DONE]\n\n')
|
||||
// Split inside the 3-byte sequence for 日.
|
||||
const splitAt = 16
|
||||
const events = await collect(parseSse(bytes(encoded.slice(0, splitAt), encoded.slice(splitAt))))
|
||||
expect(events).toEqual(['{"text":"日本語"}', DONE])
|
||||
})
|
||||
|
||||
it('tolerates CRLF line endings', async () => {
|
||||
const events = await collect(parseSse(bytes('data: {"a":1}\r\n\r\ndata: [DONE]\r\n\r\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('joins multi-data events with newlines (SSE spec)', async () => {
|
||||
const events = await collect(parseSse(bytes('data: line1\ndata: line2\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['line1\nline2', DONE])
|
||||
})
|
||||
|
||||
it('ignores comments and non-data fields', async () => {
|
||||
const events = await collect(parseSse(bytes(': keepalive\nevent: chunk\nid: 7\ndata: {"a":1}\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('skips blocks without data fields', async () => {
|
||||
const events = await collect(parseSse(bytes(': ping\n\ndata: {"a":1}\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('preserves data lines without the optional space', async () => {
|
||||
const events = await collect(parseSse(bytes('data:{"a":1}\n\ndata:[DONE]\n\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('parses several events from one read', async () => {
|
||||
const events = await collect(parseSse(bytes('data: 1\n\ndata: 2\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['1', '2', DONE])
|
||||
})
|
||||
|
||||
it('flushes a final un-terminated DONE at stream end', async () => {
|
||||
const events = await collect(parseSse(bytes('data: {"a":1}\n\ndata: [DONE]')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
it('stops yielding after DONE even when more data follows', async () => {
|
||||
const events = await collect(parseSse(bytes('data: [DONE]\n\ndata: {"late":1}\n\n')))
|
||||
expect(events).toEqual([DONE])
|
||||
})
|
||||
|
||||
it('throws STREAM_CLOSED when the stream ends without DONE', async () => {
|
||||
@@ -83,26 +49,10 @@ describe('parseSse', () => {
|
||||
await expect(collect(parseSse(bytes('data: {"a"')))).rejects.toThrow(/without \[DONE\]/)
|
||||
})
|
||||
|
||||
it('stops yielding after DONE even when more data follows', async () => {
|
||||
const events = await collect(parseSse(bytes('data: [DONE]\n\ndata: {"late":1}\n\n')))
|
||||
expect(events).toEqual([DONE])
|
||||
})
|
||||
})
|
||||
|
||||
describe('parseSse edge branches', () => {
|
||||
it('handles a lone CR-terminated data line', async () => {
|
||||
// Exercises the \r-strip branch on a line that is ONLY "data:…\r".
|
||||
const events = await collect(parseSse(bytes('data: {"a":1}\r\n\r\ndata:[DONE]\r\n\r\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('strips CR from non-data field lines too', async () => {
|
||||
const events = await collect(parseSse(bytes('event: chunk\r\ndata: {"a":1}\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['{"a":1}', DONE])
|
||||
})
|
||||
|
||||
it('treats bare "data:" lines as empty payload entries', async () => {
|
||||
const events = await collect(parseSse(bytes('data:\ndata: x\n\ndata: [DONE]\n\n')))
|
||||
expect(events).toEqual(['\nx', DONE])
|
||||
it('treats a final DONE missing its blank-line terminator as truncation', async () => {
|
||||
// Spec-strict framing: an event dispatches only on its blank-line
|
||||
// terminator, so an unterminated tail at EOF is STREAM_CLOSED — real
|
||||
// providers always terminate events, so a missing terminator is truncation.
|
||||
await expect(collect(parseSse(bytes('data: {"a":1}\n\ndata: [DONE]')))).rejects.toThrow(/without \[DONE\]/)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user