// PremiumLLMService.chatStream 계약 회귀 테스트 (red-team r1-6) // - system 메시지 보존 (회의 전사 / 음성 대화 페르소나) // - 끊긴 스트림·오류 이벤트·빈 응답은 성공이 아니다 // - 프록시 HTTP 오류는 비스트리밍으로 재요청하지 않는다 (쿼터 이중 소비 방지) // - 호출 단위 취소 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { ErrorCode } from '@d3ro/core/errors' import { parseLlmRequest } from '../../../../../server/supabase/functions/_shared/llm-contract' const cloud = vi.hoisted(() => ({ isEnabled: (): boolean => true, isAuthenticated: (): boolean => true, invokeFunctionStream: vi.fn(), invokeFunction: vi.fn(), })) vi.mock('../../../src/main/services/CloudSyncService', () => ({ getCloudSyncService: () => cloud, })) import { getPremiumLLMService, resetPremiumLLMServiceForTests, proxyHttpStatus, } from '../../../src/main/services/PremiumLLMService' const encoder = new TextEncoder() interface SseFrame { event: string data: unknown } function sseText(frames: SseFrame[]): string { return frames.map((f) => `event: ${f.event}\ndata: ${JSON.stringify(f.data)}\n\n`).join('') } function sseStream(frames: SseFrame[]): ReadableStream { return new ReadableStream({ start(controller) { controller.enqueue(encoder.encode(sseText(frames))) controller.close() }, }) } const delta = (text: string): SseFrame => ({ event: 'content_block_delta', data: { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text } }, }) const messageStart: SseFrame = { event: 'message_start', data: { type: 'message_start', message: {} } } const messageStop: SseFrame = { event: 'message_stop', data: { type: 'message_stop' } } const overloaded: SseFrame = { event: 'error', data: { type: 'error', error: { type: 'overloaded_error', message: 'Overloaded' } }, } /** 신호가 abort 되기 전까지 끝나지 않는 SSE 스트림 */ function hangingStream(signal: AbortSignal): ReadableStream { return new ReadableStream({ start(controller) { controller.enqueue(encoder.encode(sseText([messageStart, delta('partial ')]))) signal.addEventListener('abort', () => { controller.error(new DOMException('Aborted', 'AbortError')) }, { once: true }) }, }) } async function collect(gen: AsyncGenerator): Promise<{ tokens: string[]; result: string }> { const tokens: string[] = [] for (let next = await gen.next(); ; next = await gen.next()) { if (next.done) return { tokens, result: next.value } tokens.push(next.value) } } function lastStreamBody(): Record { const call = cloud.invokeFunctionStream.mock.calls.at(-1) if (!call) throw new Error('invokeFunctionStream was not called') return call[1] as Record } describe('PremiumLLMService.chatStream contract', () => { beforeEach(() => { resetPremiumLLMServiceForTests() cloud.invokeFunctionStream.mockReset() cloud.invokeFunction.mockReset() }) afterEach(() => { resetPremiumLLMServiceForTests() }) it('sends role:system content as body.system instead of dropping it', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: sseStream([messageStart, delta('short answer.'), messageStop]), error: null, }) const { tokens, result } = await collect(getPremiumLLMService().chatStream([ { role: 'system', content: 'Keep answers brief (2-3 sentences).' }, { role: 'user', content: 'hello' }, ])) expect(tokens).toEqual(['short answer.']) expect(result).toBe('short answer.') const body = lastStreamBody() expect(body.system).toBe('Keep answers brief (2-3 sentences).') expect(body.messages).toEqual([{ role: 'user', content: 'hello' }]) expect(body.stream).toBe(true) }) it('keeps a meeting transcript system prompt within the llm-proxy contract', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: sseStream([delta('answer'), messageStop]), error: null, }) const transcript = '회의 발언 '.repeat(4_000) const systemPrompt = `당신은 회의 내용을 분석하는 AI 어시스턴트입니다.\n\n## 회의 전사\n${transcript}` const history = Array.from({ length: 60 }, (_, i) => ({ role: i % 2 === 0 ? 'user' : 'assistant', content: `turn ${i}`, })) await collect(getPremiumLLMService().chatStream([ { role: 'system', content: systemPrompt }, ...history, { role: 'user', content: '결정 사항은?' }, ])) const body = lastStreamBody() const validated = parseLlmRequest(body) expect(validated.system?.startsWith('당신은 회의 내용을 분석하는 AI 어시스턴트입니다.')).toBe(true) expect(validated.system).toContain('## 회의 전사') expect(validated.messages.at(-1)).toEqual({ role: 'user', content: '결정 사항은?' }) }) it('removes an empty assistant turn so a previous failure cannot poison later turns', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: sseStream([delta('ok'), messageStop]), error: null, }) await collect(getPremiumLLMService().chatStream([ { role: 'system', content: 'persona' }, { role: 'user', content: 'first' }, { role: 'assistant', content: '' }, { role: 'user', content: 'second' }, ])) const body = lastStreamBody() expect(() => parseLlmRequest(body)).not.toThrow() expect(body.messages).toEqual([ { role: 'user', content: 'first' }, { role: 'user', content: 'second' }, ]) }) it('rejects when the provider sends an error event mid-stream', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: sseStream([messageStart, delta('partial '), overloaded]), error: null, }) const tokens: string[] = [] await expect((async () => { for await (const token of getPremiumLLMService().chatStream([{ role: 'user', content: 'q' }])) { tokens.push(token) } })()).rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed }) expect(tokens).toEqual(['partial ']) }) it('rejects when the stream ends without message_stop', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: sseStream([messageStart, delta('cut off')]), error: null, }) await expect(collect(getPremiumLLMService().chatStream([{ role: 'user', content: 'q' }]))) .rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed }) }) it('rejects an empty completed reply instead of returning ""', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: sseStream([messageStart, messageStop]), error: null, }) await expect(collect(getPremiumLLMService().chatStream([{ role: 'user', content: 'q' }]))) .rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed }) }) it('does not resend a proxy 502 as a non-streaming request', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: null, error: { message: '502: {"error":"provider_request_failed"}' }, }) await expect(collect(getPremiumLLMService().chatStream([{ role: 'user', content: 'q' }]))) .rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed }) expect(cloud.invokeFunction).not.toHaveBeenCalled() }) it('reports quota_exceeded from the stream call without a second request', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: null, error: { message: '429: {"error":"quota_exceeded"}' }, }) const upgrade = vi.fn() const service = getPremiumLLMService() service.on('upgrade-required', upgrade) await expect(collect(service.chatStream([{ role: 'user', content: 'q' }]))).rejects.toBeTruthy() expect(upgrade).toHaveBeenCalledWith({ reason: 'quota_exceeded' }) expect(cloud.invokeFunction).not.toHaveBeenCalled() }) it('falls back to non-streaming only on transport failure, keeping system and max_tokens', async () => { cloud.invokeFunctionStream.mockResolvedValue({ stream: null, error: { message: 'fetch failed' } }) cloud.invokeFunction.mockResolvedValue({ data: { id: 'm', model: 'claude', role: 'assistant', content: [{ type: 'text', text: 'fallback answer' }], stop_reason: 'end_turn', usage: { input_tokens: 1, output_tokens: 1 }, }, error: null, }) const { tokens, result } = await collect(getPremiumLLMService().chatStream( [{ role: 'system', content: 'persona' }, { role: 'user', content: 'q' }], { maxTokens: 512 }, )) expect(tokens).toEqual(['fallback answer']) expect(result).toBe('fallback answer') const [name, body, options] = cloud.invokeFunction.mock.calls[0] as [string, Record, { signal?: AbortSignal }] expect(name).toBe('llm-proxy') expect(body).toMatchObject({ system: 'persona', stream: false, max_tokens: 512 }) expect(options?.signal).toBeInstanceOf(AbortSignal) }) it('cancelGeneration cancels every active call, not only the latest one', async () => { cloud.invokeFunctionStream.mockImplementation(async (_n: string, _b: unknown, signal: AbortSignal) => ({ stream: hangingStream(signal), error: null, })) const service = getPremiumLLMService() const first = service.chatStream([{ role: 'user', content: 'a' }]) const second = service.chatStream([{ role: 'user', content: 'b' }]) expect((await first.next()).value).toBe('partial ') expect((await second.next()).value).toBe('partial ') const firstRest = first.next() const secondRest = second.next() service.cancelGeneration() await expect(firstRest).rejects.toMatchObject({ code: ErrorCode.LLMProcessingCancelled }) await expect(secondRest).rejects.toMatchObject({ code: ErrorCode.LLMProcessingCancelled }) }) it('honors a caller signal and a per-call timeout', async () => { cloud.invokeFunctionStream.mockImplementation(async (_n: string, _b: unknown, signal: AbortSignal) => ({ stream: hangingStream(signal), error: null, })) const service = getPremiumLLMService() const external = new AbortController() const cancelled = service.chatStream([{ role: 'user', content: 'a' }], { signal: external.signal }) await cancelled.next() const pending = cancelled.next() external.abort() await expect(pending).rejects.toMatchObject({ code: ErrorCode.LLMProcessingCancelled }) const timed = service.chatStream([{ role: 'user', content: 'b' }], { timeoutMs: 20 }) await timed.next() await expect(timed.next()).rejects.toMatchObject({ code: ErrorCode.LLMProcessingTimeout }) }) it('aborts the underlying request when the consumer stops early', async () => { const signals: AbortSignal[] = [] cloud.invokeFunctionStream.mockImplementation(async (_n: string, _b: unknown, signal: AbortSignal) => { signals.push(signal) return { stream: hangingStream(signal), error: null } }) for await (const token of getPremiumLLMService().chatStream([{ role: 'user', content: 'q' }])) { expect(token).toBe('partial ') break } expect(signals[0].aborted).toBe(true) }) }) describe('proxyHttpStatus', () => { it('detects proxy HTTP responses and ignores transport errors', () => { expect(proxyHttpStatus('502: {"error":"provider_request_failed"}')).toBe(502) expect(proxyHttpStatus('429: {"error":"quota_exceeded"}')).toBe(429) expect(proxyHttpStatus('fetch failed')).toBeNull() expect(proxyHttpStatus('No active session — 로그인 필요')).toBeNull() }) })