RAG/frontend/tests/chat-protocol.test.mjs

import assert from 'node:assert/strict'
import { afterEach, beforeEach, test } from 'node:test'
import { fileURLToPath } from 'node:url'
import { build } from 'esbuild'

// Exercise the production TypeScript with Vite's existing esbuild dependency, no test framework.
const root = fileURLToPath(new URL('../', import.meta.url))
async function loadModule(entry) {
  const result = await build({
    entryPoints: [root + entry], bundle: true, write: false, format: 'esm', platform: 'node',
    alias: { '@': root + 'src' },
  })
  return import('data:text/javascript;base64,' + Buffer.from(result.outputFiles[0].text).toString('base64'))
}
const { readSSEStream, readSSEStreamWithEvents } = await loadModule('src/utils/sse.ts')
const { chatSync, fetchChatResult } = await loadModule('src/api/chat.ts')
const originalFetch = globalThis.fetch
const originalStorage = Object.getOwnPropertyDescriptor(globalThis, 'localStorage')

beforeEach(() => {
  Object.defineProperty(globalThis, 'localStorage', {
    configurable: true, value: { getItem: () => 'admin-token' },
  })
})
afterEach(() => {
  globalThis.fetch = originalFetch
  if (originalStorage) Object.defineProperty(globalThis, 'localStorage', originalStorage)
  else delete globalThis.localStorage
})

const sources = [{
  documentId: '9223372036854775806', title: '报销制度', sourceName: 'policy.pdf',
  chunkIndex: 2, score: 0.125, snippet: '申请应在三十天内提交。',
}]
const metadata = JSON.stringify({ object: 'chat.completion.chunk', choices: [], sources })
const answer = JSON.stringify({ choices: [{ delta: { content: '答复。' } }] })

function responseFor(text, { open = false, fragment = false, cancel } = {}) {
  const bytes = new TextEncoder().encode(text)
  const body = new ReadableStream({
    start(controller) {
      if (fragment) for (const byte of bytes) controller.enqueue(Uint8Array.of(byte))
      else controller.enqueue(bytes)
      if (!open) controller.close()
    },
    cancel,
  })
  return new Response(body)
}

test('fragmented CRLF metadata preserves ID/snippet and DONE completes without EOF or cancellation settlement', { timeout: 1000 }, async () => {
  let cancelled = 0
  const response = responseFor(
    `data: ${answer}\r\n\r\ndata: ${metadata}\r\n\r\ndata: [DONE]\r\n\r\ndata: ignored\r\n\r\n`,
    { open: true, fragment: true, cancel() { cancelled++; return new Promise(() => {}) } },
  )
  const requests = []
  globalThis.fetch = async (...args) => { requests.push(args); return response }
  const events = []
  await readSSEStreamWithEvents('/ai/chat/stream', {
    onMessage: text => events.push(['text', text]),
    onSources: value => events.push(['sources', value]),
    onDone: () => events.push(['done']),
  }, { Authorization: 'Bearer sdk-token' })
  assert.deepEqual(events, [['text', '答复。'], ['sources', sources], ['done']])
  assert.equal(requests.length, 1)
  assert.equal(requests[0][1].headers.Authorization, 'Bearer sdk-token')
  assert.equal(cancelled, 1)
  assert.equal(response.body.locked, false)
})

test('text-only facade uses identical DONE handling and never leaks empty choices or sources', { timeout: 1000 }, async () => {
  globalThis.fetch = async () => responseFor(
    `data: ${metadata}\n\ndata: ${answer}\n\ndata: [DONE]\n\ndata: [DONE]\n\ndata: trailing\n\n`,
    { open: true },
  )
  const text = []
  let completed = 0
  await readSSEStream('/ai/chat/stream', value => text.push(value), () => completed++)
  assert.deepEqual(text, ['答复。'])
  assert.equal(completed, 1)
})

test('EOF residuals, multiline data, tool events, status and legacy text retain their semantics', async () => {
  globalThis.fetch = async () => responseFor(
    ': heartbeat\r\nid: event-1\r\nretry: 1000\r\nevent: status\r\ndata: generating\r\n\r\n'
      + 'event: tool_call_start\r\ndata: {"tool":"lookup"}\r\n\r\n'
      + 'event: tool_call_result\r\ndata: {"tool":"lookup","result":"ok"}\r\n\r\n'
      + 'event: error\r\ndata: {"message":"tool unavailable"}\r\n\r\n'
      + 'data: first\r\ndata:   indented\r\n\r\ndata:\r\n\r\nlegacy\r\ndata: 尾部',
    { fragment: true },
  )
  const events = []
  await readSSEStreamWithEvents('/stream', {
    onMessage: text => events.push(text),
    onToolCallStart: value => events.push(value),
    onToolCallResult: value => events.push(value),
    onError: value => events.push(value),
    onDone: () => events.push('done'),
  })
  assert.deepEqual(events, [
    { tool: 'lookup' }, { tool: 'lookup', result: 'ok' }, { message: 'tool unavailable' },
    'first\n  indented', '\n', 'legacy', '尾部', 'done',
  ])
})

test('EOF flushes an unterminated plain-text line and an unterminated DONE event', async () => {
  for (const body of ['legacy tail', 'data: [DONE]']) {
    globalThis.fetch = async () => responseFor(body)
    const text = []
    let done = 0
    await readSSEStream('/stream', value => text.push(value), () => done++)
    assert.deepEqual(text, body.startsWith('data:') ? [] : ['legacy tail'])
    assert.equal(done, 1)
  }
})

test('cleanup failure cannot replace callback errors or call onDone on an error', async () => {
  const original = new Error('consumer failed')
  const response = responseFor(`data: ${answer}\n\n`, {
    open: true, cancel() { throw new Error('cleanup failed') },
  })
  globalThis.fetch = async () => response
  let done = 0
  await assert.rejects(readSSEStream('/stream', () => { throw original }, () => done++), error => error === original)
  assert.equal(done, 0)
  assert.equal(response.body.locked, false)
})

test('empty sources for ordinary answers are delivered separately from content', async () => {
  globalThis.fetch = async () => responseFor('data: {"choices":[],"sources":[]}\n\ndata: [DONE]\n\n')
  const values = []
  await readSSEStreamWithEvents('/stream', {
    onMessage: () => assert.fail('metadata is not text'),
    onSources: value => values.push(value),
  })
  assert.deepEqual(values, [[]])
})

test('synchronous chat makes one result request and retains direct answer, sources, explicit strategy and signal', async () => {
  const result = { text: '答复。', mcpEvents: [], suggestions: [], sources }
  const calls = []
  globalThis.fetch = async (...args) => { calls.push(args); return Response.json(result) }
  const controller = new AbortController()
  assert.deepEqual(await chatSync('费用?', 'conversation-1', {
    enableRag: true, roleId: '9223372036854775806', rewriteStrategy: 'MULTI_QUERY', categoryIds: ['123'],
  }, controller.signal), result)
  assert.equal(calls.length, 1)
  const url = new URL(calls[0][0], 'https://test.invalid')
  assert.equal(url.pathname, '/ai/chat/result')
  assert.equal(url.searchParams.get('rewriteStrategy'), 'MULTI_QUERY')
  assert.equal(url.searchParams.get('roleId'), '9223372036854775806')
  assert.equal(calls[0][1].signal, controller.signal)
  assert.equal(calls[0][1].headers.Authorization, 'Bearer admin-token')
})

test('synchronous SDK response uses the same direct contract and surfaces server errors', async () => {
  const result = { text: '普通回答', mcpEvents: [], suggestions: [], sources: [] }
  globalThis.fetch = async () => Response.json(result)
  assert.deepEqual(await fetchChatResult('https://sdk.invalid/ai/chat/result', {}), result)
  globalThis.fetch = async () => new Response('角色无访问权限', { status: 403 })
  await assert.rejects(fetchChatResult('/ai/chat/result', {}), /角色无访问权限/)
  await assert.rejects(readSSEStream('/ai/chat/stream', () => {}), /角色无访问权限/)
})