761 lines
29 KiB
TypeScript
761 lines
29 KiB
TypeScript
/**
|
||
* SSE 流式解析器单元测试（v0.4.1 测试补齐）
|
||
* 覆盖：TEXT_DELTA / REASONING_DELTA / TOOL_CALL 增量拼接 / USAGE /
|
||
* [DONE] / finish_reason=tool_calls 提前 flush / 坏 JSON 行容错 / 损坏工具调用跳过
|
||
*/
|
||
|
||
import { describe, it, expect, vi, afterEach } from 'vitest';
|
||
import {
|
||
parseSSEStream,
|
||
parseOpenAICompatibleResponse,
|
||
readStreamChunkWithIdleTimeout,
|
||
SseUpstreamError,
|
||
} from '../shared/sse-stream';
|
||
import { MetonaStreamEventType, type MetonaStreamEvent } from '../../types';
|
||
|
||
/** 构造 SSE 测试流 */
|
||
function makeStream(chunks: string[]): ReadableStream<Uint8Array> {
|
||
const encoder = new TextEncoder();
|
||
return new ReadableStream<Uint8Array>({
|
||
start(controller) {
|
||
for (const c of chunks) controller.enqueue(encoder.encode(c));
|
||
controller.close();
|
||
},
|
||
});
|
||
}
|
||
|
||
async function collect(
|
||
stream: ReadableStream<Uint8Array>,
|
||
): Promise<Array<{ type: string; [k: string]: unknown }>> {
|
||
const events: Array<{ type: string; [k: string]: unknown }> = [];
|
||
for await (const ev of parseSSEStream(stream, 'req_test', 'sess_test', 1)) {
|
||
events.push(ev as unknown as { type: string; [k: string]: unknown });
|
||
}
|
||
return events;
|
||
}
|
||
|
||
describe('parseSSEStream — 文本与推理增量', () => {
|
||
it('TEXT_DELTA 事件按序产出', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"content":"你好"}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"content":"，世界"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const deltas = events.filter((e) => e.type === MetonaStreamEventType.TEXT_DELTA);
|
||
expect(deltas).toHaveLength(2);
|
||
expect(deltas[0].delta).toBe('你好');
|
||
expect(deltas[1].delta).toBe('，世界');
|
||
expect(events[events.length - 1].type).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
|
||
it('REASONING_DELTA（Thinking 模式）事件产出', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"reasoning_content":"让我想想"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const reasoning = events.find((e) => e.type === MetonaStreamEventType.REASONING_DELTA);
|
||
expect(reasoning?.delta).toBe('让我想想');
|
||
});
|
||
|
||
it('同一个 chunk 中 content 和 reasoning_content 同时产出', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"content":"答","reasoning_content":"思考"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
expect(events.filter((e) => e.type === MetonaStreamEventType.TEXT_DELTA)).toHaveLength(1);
|
||
expect(events.filter((e) => e.type === MetonaStreamEventType.REASONING_DELTA)).toHaveLength(1);
|
||
});
|
||
});
|
||
|
||
describe('parseSSEStream — 工具调用增量拼接', () => {
|
||
it('分段 arguments 拼接为完整 JSON 并在 finish_reason=tool_calls 时 flush', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"read_file","arguments":"{\\"pa"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"th\\": \\"a.ts\\"}"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const completes = events.filter((e) => e.type === MetonaStreamEventType.TOOL_CALL_COMPLETE);
|
||
expect(completes).toHaveLength(1);
|
||
const tc = completes[0].toolCall as { name: string; args: Record<string, unknown> };
|
||
expect(tc.name).toBe('read_file');
|
||
expect(tc.args).toEqual({ path: 'a.ts' });
|
||
});
|
||
|
||
it('多个工具调用按 index 分别拼接', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"tool_a","arguments":"{}"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":1,"function":{"name":"tool_b","arguments":"{}"}}]}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const completes = events.filter((e) => e.type === MetonaStreamEventType.TOOL_CALL_COMPLETE);
|
||
expect(completes).toHaveLength(2);
|
||
const names = completes.map((e) => (e.toolCall as { name: string }).name);
|
||
expect(names).toEqual(['tool_a', 'tool_b']);
|
||
});
|
||
|
||
it('损坏的 args JSON 转为截断错误 tool call 回传模型（不丢弃、不中断流）', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"bad_tool","arguments":"{invalid json"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"content":"后续文本"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
// v0.6.3 契约: 坏 JSON 的工具调用不再静默丢弃（丢弃会让引擎误判 COMPLETED
|
||
// 空回复终止会话）— 转为携带 _truncatedArguments 错误说明的 tool call，
|
||
// 工具执行失败后错误结果回传模型触发自愈
|
||
const completes = events.filter((e) => e.type === MetonaStreamEventType.TOOL_CALL_COMPLETE);
|
||
expect(completes).toHaveLength(1);
|
||
const tc = completes[0].toolCall as { name: string; args: Record<string, unknown> };
|
||
expect(tc.name).toBe('bad_tool');
|
||
expect(tc.args._truncatedArguments).toBe(true);
|
||
// 流继续处理后续事件
|
||
expect(events.some((e) => e.type === MetonaStreamEventType.TEXT_DELTA)).toBe(true);
|
||
expect(events[events.length - 1].type).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
});
|
||
|
||
describe('parseSSEStream — USAGE 与容错', () => {
|
||
it('USAGE 事件解析 DeepSeek 缓存字段', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[],"usage":{"prompt_tokens":100,"completion_tokens":50,"total_tokens":150,"prompt_cache_hit_tokens":80,"prompt_cache_miss_tokens":20}}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const usageEvent = events.find((e) => e.type === MetonaStreamEventType.USAGE);
|
||
const usage = usageEvent?.usage as Record<string, number>;
|
||
expect(usage.inputTokens).toBe(100);
|
||
expect(usage.outputTokens).toBe(50);
|
||
expect(usage.totalTokens).toBe(150);
|
||
expect(usage.cacheHitTokens).toBe(80);
|
||
expect(usage.cacheMissTokens).toBe(20);
|
||
});
|
||
|
||
it('坏 JSON 行不中断后续事件（记录警告后继续）', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {broken json\n\n',
|
||
'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
expect(events.some((e) => e.type === MetonaStreamEventType.TEXT_DELTA)).toBe(true);
|
||
expect(events[events.length - 1].type).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
|
||
it('跨 chunk 分割的 SSE 行正确拼接', async () => {
|
||
// "data: {...}\n\n" 被切到两个网络 chunk 中
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"cont',
|
||
'ent":"拼接成功"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const delta = events.find((e) => e.type === MetonaStreamEventType.TEXT_DELTA);
|
||
expect(delta?.delta).toBe('拼接成功');
|
||
});
|
||
|
||
it('空行与非 data 行被忽略', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
': comment line\n\n',
|
||
'\n',
|
||
'data: {"choices":[{"delta":{"content":"x"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
expect(events.filter((e) => e.type === MetonaStreamEventType.TEXT_DELTA)).toHaveLength(1);
|
||
});
|
||
});
|
||
|
||
describe('parseOpenAICompatibleResponse — 非流式响应', () => {
|
||
it('解析普通文本响应', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [{ message: { content: 'hello' }, finish_reason: 'stop' }],
|
||
usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 },
|
||
});
|
||
expect(result.content).toBe('hello');
|
||
expect(result.finishReason).toBe('stop');
|
||
expect(result.usage.inputTokens).toBe(10);
|
||
expect(result.toolCalls).toBeUndefined();
|
||
});
|
||
|
||
it('解析 reasoning_content（Thinking 模式）', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{ message: { content: 'answer', reasoning_content: 'thinking...' }, finish_reason: 'stop' },
|
||
],
|
||
usage: {},
|
||
});
|
||
expect(result.reasoningContent).toBe('thinking...');
|
||
});
|
||
|
||
it('解析 tool_calls（字符串 arguments 反序列化）', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{
|
||
message: {
|
||
content: null,
|
||
tool_calls: [{ id: 'tc_1', function: { name: 'run', arguments: '{"cmd":"ls"}' } }],
|
||
},
|
||
finish_reason: 'tool_calls',
|
||
},
|
||
],
|
||
usage: {},
|
||
});
|
||
expect(result.toolCalls).toHaveLength(1);
|
||
expect(result.toolCalls![0].name).toBe('run');
|
||
expect(result.toolCalls![0].args).toEqual({ cmd: 'ls' });
|
||
});
|
||
|
||
it('损坏的 tool_calls arguments 转为 _truncatedArguments 自愈载荷（v0.6.4: 不再静默降级 {}）', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{
|
||
message: {
|
||
content: null,
|
||
tool_calls: [{ id: 'tc_1', function: { name: 'run', arguments: '{bad' } }],
|
||
},
|
||
finish_reason: 'tool_calls',
|
||
},
|
||
],
|
||
usage: {},
|
||
});
|
||
const args = result.toolCalls![0].args as Record<string, unknown>;
|
||
expect(args._truncatedArguments).toBe(true);
|
||
expect(String(args._truncatedReason)).toContain('truncated');
|
||
});
|
||
|
||
it('mapOpenAIFinishReason 覆盖 MiMo repetition_truncation', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [{ message: { content: 'x' }, finish_reason: 'repetition_truncation' }],
|
||
usage: {},
|
||
});
|
||
expect(result.finishReason).toBe('stop');
|
||
});
|
||
});
|
||
|
||
// ===== v0.7.4 P1-2: 流空闲超时 =====
|
||
|
||
describe('parseSSEStream — 流空闲超时（P1-2）', () => {
|
||
it('连续无数据超过空闲阈值时抛 SseUpstreamError(504)', async () => {
|
||
const { readStreamChunkWithIdleTimeout } = await import('../shared/sse-stream');
|
||
|
||
// 构造一个永不返回数据的 reader（挂死流）
|
||
const neverReader = {
|
||
read: () => new Promise<{ done: boolean; value?: Uint8Array }>(() => {}),
|
||
releaseLock: () => {},
|
||
cancel: () => Promise.resolve(),
|
||
closed: Promise.resolve(),
|
||
} as unknown as ReadableStreamDefaultReader<Uint8Array>;
|
||
|
||
// 使用共享辅助函数直接验证：空闲超时抛 SseUpstreamError(504)
|
||
await expect(readStreamChunkWithIdleTimeout(neverReader, 50)).rejects.toMatchObject({
|
||
status: 504,
|
||
});
|
||
|
||
// 数据到达时重置空闲窗口（不抛错）
|
||
const delayedReader = {
|
||
read: () =>
|
||
new Promise<{ done: boolean; value?: Uint8Array }>((resolve) => {
|
||
setTimeout(() => resolve({ done: true, value: undefined }), 10);
|
||
}),
|
||
releaseLock: () => {},
|
||
cancel: () => Promise.resolve(),
|
||
closed: Promise.resolve(),
|
||
} as unknown as ReadableStreamDefaultReader<Uint8Array>;
|
||
const result = await readStreamChunkWithIdleTimeout(delayedReader, 200);
|
||
expect(result.done).toBe(true);
|
||
});
|
||
});
|
||
|
||
// ===== 追加：readStreamChunkWithIdleTimeout 专项 =====
|
||
|
||
describe('readStreamChunkWithIdleTimeout — 空闲超时语义', () => {
|
||
afterEach(() => {
|
||
vi.restoreAllMocks();
|
||
vi.useRealTimers();
|
||
});
|
||
|
||
function makeReader(
|
||
reads: Array<{ delay: number; value?: string; done?: boolean }>,
|
||
): ReadableStreamDefaultReader<Uint8Array> {
|
||
const encoder = new TextEncoder();
|
||
let i = 0;
|
||
return {
|
||
read: () =>
|
||
new Promise<{ done: boolean; value?: Uint8Array }>((resolve) => {
|
||
const spec = reads[Math.min(i, reads.length - 1)];
|
||
i++;
|
||
setTimeout(() => {
|
||
if (spec.done) resolve({ done: true, value: undefined });
|
||
else
|
||
resolve({ done: false, value: spec.value ? encoder.encode(spec.value) : undefined });
|
||
}, spec.delay);
|
||
}),
|
||
releaseLock: () => {},
|
||
cancel: () => Promise.resolve(),
|
||
closed: Promise.resolve(),
|
||
} as unknown as ReadableStreamDefaultReader<Uint8Array>;
|
||
}
|
||
|
||
it('空闲超时抛 SseUpstreamError(504)，消息含 idle timeout', async () => {
|
||
const reader = makeReader([{ delay: 10_000 }]);
|
||
const promise = readStreamChunkWithIdleTimeout(reader, 30);
|
||
await expect(promise).rejects.toMatchObject({
|
||
name: 'SseUpstreamError',
|
||
status: 504,
|
||
message: expect.stringContaining('idle timeout'),
|
||
});
|
||
});
|
||
|
||
it('数据在超时前到达 → 正常返回数据（空闲窗口被重置）', async () => {
|
||
const reader = makeReader([
|
||
{ delay: 5, value: 'data: {}' },
|
||
{ done: true, delay: 1 },
|
||
]);
|
||
const first = await readStreamChunkWithIdleTimeout(reader, 100);
|
||
expect(first.done).toBe(false);
|
||
expect(new TextDecoder().decode(first.value)).toBe('data: {}');
|
||
const second = await readStreamChunkWithIdleTimeout(reader, 100);
|
||
expect(second.done).toBe(true);
|
||
});
|
||
|
||
it('连续多次 read 均未超时（每次数据到达重置计时器）', async () => {
|
||
const reader = makeReader([
|
||
{ delay: 5, value: 'a' },
|
||
{ delay: 5, value: 'b' },
|
||
{ done: true, delay: 5 },
|
||
]);
|
||
for (const expectValue of ['a', 'b']) {
|
||
const r = await readStreamChunkWithIdleTimeout(reader, 50);
|
||
expect(new TextDecoder().decode(r.value)).toBe(expectValue);
|
||
}
|
||
const last = await readStreamChunkWithIdleTimeout(reader, 50);
|
||
expect(last.done).toBe(true);
|
||
});
|
||
|
||
it('超时后 timer 被清理（fake timers 验证无遗留定时器）', async () => {
|
||
vi.useFakeTimers();
|
||
const reader = {
|
||
read: () => new Promise<{ done: boolean; value?: Uint8Array }>(() => {}),
|
||
releaseLock: () => {},
|
||
cancel: () => Promise.resolve(),
|
||
closed: Promise.resolve(),
|
||
} as unknown as ReadableStreamDefaultReader<Uint8Array>;
|
||
const promise = readStreamChunkWithIdleTimeout(reader, 100);
|
||
// 先挂上 rejection 处理器，再推进计时器，避免 unhandledRejection 竞态
|
||
const assertion = expect(promise).rejects.toMatchObject({ status: 504 });
|
||
await vi.advanceTimersByTimeAsync(150);
|
||
await assertion;
|
||
// finally 清理后不应有残留 timer
|
||
expect(vi.getTimerCount()).toBe(0);
|
||
});
|
||
|
||
it('数据到达正常结束后 timer 同样被清理', async () => {
|
||
vi.useFakeTimers();
|
||
const encoder = new TextEncoder();
|
||
const reader = {
|
||
read: vi.fn(
|
||
() =>
|
||
Promise.resolve({ done: true, value: undefined }) as Promise<{
|
||
done: boolean;
|
||
value?: Uint8Array;
|
||
}>,
|
||
),
|
||
releaseLock: () => {},
|
||
cancel: () => Promise.resolve(),
|
||
closed: Promise.resolve(),
|
||
} as unknown as ReadableStreamDefaultReader<Uint8Array>;
|
||
const r = await readStreamChunkWithIdleTimeout(reader, 100);
|
||
expect(r.done).toBe(true);
|
||
expect(vi.getTimerCount()).toBe(0);
|
||
void encoder;
|
||
});
|
||
});
|
||
|
||
// ===== 追加：parseSSEStream 更多解析细节 =====
|
||
|
||
describe('parseSSEStream — 帧格式兼容扩展', () => {
|
||
it('CRLF 行结束符（\\r\\n）正常解析', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"content":"crlf"}}]}\r\n\r\n',
|
||
'data: [DONE]\r\n\r\n',
|
||
]),
|
||
);
|
||
expect(
|
||
events.some((e) => e.type === MetonaStreamEventType.TEXT_DELTA && e.delta === 'crlf'),
|
||
).toBe(true);
|
||
expect(events[events.length - 1].type).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
|
||
it('同一 chunk 内连续多个 SSE 事件均被处理', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"content":"一"}}]}\n\n' +
|
||
'data: {"choices":[{"delta":{"content":"二"}}]}\n\n' +
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const deltas = events.filter((e) => e.type === MetonaStreamEventType.TEXT_DELTA);
|
||
expect(deltas.map((e) => e.delta)).toEqual(['一', '二']);
|
||
});
|
||
|
||
it('一个 SSE 行被切成 4 个网络 chunk 仍能拼接', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"del',
|
||
'ta":{"content":"跨',
|
||
'chunk拼接成功',
|
||
'"}}]}\n\ndata: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const delta = events.find((e) => e.type === MetonaStreamEventType.TEXT_DELTA);
|
||
expect(delta?.delta).toBe('跨chunk拼接成功');
|
||
});
|
||
|
||
it('data: 后无空格且无内容的空 data 行被忽略', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data:\n\n',
|
||
'data: {"choices":[{"delta":{"content":"x"}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
expect(events.filter((e) => e.type === MetonaStreamEventType.TEXT_DELTA)).toHaveLength(1);
|
||
});
|
||
|
||
it('choices 存在但 delta 为空对象 → 不产生事件（静默帧）', async () => {
|
||
const events = await collect(
|
||
makeStream(['data: {"choices":[{"delta":{}}]}\n\n', 'data: [DONE]\n\n']),
|
||
);
|
||
expect(events.filter((e) => e.type === MetonaStreamEventType.TEXT_DELTA)).toHaveLength(0);
|
||
expect(events[events.length - 1].type).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
|
||
it('工具调用：arguments 被切成 5 段拼接为合法 JSON', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"f","arguments":"{\\"a\\":"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"1,"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\\"b\\":"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"[1,2]}"}}]}}]}\n\n',
|
||
'data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const completes = events.filter((e) => e.type === MetonaStreamEventType.TOOL_CALL_COMPLETE);
|
||
expect(completes).toHaveLength(1);
|
||
expect((completes[0].toolCall as { args: Record<string, unknown> }).args).toEqual({
|
||
a: 1,
|
||
b: [1, 2],
|
||
});
|
||
});
|
||
|
||
it('工具调用 arguments 为空 → 完成时 args 为 {}（无参工具合法）', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"noop"}}]}}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const completes = events.filter((e) => e.type === MetonaStreamEventType.TOOL_CALL_COMPLETE);
|
||
expect(completes).toHaveLength(1);
|
||
expect((completes[0].toolCall as { args: Record<string, unknown> }).args).toEqual({});
|
||
});
|
||
|
||
it('流断开（无 [DONE]）时 flush 缓冲工具调用 + 补发 DONE', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"last_tool","arguments":"{\\"ok\\":true}"}}]}}]}\n\n',
|
||
]),
|
||
);
|
||
const completes = events.filter((e) => e.type === MetonaStreamEventType.TOOL_CALL_COMPLETE);
|
||
expect(completes).toHaveLength(1);
|
||
expect(events[events.length - 1].type).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
});
|
||
|
||
// ===== 追加：USAGE 扩展字段 =====
|
||
|
||
describe('parseSSEStream — USAGE 扩展字段', () => {
|
||
it('MiMo prompt_tokens_details.cached_tokens 映射 cacheHitTokens', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":5,"total_tokens":15,"prompt_tokens_details":{"cached_tokens":7}}}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const usage = events.find((e) => e.type === MetonaStreamEventType.USAGE)?.usage as Record<
|
||
string,
|
||
unknown
|
||
>;
|
||
expect(usage.cacheHitTokens).toBe(7);
|
||
expect(usage.cacheMissTokens).toBeUndefined();
|
||
});
|
||
|
||
it('DeepSeek reasoning_tokens 映射 reasoningTokens', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2,"completion_tokens_details":{"reasoning_tokens":99}}}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const usage = events.find((e) => e.type === MetonaStreamEventType.USAGE)?.usage as Record<
|
||
string,
|
||
unknown
|
||
>;
|
||
expect(usage.reasoningTokens).toBe(99);
|
||
});
|
||
|
||
it('无 usage 字段的数据帧不产生 USAGE 事件', async () => {
|
||
const events = await collect(
|
||
makeStream(['data: {"choices":[{"delta":{"content":"x"}}]}\n\n', 'data: [DONE]\n\n']),
|
||
);
|
||
expect(events.some((e) => e.type === MetonaStreamEventType.USAGE)).toBe(false);
|
||
});
|
||
});
|
||
|
||
// ===== 追加：上游错误帧 status 推断矩阵 =====
|
||
|
||
describe('parseSSEStream — 上游错误帧 status 推断矩阵', () => {
|
||
async function collectError(json: unknown): Promise<Error> {
|
||
const stream = makeStream([`data: ${JSON.stringify(json)}\n\n`]);
|
||
try {
|
||
for await (const _ev of parseSSEStream(stream, 'r', 's', 1)) void _ev;
|
||
} catch (err) {
|
||
return err as Error;
|
||
}
|
||
throw new Error('expected throw');
|
||
}
|
||
|
||
it.each([
|
||
[{ error: { code: 'rate_limit_exceeded', message: 'rl' } }, 429],
|
||
[{ error: { code: 'too_many_requests', message: 'rl' } }, 429],
|
||
[{ error: { code: 'insufficient_quota', message: 'q' } }, 402],
|
||
[{ error: { code: 'billing_error', message: 'q' } }, 402],
|
||
[{ error: { code: 'exceeded_balance', message: 'q' } }, 402],
|
||
[{ error: { code: 'invalid_api_key', message: 'auth' } }, 401],
|
||
[{ error: { code: 'authentication_error', message: 'auth' } }, 401],
|
||
[{ error: { code: 'forbidden', message: 'deny' } }, 403],
|
||
[{ error: { code: 'permission_denied', message: 'deny' } }, 403],
|
||
[{ error: { code: 'model_not_found', message: 'no model' } }, 404],
|
||
[{ error: { code: 'overloaded', message: 'busy' } }, 503],
|
||
[{ error: { code: 'capacity_exceeded', message: 'busy' } }, 503],
|
||
[{ error: { code: 'server_error', message: '500' } }, 500],
|
||
[{ error: { code: 'internal_error', message: '500' } }, 500],
|
||
] as const)('providerCode %s → status %d', async (errorObj, expectedStatus) => {
|
||
const err = await collectError(errorObj);
|
||
expect(err).toBeInstanceOf(SseUpstreamError);
|
||
expect((err as SseUpstreamError).status).toBe(expectedStatus);
|
||
});
|
||
|
||
it('无效参数类 code（无匹配规则）→ status undefined（非重试语义）', async () => {
|
||
const err = await collectError({
|
||
error: { code: 'invalid_request_error', message: 'bad param' },
|
||
});
|
||
expect(err).toBeInstanceOf(SseUpstreamError);
|
||
expect((err as SseUpstreamError).status).toBeUndefined();
|
||
expect((err as SseUpstreamError).providerCode).toBe('invalid_request_error');
|
||
});
|
||
|
||
it('status_code 数字字段被识别（非 status 命名）', async () => {
|
||
const err = await collectError({
|
||
error: { message: 'gateway timeout', status_code: 504 },
|
||
});
|
||
expect((err as SseUpstreamError).status).toBe(504);
|
||
});
|
||
|
||
it('msg 字段作为错误消息（非 message 命名）', async () => {
|
||
const err = await collectError({ error: { msg: 'custom msg', code: 'overloaded' } });
|
||
expect((err as Error).message).toContain('custom msg');
|
||
expect((err as SseUpstreamError).status).toBe(503);
|
||
});
|
||
|
||
it('error 为空对象 → 不构成错误帧（v0.8.0 P1-3.1 防御分支根治后契约）', async () => {
|
||
// v0.8.0 P1-3.1 根治: 旧实现 message 兜底取 JSON.stringify({})='{}' 恒真值，
|
||
// `{"error":{}}` 被误判为错误帧抛 SseUpstreamError 并触发不必要的重试。
|
||
// 新契约：无 message 且无 status/code 标识的空对象为无害帧，静默跳过。
|
||
const events: MetonaStreamEvent[] = [];
|
||
let error: Error | null = null;
|
||
try {
|
||
for await (const ev of parseSSEStream(
|
||
makeStream(['data: {"error":{}}\n\n', 'data: [DONE]\n\n']),
|
||
'r',
|
||
's',
|
||
1,
|
||
)) {
|
||
events.push(ev);
|
||
}
|
||
} catch (e) {
|
||
error = e as Error;
|
||
}
|
||
expect(error).toBeNull();
|
||
expect(events.some((ev) => ev.type === MetonaStreamEventType.DONE)).toBe(true);
|
||
});
|
||
|
||
it('error 有 code 无 message → 仍构成错误帧（标识存在，占位文案兜底）', async () => {
|
||
const err = await collectError({ error: { code: 'rate_limit_exceeded' } });
|
||
expect(err).toBeInstanceOf(SseUpstreamError);
|
||
expect((err as Error).message).toContain('upstream_error');
|
||
expect((err as SseUpstreamError).status).toBe(429);
|
||
});
|
||
|
||
it('error 为空白字符串 → 不算错误帧', async () => {
|
||
const events: string[] = [];
|
||
for await (const ev of parseSSEStream(
|
||
makeStream(['data: {"error":" "}\n\n', 'data: [DONE]\n\n']),
|
||
'r',
|
||
's',
|
||
1,
|
||
)) {
|
||
events.push(ev.type);
|
||
}
|
||
expect(events[events.length - 1]).toBe(MetonaStreamEventType.DONE);
|
||
});
|
||
});
|
||
|
||
// ===== 追加：parseOpenAICompatibleResponse 更多形态 =====
|
||
|
||
describe('parseOpenAICompatibleResponse — 补充形态', () => {
|
||
it('空 choices 数组 → content 空字符串 + finishReason stop + 无 toolCalls', () => {
|
||
const result = parseOpenAICompatibleResponse({ choices: [], usage: {} });
|
||
expect(result.content).toBe('');
|
||
expect(result.finishReason).toBe('stop');
|
||
expect(result.toolCalls).toBeUndefined();
|
||
});
|
||
|
||
it('无 message 的 choice → 安全回退', () => {
|
||
const result = parseOpenAICompatibleResponse({ choices: [{ finish_reason: 'stop' }] });
|
||
expect(result.content).toBe('');
|
||
expect(result.reasoningContent).toBeUndefined();
|
||
});
|
||
|
||
it('usage 缺失 → 全 0 token', () => {
|
||
const result = parseOpenAICompatibleResponse({ choices: [{ message: { content: 'x' } }] });
|
||
expect(result.usage).toEqual({ inputTokens: 0, outputTokens: 0, totalTokens: 0 });
|
||
});
|
||
|
||
it('usage cache 双格式回退（DeepSeek 命中 + MiMo cached_tokens）', () => {
|
||
const ds = parseOpenAICompatibleResponse({
|
||
choices: [{ message: { content: 'x' } }],
|
||
usage: { prompt_cache_hit_tokens: 5, prompt_cache_miss_tokens: 6 },
|
||
});
|
||
expect(ds.usage.cacheHitTokens).toBe(5);
|
||
expect(ds.usage.cacheMissTokens).toBe(6);
|
||
|
||
const mimo = parseOpenAICompatibleResponse({
|
||
choices: [{ message: { content: 'x' } }],
|
||
usage: { prompt_tokens_details: { cached_tokens: 9 } },
|
||
});
|
||
expect(mimo.usage.cacheHitTokens).toBe(9);
|
||
});
|
||
|
||
it('arguments 为空字符串 → JSON.parse 失败走截断自愈载荷（与坏参同轨）', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{
|
||
message: { tool_calls: [{ id: 'c', function: { name: 'f', arguments: '' } }] },
|
||
finish_reason: 'tool_calls',
|
||
},
|
||
],
|
||
});
|
||
expect((result.toolCalls![0].args as Record<string, unknown>)._truncatedArguments).toBe(true);
|
||
});
|
||
|
||
it('arguments 为对象类型 → 原样保留（不解析）', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{
|
||
message: { tool_calls: [{ id: 'c', function: { name: 'f', arguments: { a: 1 } } }] },
|
||
finish_reason: 'tool_calls',
|
||
},
|
||
],
|
||
});
|
||
expect(result.toolCalls![0].args).toEqual({ a: 1 });
|
||
});
|
||
|
||
it('reasoning_tokens 透传到 usage.reasoningTokens', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [{ message: { content: 'x' } }],
|
||
usage: { completion_tokens_details: { reasoning_tokens: 42 } },
|
||
});
|
||
expect(result.usage.reasoningTokens).toBe(42);
|
||
});
|
||
|
||
it('tool_calls 元素缺 id → id undefined 不抛错', () => {
|
||
const result = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{
|
||
message: { tool_calls: [{ function: { name: 'f', arguments: '{}' } }] },
|
||
finish_reason: 'tool_calls',
|
||
},
|
||
],
|
||
});
|
||
expect(result.toolCalls).toHaveLength(1);
|
||
expect(result.toolCalls![0].id).toBeUndefined();
|
||
});
|
||
});
|
||
|
||
describe('parseSSEStream — v0.8.2 P2-6 MiMo 联网搜索引用回填', () => {
|
||
it('message.annotations 引用以 TEXT_DELTA 回填正文（DONE 前）', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"choices":[{"delta":{"content":"今天的新闻"}}],"annotations":[]}\n\n',
|
||
'data: {"choices":[{"delta":{},"finish_reason":null}],"annotations":[{"title":"科技日报","url":"https://news.example.com/a","site_name":"example"}]}\n\n',
|
||
'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const citation = events.filter(
|
||
(e) => e.type === MetonaStreamEventType.TEXT_DELTA && String(e.delta).includes('References'),
|
||
);
|
||
expect(citation).toHaveLength(1);
|
||
expect(String(citation[0].delta)).toContain('https://news.example.com/a');
|
||
expect(String(citation[0].delta)).toContain('科技日报');
|
||
// 引用块必须在 DONE 之前
|
||
expect(events.indexOf(citation[0])).toBeLessThan(
|
||
events.findIndex((e) => e.type === MetonaStreamEventType.DONE),
|
||
);
|
||
});
|
||
|
||
it('按 url 去重；非流式路径同样回填', async () => {
|
||
const events = await collect(
|
||
makeStream([
|
||
'data: {"annotations":[{"title":"a","url":"https://x.example/1"},{"title":"a-dup","url":"https://x.example/1"}]}\n\n',
|
||
'data: [DONE]\n\n',
|
||
]),
|
||
);
|
||
const citation = events.find(
|
||
(e) => e.type === MetonaStreamEventType.TEXT_DELTA && String(e.delta).includes('References'),
|
||
);
|
||
expect(citation).toBeDefined();
|
||
expect(String(citation!.delta).match(/https:\/\/x.example\/1/g)).toHaveLength(1);
|
||
|
||
const res = parseOpenAICompatibleResponse({
|
||
choices: [
|
||
{
|
||
message: { content: 'answer', annotations: [{ title: 'b', url: 'https://x.example/2' }] },
|
||
},
|
||
],
|
||
usage: {},
|
||
});
|
||
expect(res.content).toContain('answer');
|
||
expect(res.content).toContain('https://x.example/2');
|
||
});
|
||
});
|