metona-ai-desktop/electron/harness/adapters/__tests__/sse-stream.test.ts

/**
 * SSE 流式解析器单元测试（v0.4.1 测试补齐）
 * 覆盖：TEXT_DELTA / REASONING_DELTA / TOOL_CALL 增量拼接 / USAGE /
 *       [DONE] / finish_reason=tool_calls 提前 flush / 坏 JSON 行容错 / 损坏工具调用跳过
 */

import { describe, it, expect } from 'vitest';
import { parseSSEStream, parseOpenAICompatibleResponse } from '../shared/sse-stream';
import { MetonaStreamEventType } 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');
  });
});