d3ro-voice/apps/mobile-rn/__tests__/talk-llm-stream.test.ts

import {
  parseSseEventBlock,
  streamTalkResponse,
} from '../src/features/talk/llm-stream-service'

const originalFetch = global.fetch
const GENERATION_ID = '11111111-1111-4111-8111-111111111111'

function headers(
  contentType: string,
  generationId: string | null = GENERATION_ID,
): { get(name: string): string | null } {
  return {
    get: (name) => {
      if (name.toLowerCase() === 'content-type') return contentType
      if (name.toLowerCase() === 'x-d3ro-generation-id') return generationId
      return null
    },
  }
}

function byteChunks(value: string, chunkSizes: number[]): Uint8Array[] {
  const bytes = Uint8Array.from(Buffer.from(value, 'utf8'))
  const chunks: Uint8Array[] = []
  let offset = 0
  for (const size of chunkSizes) {
    if (offset >= bytes.length) break
    chunks.push(bytes.slice(offset, Math.min(bytes.length, offset + size)))
    offset += size
  }
  if (offset < bytes.length) chunks.push(bytes.slice(offset))
  return chunks
}

function streamingResponse(chunks: Uint8Array[]): Response {
  let index = 0
  return {
    ok: true,
    status: 200,
    headers: headers('text/event-stream'),
    body: {
      getReader: () => ({
        read: async () => index < chunks.length
          ? { done: false, value: chunks[index++] }
          : { done: true, value: undefined },
        releaseLock: jest.fn(),
      }),
    },
  } as unknown as Response
}

afterEach(() => {
  global.fetch = originalFetch
  jest.restoreAllMocks()
})

describe('Talk LLM streaming', () => {
  it('parses SSE fields and joins multiple data lines', () => {
    expect(parseSseEventBlock('event: message\ndata: {"a":1}\ndata: {"b":2}'))
      .toEqual({ event: 'message', data: '{"a":1}\n{"b":2}' })
    expect(parseSseEventBlock(': ping\nevent: ping')).toBeNull()
  })

  it('streams real Anthropic text deltas across UTF-8 and network chunk boundaries', async () => {
    const sse = [
      'event: content_block_delta\n',
      'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"안녕 "}}\n\n',
      'event: content_block_delta\r\n',
      'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"world"}}\r\n\r\n',
      'event: message_stop\n',
      'data: {"type":"message_stop"}\n\n',
    ].join('')
    global.fetch = jest.fn().mockResolvedValue(streamingResponse(byteChunks(sse, [1, 2, 5, 7, 13, 3])))
    const updates: string[] = []

    const result = await streamTalkResponse(
      [{ role: 'user', content: 'hello' }],
      {
        accessToken: 'real-user-token',
        onTextDelta: (_delta, accumulated) => updates.push(accumulated),
      },
    )

    expect(result).toEqual({ text: '안녕 world', generationId: GENERATION_ID })
    expect(updates).toEqual(['안녕 ', '안녕 world'])
    const request = (global.fetch as jest.Mock).mock.calls[0][1]
    expect(JSON.parse(request.body)).toMatchObject({ stream: true, max_tokens: 1024 })
    expect(request.headers).toMatchObject({
      Authorization: 'Bearer real-user-token',
      Accept: 'text/event-stream',
      'X-D3RO-Generation-Purpose': 'talk_response',
    })
  })

  it('parses the same SSE response without issuing a second request when RN buffers the body', async () => {
    const body = [
      'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"buffered"}}\n\n',
      'data: {"type":"message_stop"}\n\n',
    ].join('')
    global.fetch = jest.fn().mockResolvedValue({
      ok: true,
      status: 200,
      headers: headers('text/event-stream'),
      body: null,
      text: async () => body,
    })
    const updates: string[] = []

    await expect(streamTalkResponse(
      [{ role: 'user', content: 'hello' }],
      { accessToken: 'token', onTextDelta: (delta) => updates.push(delta) },
    )).resolves.toEqual({ text: 'buffered', generationId: GENERATION_ID })
    expect(updates).toEqual(['buffered'])
    expect(global.fetch).toHaveBeenCalledTimes(1)
  })

  it('accepts a truthful non-streaming JSON result if the edge runtime cannot expose a stream body', async () => {
    global.fetch = jest.fn().mockResolvedValue({
      ok: true,
      status: 200,
      headers: headers('application/json'),
      json: async () => ({ content: [{ type: 'text', text: 'real answer' }] }),
    })
    const updates: string[] = []

    await expect(streamTalkResponse(
      [{ role: 'user', content: 'hello' }],
      { accessToken: 'token', onTextDelta: (delta) => updates.push(delta) },
    )).resolves.toEqual({ text: 'real answer', generationId: GENERATION_ID })
    expect(updates).toEqual(['real answer'])
  })

  it('fails closed on truncated SSE and maps provider outages without fallback text', async () => {
    global.fetch = jest.fn().mockResolvedValueOnce(streamingResponse(byteChunks(
      'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"partial"}}\n\n',
      [4, 9],
    )))
    await expect(streamTalkResponse(
      [{ role: 'user', content: 'hello' }],
      { accessToken: 'token', onTextDelta: jest.fn() },
    )).rejects.toMatchObject({ code: 'INVALID_RESPONSE' })

    global.fetch = jest.fn().mockResolvedValueOnce({
      ok: false,
      status: 503,
      headers: headers('application/json'),
      text: async () => JSON.stringify({ error: 'provider_unavailable' }),
    })
    await expect(streamTalkResponse(
      [{ role: 'user', content: 'hello' }],
      { accessToken: 'token', onTextDelta: jest.fn() },
    )).rejects.toMatchObject({ code: 'PROVIDER_UNAVAILABLE', retryable: true })
  })

  it('fails closed when a successful response has no reportable generation receipt', async () => {
    const body = [
      'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"answer"}}\n\n',
      'data: {"type":"message_stop"}\n\n',
    ].join('')
    global.fetch = jest.fn().mockResolvedValue({
      ok: true,
      status: 200,
      headers: headers('text/event-stream', null),
      body: null,
      text: async () => body,
    })

    await expect(streamTalkResponse(
      [{ role: 'user', content: 'hello' }],
      { accessToken: 'token', onTextDelta: jest.fn() },
    )).rejects.toMatchObject({ code: 'INVALID_RESPONSE', retryable: true })
  })
})