diff --git a/apps/desktop/src/main/services/LocalLLMService.ts b/apps/desktop/src/main/services/LocalLLMService.ts index ced05f9..8293728 100644 --- a/apps/desktop/src/main/services/LocalLLMService.ts +++ b/apps/desktop/src/main/services/LocalLLMService.ts @@ -13,6 +13,13 @@ import type { LLMStatus, LLMModel, LLMAction, LLMConnectionState } from '@d3ro/c import { resolveSystemPrompt } from './llm-prompts' import { getBundledOllamaPath } from '../utils/paths' import { normalizeLoopbackUrl } from '../utils/loopback' +import { readNdjsonLines } from '../utils/ndjson-reader' +import { + toChatRequest, + toRoleMessages, + type ChatStreamOptions as CoreChatStreamOptions, + type RoleMessage, +} from '@d3ro/core/llm-chat' const logger = getLogger('LocalLLMService') @@ -73,15 +80,30 @@ interface OllamaGenerateResponse { eval_count?: number } -interface ChatStreamOptions { - model?: string - temperature?: number - maxTokens?: number - signal?: AbortSignal - timeoutMs?: number +/** @d3ro/core/llm-chat 공통 옵션 + Ollama 전용 keep_alive. */ +interface ChatStreamOptions extends CoreChatStreamOptions { keepAlive?: string } +interface OllamaChatFrame { + message?: { content: string } + done: boolean +} + +/** NDJSON 한 줄을 프레임 객체로 파싱한다. 깨진 줄은 스트림 전체 실패로 본다. */ +function parseOllamaFrame(line: string): T { + let parsed: unknown + try { + parsed = JSON.parse(line) + } catch { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Ollama returned malformed NDJSON') + } + if (typeof parsed !== 'object' || parsed === null) { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Ollama returned malformed NDJSON') + } + return parsed as T +} + type AbortCause = 'timeout' | 'cancelled' interface ActiveRequest { @@ -489,8 +511,6 @@ class LocalLLMService extends EventEmitter { } reader = response.body.getReader() - const decoder = new TextDecoder() - let buffer = '' let fullText = '' let lastChunk: OllamaGenerateResponse | null = null const complete = (): GenerateResult => { @@ -505,52 +525,18 @@ class LocalLLMService extends EventEmitter { return result } - while (true) { - const { done, value } = await reader.read() - if (done) break - - buffer += decoder.decode(value, { stream: true }) - const lines = buffer.split('\n') - buffer = lines.pop() ?? '' - - for (const line of lines) { - if (!line.trim()) continue - try { - const chunk = JSON.parse(line) as OllamaGenerateResponse - fullText += chunk.response - if (chunk.done) { - lastChunk = chunk - doneFrame = true - this.emit('token', { token: chunk.response, done: true }) - if (chunk.response) yield chunk.response - return complete() - } - this.emit('token', { token: chunk.response, done: chunk.done }) - yield chunk.response - } catch { - throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Ollama returned malformed NDJSON') - } - } - } - - buffer += decoder.decode() - const trailing = buffer.trim() - if (trailing) { - try { - const chunk = JSON.parse(trailing) as OllamaGenerateResponse - fullText += chunk.response - if (chunk.done) { - lastChunk = chunk - doneFrame = true - this.emit('token', { token: chunk.response, done: true }) - if (chunk.response) yield chunk.response - return complete() - } - this.emit('token', { token: chunk.response, done: chunk.done }) - yield chunk.response - } catch { - throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Ollama returned malformed NDJSON') + for await (const line of readNdjsonLines(reader)) { + const chunk = parseOllamaFrame(line) + fullText += chunk.response + if (chunk.done) { + lastChunk = chunk + doneFrame = true + this.emit('token', { token: chunk.response, done: true }) + if (chunk.response) yield chunk.response + return complete() } + this.emit('token', { token: chunk.response, done: chunk.done }) + yield chunk.response } if (!doneFrame) { @@ -845,7 +831,7 @@ class LocalLLMService extends EventEmitter { * 각 토큰마다 yield, 완료 시 전체 응답 텍스트를 return. */ async *chatStream( - messages: Array<{ role: string; content: string }>, + messages: RoleMessage[], options?: ChatStreamOptions, ): AsyncGenerator { if (!this._available) { @@ -865,7 +851,8 @@ class LocalLLMService extends EventEmitter { headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ model, - messages, + // @d3ro/core/llm-chat 계약: system 은 선두 system 메시지 하나로 정규화해 보낸다. + messages: toRoleMessages(toChatRequest(messages)), stream: true, keep_alive: options?.keepAlive, think: false, @@ -882,55 +869,19 @@ class LocalLLMService extends EventEmitter { } reader = response.body.getReader() - const decoder = new TextDecoder() - let buffer = '' let accumulated = '' - while (true) { - const { done, value } = await reader.read() - if (done) break - - buffer += decoder.decode(value, { stream: true }) - const lines = buffer.split('\n') - buffer = lines.pop() ?? '' - - for (const line of lines) { - if (!line.trim()) continue - try { - const chunk = JSON.parse(line) as { message?: { content: string }; done: boolean } - if (chunk.done) { - doneFrame = true - } - if (chunk.message?.content) { - accumulated += chunk.message.content - yield chunk.message.content - } - if (chunk.done) { - return accumulated - } - } catch { - throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Ollama returned malformed NDJSON') - } + for await (const line of readNdjsonLines(reader)) { + const chunk = parseOllamaFrame(line) + if (chunk.done) { + doneFrame = true } - } - - buffer += decoder.decode() - const trailing = buffer.trim() - if (trailing) { - try { - const chunk = JSON.parse(trailing) as { message?: { content: string }; done: boolean } - if (chunk.done) { - doneFrame = true - } - if (chunk.message?.content) { - accumulated += chunk.message.content - yield chunk.message.content - } - if (chunk.done) { - return accumulated - } - } catch { - throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Ollama returned malformed NDJSON') + if (chunk.message?.content) { + accumulated += chunk.message.content + yield chunk.message.content + } + if (chunk.done) { + return accumulated } } diff --git a/apps/desktop/src/main/services/PremiumLLMService.ts b/apps/desktop/src/main/services/PremiumLLMService.ts index fb2e726..2d52e32 100644 --- a/apps/desktop/src/main/services/PremiumLLMService.ts +++ b/apps/desktop/src/main/services/PremiumLLMService.ts @@ -15,7 +15,16 @@ import { getCloudSyncService } from './CloudSyncService' import { D3ROError, ErrorCode } from '@d3ro/core/errors' import type { LLMAction } from '@d3ro/core/types' import { resolveSystemPrompt } from './llm-prompts' -import { parseAnthropicSSE } from '../utils/sse-parser' +import { parseAnthropicSSE, AnthropicStreamError } from '../utils/sse-parser' +import { + fitChatRequest, + LLM_PROXY_CHAT_LIMITS, + toChatRequest, + type ChatRequest, + type ChatStreamOptions, + type ChatTurn, + type RoleMessage, +} from '@d3ro/core/llm-chat' const logger = getLogger('PremiumLLMService') @@ -23,15 +32,9 @@ const logger = getLogger('PremiumLLMService') // 내부 타입 // ============================================================ -/** Ollama-style 메시지 → Claude Messages 변환용 */ -interface ChatMessage { - role: 'user' | 'assistant' - content: string -} - /** llm-proxy Edge Function 요청 body */ interface LlmProxyRequest { - messages: ChatMessage[] + messages: ChatTurn[] system?: string max_tokens?: number model?: string @@ -55,9 +58,105 @@ export interface QuotaUsageSnapshot { overageCredits: number } +type UpgradeReason = 'quota_exceeded' | 'model_not_allowed' | 'auth_required' + +/** 호출 단위 취소·기한 상태. cancelGeneration() 은 활성 호출 전부를 취소한다. */ +type AbortCause = 'timeout' | 'cancelled' + +interface PremiumCall { + controller: AbortController + abortCause: AbortCause | null + abort: (cause: AbortCause) => void + close: () => void +} + +/** llm-proxy 출력 토큰 상한 (llm-contract MAX_OUTPUT_TOKENS) */ +const PROXY_MAX_OUTPUT_TOKENS = 4096 +const DEFAULT_CHAT_MAX_TOKENS = 2048 +/** 클라이언트 측 채팅 기한. 프록시가 공급자 호출에 45초 기한을 두므로 여유 있게 잡는다. */ +const DEFAULT_CHAT_TIMEOUT_MS = 120_000 + +/** + * llm-proxy 오류 메시지를 D3ROError 로 분류한다 (순수 함수). + * upgradeReason 이 있으면 호출자가 'upgrade-required' 를 emit 한다. + */ +export function classifyProxyError(message: string): { error: D3ROError; upgradeReason: UpgradeReason | null } { + if (message.includes('401') || message.includes('Unauthorized') || message.includes('auth')) { + return { + error: new D3ROError(ErrorCode.LLMServerUnreachable, `인증 실패: ${message}`), + upgradeReason: 'auth_required', + } + } + if (message.includes('quota_exceeded') || message.includes('429')) { + return { + error: new D3ROError(ErrorCode.LLMProcessingFailed, `쿼터 초과: ${message}`), + upgradeReason: 'quota_exceeded', + } + } + if (message.includes('model_not_allowed') || message.includes('403')) { + return { + error: new D3ROError(ErrorCode.LLMInvalidAction, `모델 권한 없음: ${message}`), + upgradeReason: 'model_not_allowed', + } + } + return { + error: new D3ROError(ErrorCode.LLMProcessingFailed, `llm-proxy: ${message}`), + upgradeReason: null, + } +} + +/** + * invokeFunctionStream 오류가 프록시의 HTTP 응답(`: `)인지 판별한다. + * 프록시가 응답했다면 같은 요청을 비스트리밍으로 다시 보내도 같은 실패(와 쿼터 소비)만 + * 되풀이되므로, 비스트리밍 폴백은 HTTP 상태가 없는 전송 계층 실패에서만 쓴다. + */ +export function proxyHttpStatus(message: string): number | null { + const match = /^(\d{3}):/.exec(message) + return match ? Number(match[1]) : null +} + +function resolveMaxTokens(maxTokens: number | undefined): number { + if (maxTokens === undefined) return DEFAULT_CHAT_MAX_TOKENS + if (!Number.isSafeInteger(maxTokens) || maxTokens <= 0) { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'maxTokens must be a positive safe integer') + } + return Math.min(maxTokens, PROXY_MAX_OUTPUT_TOKENS) +} + +function resolveTimeoutMs(timeoutMs: number | undefined): number { + if (timeoutMs === undefined) return DEFAULT_CHAT_TIMEOUT_MS + if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'timeoutMs must be a positive safe integer') + } + return timeoutMs +} + +/** ChatRequest → llm-proxy body. system 은 최상위 system 필드로 보낸다. */ +function toProxyBody( + request: ChatRequest, + maxTokens: number, + model: string | undefined, + stream: boolean, +): LlmProxyRequest { + const body: LlmProxyRequest = { + messages: request.turns, + max_tokens: maxTokens, + stream, + } + if (request.system !== undefined) body.system = request.system + if (model !== undefined) body.model = model + return body +} + +function firstText(response: ClaudeMessageResponse): string { + const firstBlock = response.content?.[0] + return firstBlock?.type === 'text' ? firstBlock.text : '' +} + + interface PremiumLLMEvents { 'quota-warning': (payload: { current: number; limit: number; overageCredits: number }) => void - 'upgrade-required': (payload: { reason: 'quota_exceeded' | 'model_not_allowed' | 'auth_required' }) => void + 'upgrade-required': (payload: { reason: UpgradeReason }) => void 'fallback-triggered': (payload: { reason: string }) => void } @@ -66,7 +165,7 @@ interface PremiumLLMEvents { // ============================================================ class PremiumLLMService extends EventEmitter { - private _abortController: AbortController | null = null + private _activeCalls = new Set() private _disposed = false private _lastQuota: QuotaUsageSnapshot | null = null @@ -145,64 +244,78 @@ class PremiumLLMService extends EventEmitter { } /** - * 스트리밍 대화 (Voice Conversation용). + * 스트리밍 대화 (Voice Conversation / 회의 채팅용). * SSE 스트리밍: llm-proxy에 stream=true로 요청, Anthropic SSE를 토큰 단위 yield. + * + * @d3ro/core/llm-chat 계약을 따른다: + * - role:'system' 메시지는 버리지 않고 body.system 으로 보낸다 (한도 초과 시 앞부분 보존). + * - message_stop 없이 끊긴 스트림, 공급자 오류 이벤트, 빈 응답은 D3ROError 로 throw 한다. + * - signal / timeoutMs / maxTokens 는 호출 단위로 적용된다. temperature 는 프록시가 받지 않는다. + * - 비스트리밍 폴백은 전송 계층 실패에서만 쓴다 (프록시 HTTP 오류는 그대로 전파). */ async *chatStream( - messages: Array<{ role: string; content: string }>, - options?: { model?: string; temperature?: number }, + messages: RoleMessage[], + options?: ChatStreamOptions, ): AsyncGenerator { this._ensureAuth() - const claudeMessages: ChatMessage[] = messages - .filter((m) => m.role === 'user' || m.role === 'assistant') - .map((m) => ({ role: m.role as 'user' | 'assistant', content: m.content })) - - const body: LlmProxyRequest = { - messages: claudeMessages, - model: options?.model, - max_tokens: 2048, - stream: true, + const request = fitChatRequest(toChatRequest(messages), LLM_PROXY_CHAT_LIMITS) + if (request.turns.length === 0) { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Premium chat requires a user message') } + const body = toProxyBody(request, resolveMaxTokens(options?.maxTokens), options?.model, true) + const call = this._beginCall(options?.signal, resolveTimeoutMs(options?.timeoutMs)) + let completed = false - this._abortController = new AbortController() - const cloud = getCloudSyncService() - - const { stream, error } = await cloud.invokeFunctionStream( - 'llm-proxy', - body as unknown as Record, - this._abortController.signal, - ) - - if (error || !stream) { - const msg = error?.message ?? 'Stream unavailable' - logger.error(`SSE stream failed: ${msg}`) - - // SSE 실패 시 비스트리밍 fallback - logger.info('Falling back to non-streaming Premium LLM') - const fallbackBody = { ...body, stream: false } - const response = await this._invokeProxy(fallbackBody) - const firstBlock = response.content?.[0] - const text = firstBlock?.type === 'text' ? firstBlock.text : '' - if (text.length > 0) yield text - return text - } - - let accumulated = '' try { + const cloud = getCloudSyncService() + const { stream, error } = await cloud.invokeFunctionStream( + 'llm-proxy', + body as unknown as Record, + call.controller.signal, + ) + + if (error || !stream) { + const msg = error?.message ?? 'Stream unavailable' + if (call.controller.signal.aborted) { + throw new D3ROError(ErrorCode.LLMProcessingCancelled, msg) + } + logger.error(`SSE stream failed: ${msg}`) + + if (proxyHttpStatus(msg) !== null) { + // 프록시가 응답한 오류 — 재요청은 같은 실패와 쿼터 소비만 되풀이한다. + throw this._rejectProxyError(msg) + } + + // 전송 계층 실패 시 비스트리밍 fallback (같은 system/maxTokens/signal 적용) + logger.info('Falling back to non-streaming Premium LLM') + const response = await this._invokeProxy({ ...body, stream: false }, call.controller.signal) + const text = firstText(response) + if (!text.trim()) { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Premium LLM returned empty content') + } + completed = true + yield text + return text + } + + let accumulated = '' for await (const token of parseAnthropicSSE(stream)) { accumulated += token yield token } - } catch (err) { - if ((err as Error).name !== 'AbortError') { - logger.warn(`SSE parse error: ${err instanceof Error ? err.message : String(err)}`) + completed = true + if (!accumulated.trim()) { + throw new D3ROError(ErrorCode.LLMProcessingFailed, 'Premium LLM returned empty content') } + return accumulated + } catch (err) { + throw this._toChatError(err, call) } finally { - this._abortController = null + // 완료 전 종료(소비자 break, 오류) 시 연결을 끊어 프록시 스트림을 정리한다. + if (!completed) call.abort('cancelled') + call.close() } - - return accumulated } /** @@ -227,11 +340,11 @@ class PremiumLLMService extends EventEmitter { } cancelGeneration(): void { - if (this._abortController) { - this._abortController.abort() - this._abortController = null - logger.info('Premium LLM generation cancelled') + if (this._activeCalls.size === 0) return + for (const call of [...this._activeCalls]) { + call.abort('cancelled') } + logger.info('Premium LLM generation cancelled') } dispose(): void { @@ -243,32 +356,19 @@ class PremiumLLMService extends EventEmitter { // ── 내부: Edge Function 호출 ───────────────────────────── - private async _invokeProxy(body: LlmProxyRequest): Promise { + private async _invokeProxy(body: LlmProxyRequest, signal?: AbortSignal): Promise { const cloud = getCloudSyncService() // Supabase JS 클라이언트의 functions.invoke() 사용 — auth 헤더를 올바르게 처리. // raw fetch + Authorization: Bearer 방식은 Supabase gateway가 401로 거부. - const { data, error } = await cloud.invokeFunction('llm-proxy', body as unknown as Record) + const { data, error } = signal + ? await cloud.invokeFunction('llm-proxy', body as unknown as Record, { signal }) + : await cloud.invokeFunction('llm-proxy', body as unknown as Record) if (error) { const msg = error.message ?? 'Edge Function error' logger.error(`llm-proxy error: ${msg}`) - - // 에러 메시지 기반 분류 - if (msg.includes('401') || msg.includes('Unauthorized') || msg.includes('auth')) { - this.emit('upgrade-required', { reason: 'auth_required' }) - throw new D3ROError(ErrorCode.LLMServerUnreachable, `인증 실패: ${msg}`) - } - if (msg.includes('quota_exceeded') || msg.includes('429')) { - this.emit('upgrade-required', { reason: 'quota_exceeded' }) - throw new D3ROError(ErrorCode.LLMProcessingFailed, `쿼터 초과: ${msg}`) - } - if (msg.includes('model_not_allowed') || msg.includes('403')) { - this.emit('upgrade-required', { reason: 'model_not_allowed' }) - throw new D3ROError(ErrorCode.LLMInvalidAction, `모델 권한 없음: ${msg}`) - } - - throw new D3ROError(ErrorCode.LLMProcessingFailed, `llm-proxy: ${msg}`) + throw this._rejectProxyError(msg) } // functions.invoke는 response body를 자동 파싱해서 data에 넣음 @@ -281,6 +381,60 @@ class PremiumLLMService extends EventEmitter { return result } + /** 프록시 오류를 분류하고 필요 시 upgrade-required 를 알린다. */ + private _rejectProxyError(message: string): D3ROError { + const { error, upgradeReason } = classifyProxyError(message) + if (upgradeReason) this.emit('upgrade-required', { reason: upgradeReason }) + return error + } + + /** 호출 단위 AbortController 를 만들고 외부 signal·기한·cancelGeneration 에 연결한다. */ + private _beginCall(externalSignal: AbortSignal | undefined, timeoutMs: number): PremiumCall { + const controller = new AbortController() + const onExternalAbort = (): void => call.abort('cancelled') + const timer = setTimeout(() => call.abort('timeout'), timeoutMs) + const call: PremiumCall = { + controller, + abortCause: null, + abort: (cause) => { + if (call.abortCause !== null) return + call.abortCause = cause + controller.abort() + }, + close: () => { + clearTimeout(timer) + externalSignal?.removeEventListener('abort', onExternalAbort) + this._activeCalls.delete(call) + }, + } + this._activeCalls.add(call) + if (externalSignal?.aborted) { + onExternalAbort() + } else { + externalSignal?.addEventListener('abort', onExternalAbort, { once: true }) + } + return call + } + + /** 채팅 스트림 실패를 호출 단위 원인에 맞는 D3ROError 로 정규화한다. */ + private _toChatError(error: unknown, call: PremiumCall): D3ROError { + if (call.abortCause === 'timeout') { + return new D3ROError(ErrorCode.LLMProcessingTimeout, 'Premium LLM generation timed out') + } + if (call.abortCause === 'cancelled' || call.controller.signal.aborted) { + return new D3ROError(ErrorCode.LLMProcessingCancelled, 'Premium LLM generation cancelled') + } + if (error instanceof D3ROError) return error + if (error instanceof AnthropicStreamError) { + logger.warn(`Premium SSE stream failed (${error.kind}): ${error.message}`) + return new D3ROError(ErrorCode.LLMProcessingFailed, `Premium LLM stream failed: ${error.message}`) + } + return new D3ROError( + ErrorCode.LLMProcessingFailed, + `Premium LLM failed: ${error instanceof Error ? error.message : String(error)}`, + ) + } + // ── EventEmitter 타입 오버라이드 ─────────────────────── on(event: K, listener: PremiumLLMEvents[K]): this { diff --git a/apps/desktop/src/main/utils/ndjson-reader.ts b/apps/desktop/src/main/utils/ndjson-reader.ts new file mode 100644 index 0000000..2fc5ae3 --- /dev/null +++ b/apps/desktop/src/main/utils/ndjson-reader.ts @@ -0,0 +1,33 @@ +// src/main/utils/ndjson-reader.ts +// NDJSON 스트림을 줄 단위로 읽는다 (버퍼링·청크 경계·마지막 개행 없는 줄 처리만 담당). +// 줄 파싱 정책(깨진 줄 처리, 완료 프레임 요구)은 호출자가 정한다. + +/** + * reader 에서 비어 있지 않은 NDJSON 줄을 차례로 yield 한다. + * 개행 없이 끝난 마지막 줄은 trim 해서 yield 한다. + * reader 의 해제·취소는 호출자 책임이다. + */ +export async function* readNdjsonLines( + reader: ReadableStreamDefaultReader, +): AsyncGenerator { + const decoder = new TextDecoder() + let buffer = '' + + while (true) { + const { done, value } = await reader.read() + if (done) break + + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() ?? '' + + for (const line of lines) { + if (!line.trim()) continue + yield line + } + } + + buffer += decoder.decode() + const trailing = buffer.trim() + if (trailing) yield trailing +} diff --git a/apps/desktop/src/main/utils/sse-parser.ts b/apps/desktop/src/main/utils/sse-parser.ts index dab2eff..f64b69c 100644 --- a/apps/desktop/src/main/utils/sse-parser.ts +++ b/apps/desktop/src/main/utils/sse-parser.ts @@ -1,6 +1,12 @@ // src/main/utils/sse-parser.ts // Anthropic Claude Messages API SSE 스트림 파서 -// content_block_delta 이벤트에서 텍스트 토큰을 추출하는 AsyncGenerator +// content_block_delta 이벤트에서 텍스트 토큰을 추출하는 AsyncGenerator. +// +// 종료 계약 (@d3ro/core/llm-chat 참조): +// - message_stop 또는 data: [DONE] 을 받아야 정상 종료한다. +// - `error` 이벤트(예: overloaded_error)는 AnthropicStreamError('provider_error')로 throw. +// - 종료 프레임 없이 스트림이 끝나면 AnthropicStreamError('incomplete')로 throw. +// (프록시 타임아웃 등으로 잘린 응답을 완료로 오인하지 않기 위함) interface ContentBlockDelta { type: 'content_block_delta' @@ -10,15 +16,80 @@ interface ContentBlockDelta { } } +interface AnthropicErrorEvent { + type: 'error' + error?: { type?: string; message?: string } +} + interface SSEEvent { type: string [key: string]: unknown } +export type AnthropicStreamErrorKind = 'provider_error' | 'incomplete' + +/** SSE 스트림이 정상 완료되지 못했음을 나타낸다. */ +export class AnthropicStreamError extends Error { + readonly kind: AnthropicStreamErrorKind + readonly providerErrorType: string | null + + constructor(kind: AnthropicStreamErrorKind, message: string, providerErrorType: string | null = null) { + super(message) + this.name = 'AnthropicStreamError' + this.kind = kind + this.providerErrorType = providerErrorType + } +} + +type LineOutcome = + | { kind: 'skip' } + | { kind: 'token'; text: string } + | { kind: 'stop' } + +/** SSE 한 줄을 해석한다. 오류 이벤트는 throw 한다. */ +function interpretLine(line: string): LineOutcome { + const trimmed = line.trim() + + // 빈 줄, 이벤트 타입 라인 (event:), 주석(:) 건너뜀 — 타입은 data 의 type 필드로 판별 + if (!trimmed || !trimmed.startsWith('data:')) return { kind: 'skip' } + + const payload = trimmed.slice(5).trimStart() + + // "data: [DONE]" — 종료 시그널 + if (payload === '[DONE]') return { kind: 'stop' } + + let evt: SSEEvent + try { + evt = JSON.parse(payload) as SSEEvent + } catch { + // JSON 파싱 실패 — 건너뜀 + return { kind: 'skip' } + } + + if (evt.type === 'message_stop') return { kind: 'stop' } + + if (evt.type === 'error') { + const errorEvent = evt as unknown as AnthropicErrorEvent + const providerType = errorEvent.error?.type ?? 'unknown_error' + const detail = errorEvent.error?.message ?? 'Provider stream error' + throw new AnthropicStreamError('provider_error', `${providerType}: ${detail}`, providerType) + } + + if (evt.type === 'content_block_delta') { + const delta = evt as unknown as ContentBlockDelta + if (delta.delta?.type === 'text_delta' && delta.delta.text) { + return { kind: 'token', text: delta.delta.text } + } + } + // 그 외 이벤트 (message_start, content_block_start, ping 등)는 건너뜀 + return { kind: 'skip' } +} + /** * ReadableStream을 파싱하여 텍스트 토큰을 yield. - * Anthropic SSE 형식: "data: {json}\n\n" 라인 단위. - * content_block_delta.delta.text 추출, message_stop 또는 [DONE] 시 종료. + * Anthropic SSE 형식: "event: x\ndata: {json}\n\n" 라인 단위. + * content_block_delta.delta.text 추출, message_stop 또는 [DONE] 시 정상 종료. + * 오류 이벤트나 종료 프레임 없는 EOF 는 AnthropicStreamError 로 throw. */ export async function* parseAnthropicSSE( stream: ReadableStream, @@ -40,37 +111,19 @@ export async function* parseAnthropicSSE( buffer = lines.pop() ?? '' for (const line of lines) { - const trimmed = line.trim() - - // 빈 줄 또는 이벤트 타입 라인 (event:) 건너뜀 - if (!trimmed || trimmed.startsWith('event:')) continue - - // "data: [DONE]" — 종료 시그널 - if (trimmed === 'data: [DONE]') return - - // "data: {...}" — JSON 파싱 - if (trimmed.startsWith('data: ')) { - const json = trimmed.substring(6) - try { - const evt = JSON.parse(json) as SSEEvent - - // message_stop → 스트림 종료 - if (evt.type === 'message_stop') return - - // content_block_delta → 텍스트 토큰 yield - if (evt.type === 'content_block_delta') { - const delta = evt as unknown as ContentBlockDelta - if (delta.delta?.type === 'text_delta' && delta.delta.text) { - yield delta.delta.text - } - } - // 그 외 이벤트 (message_start, content_block_start 등)는 건너뜀 - } catch { - // JSON 파싱 실패 — 건너뜀 - } - } + const outcome = interpretLine(line) + if (outcome.kind === 'stop') return + if (outcome.kind === 'token') yield outcome.text } } + + // 개행 없이 끝난 마지막 줄 처리 + buffer += decoder.decode() + const outcome = interpretLine(buffer) + if (outcome.kind === 'stop') return + if (outcome.kind === 'token') yield outcome.text + + throw new AnthropicStreamError('incomplete', 'Anthropic stream ended before message_stop') } finally { reader.releaseLock() } diff --git a/apps/desktop/tests/main/services/local-llm-chat-contract-r1-6.test.ts b/apps/desktop/tests/main/services/local-llm-chat-contract-r1-6.test.ts new file mode 100644 index 0000000..d1f3217 --- /dev/null +++ b/apps/desktop/tests/main/services/local-llm-chat-contract-r1-6.test.ts @@ -0,0 +1,100 @@ +// LocalLLMService.chatStream 이 @d3ro/core/llm-chat 계약을 따르는지 (red-team r1-6) +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { ErrorCode } from '@d3ro/core/errors' +import { initInMemoryConfig, resetInMemoryConfig } from '../../../src/main/services/ConfigService' +import { + getLocalLLMService, + resetLocalLLMServiceForTests, +} from '../../../src/main/services/LocalLLMService' + +const encoder = new TextEncoder() + +function ndjsonResponse(parts: string[]): Response { + return new Response(new ReadableStream({ + start(controller) { + for (const part of parts) controller.enqueue(encoder.encode(part)) + controller.close() + }, + }), { status: 200 }) +} + +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) + } +} + +describe('LocalLLMService chat contract', () => { + beforeEach(() => { + initInMemoryConfig() + resetLocalLLMServiceForTests() + ;(getLocalLLMService() as unknown as { _available: boolean })._available = true + }) + + afterEach(() => { + resetLocalLLMServiceForTests() + resetInMemoryConfig() + vi.unstubAllGlobals() + }) + + it('sends the system prompt as a single leading system message', async () => { + let sent: Array<{ role: string; content: string }> = [] + vi.stubGlobal('fetch', vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + sent = (JSON.parse(String(init?.body)) as { messages: typeof sent }).messages + return ndjsonResponse([ + `${JSON.stringify({ message: { content: 'hi' }, done: false })}\n`, + JSON.stringify({ message: { content: '' }, done: true }), + ]) + })) + + const { tokens, result } = await collect(getLocalLLMService().chatStream([ + { role: 'system', content: 'persona' }, + { role: 'user', content: 'hello' }, + ])) + + expect(tokens).toEqual(['hi']) + expect(result).toBe('hi') + expect(sent).toEqual([ + { role: 'system', content: 'persona' }, + { role: 'user', content: 'hello' }, + ]) + }) + + it('rejects a stream that ends without a done frame', async () => { + vi.stubGlobal('fetch', vi.fn(async () => ndjsonResponse([ + `${JSON.stringify({ message: { content: 'partial' }, done: false })}\n`, + ]))) + + await expect(collect(getLocalLLMService().chatStream([{ role: 'user', content: 'q' }]))) + .rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed }) + }) + + it('rejects malformed NDJSON frames, including non-object JSON', async () => { + vi.stubGlobal('fetch', vi.fn(async () => ndjsonResponse(['{broken\n']))) + await expect(collect(getLocalLLMService().chatStream([{ role: 'user', content: 'q' }]))) + .rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed, message: 'Ollama returned malformed NDJSON' }) + + vi.stubGlobal('fetch', vi.fn(async () => ndjsonResponse(['null\n']))) + await expect(collect(getLocalLLMService().chatStream([{ role: 'user', content: 'q' }]))) + .rejects.toMatchObject({ code: ErrorCode.LLMProcessingFailed, message: 'Ollama returned malformed NDJSON' }) + }) + + it('streamGenerate still completes on a trailing done frame without newline', async () => { + vi.stubGlobal('fetch', vi.fn(async () => ndjsonResponse([ + `${JSON.stringify({ model: 'm', response: 'a', done: false })}\n`, + JSON.stringify({ model: 'm', response: 'b', done: true, eval_count: 2 }), + ]))) + + const gen = getLocalLLMService().streamGenerate('p') + const tokens: string[] = [] + let next = await gen.next() + while (!next.done) { + tokens.push(next.value) + next = await gen.next() + } + expect(tokens).toEqual(['a', 'b']) + expect(next.value).toMatchObject({ text: 'ab', completionTokens: 2 }) + }) +}) diff --git a/apps/desktop/tests/main/services/premium-llm-redteam-r1-6.test.ts b/apps/desktop/tests/main/services/premium-llm-redteam-r1-6.test.ts new file mode 100644 index 0000000..9141d1c --- /dev/null +++ b/apps/desktop/tests/main/services/premium-llm-redteam-r1-6.test.ts @@ -0,0 +1,304 @@ +// 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() + }) +}) diff --git a/apps/desktop/tests/main/utils/llm-stream-parsers-r1-6.test.ts b/apps/desktop/tests/main/utils/llm-stream-parsers-r1-6.test.ts new file mode 100644 index 0000000..5f850b7 --- /dev/null +++ b/apps/desktop/tests/main/utils/llm-stream-parsers-r1-6.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it } from 'vitest' +import { AnthropicStreamError, parseAnthropicSSE } from '../../../src/main/utils/sse-parser' +import { readNdjsonLines } from '../../../src/main/utils/ndjson-reader' + +const encoder = new TextEncoder() + +function chunked(parts: string[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const part of parts) controller.enqueue(encoder.encode(part)) + controller.close() + }, + }) +} + +async function drain(stream: ReadableStream): Promise { + const out: string[] = [] + for await (const token of parseAnthropicSSE(stream)) out.push(token) + return out +} + +const deltaLine = (text: string): string => + `data: ${JSON.stringify({ type: 'content_block_delta', delta: { type: 'text_delta', text } })}\n` + +describe('parseAnthropicSSE', () => { + it('yields text deltas split across chunk boundaries and stops at message_stop', async () => { + const body = `event: content_block_delta\n${deltaLine('Hel')}\n${deltaLine('lo')}\ndata: {"type":"message_stop"}\n\n` + const parts = [body.slice(0, 17), body.slice(17, 60), body.slice(60)] + expect(await drain(chunked(parts))).toEqual(['Hel', 'lo']) + }) + + it('accepts [DONE], data: without a space, and a final line without newline', async () => { + expect(await drain(chunked([deltaLine('a'), 'data: [DONE]\n']))).toEqual(['a']) + expect(await drain(chunked([ + `data:${JSON.stringify({ type: 'content_block_delta', delta: { type: 'text_delta', text: 'b' } })}\n`, + 'data: {"type":"message_stop"}', + ]))).toEqual(['b']) + }) + + it('skips malformed JSON lines and ping events', async () => { + expect(await drain(chunked([ + 'data: {not json}\n', + 'event: ping\ndata: {"type":"ping"}\n', + deltaLine('x'), + 'data: {"type":"message_stop"}\n', + ]))).toEqual(['x']) + }) + + it('throws a provider_error on an error event', async () => { + const stream = chunked([ + deltaLine('partial'), + 'event: error\ndata: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}\n\n', + ]) + const tokens: string[] = [] + const error = await (async () => { + try { + for await (const token of parseAnthropicSSE(stream)) tokens.push(token) + return null + } catch (err) { + return err + } + })() + expect(tokens).toEqual(['partial']) + expect(error).toBeInstanceOf(AnthropicStreamError) + expect((error as AnthropicStreamError).kind).toBe('provider_error') + expect((error as AnthropicStreamError).providerErrorType).toBe('overloaded_error') + }) + + it('throws incomplete when the stream ends before message_stop', async () => { + await expect(drain(chunked([deltaLine('cut')]))).rejects.toMatchObject({ + name: 'AnthropicStreamError', + kind: 'incomplete', + }) + await expect(drain(chunked([]))).rejects.toMatchObject({ kind: 'incomplete' }) + }) +}) + +describe('readNdjsonLines', () => { + it('reassembles lines across chunks, skips blanks and yields a trailing line', async () => { + const reader = chunked(['{"a":', '1}\n\n \n{"b"', ':2}\n{"c":3}']).getReader() + const lines: string[] = [] + for await (const line of readNdjsonLines(reader)) lines.push(line) + expect(lines).toEqual(['{"a":1}', '{"b":2}', '{"c":3}']) + }) + + it('decodes multi-byte characters split across chunks', async () => { + const bytes = encoder.encode('{"t":"한글"}\n') + const reader = new ReadableStream({ + start(controller) { + controller.enqueue(bytes.slice(0, 8)) + controller.enqueue(bytes.slice(8)) + controller.close() + }, + }).getReader() + const lines: string[] = [] + for await (const line of readNdjsonLines(reader)) lines.push(line) + expect(lines).toEqual(['{"t":"한글"}']) + }) +}) diff --git a/packages/core/__tests__/llm-chat.test.ts b/packages/core/__tests__/llm-chat.test.ts new file mode 100644 index 0000000..eb2e612 --- /dev/null +++ b/packages/core/__tests__/llm-chat.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from 'vitest' +import { + CHAT_TRUNCATION_MARKER, + fitChatRequest, + LLM_PROXY_CHAT_LIMITS, + toChatRequest, + toRoleMessages, +} from '../src/llm-chat' + +describe('toChatRequest', () => { + it('separates system messages from turns and keeps turn order', () => { + expect(toChatRequest([ + { role: 'system', content: 'persona' }, + { role: 'user', content: 'hi' }, + { role: 'assistant', content: 'hello' }, + { role: 'user', content: 'again' }, + ])).toEqual({ + system: 'persona', + turns: [ + { role: 'user', content: 'hi' }, + { role: 'assistant', content: 'hello' }, + { role: 'user', content: 'again' }, + ], + }) + }) + + it('joins several system messages and drops non-chat roles', () => { + expect(toChatRequest([ + { role: 'system', content: 'a' }, + { role: 'tool', content: 'ignored' }, + { role: 'system', content: 'b' }, + { role: 'system', content: ' ' }, + { role: 'user', content: 'q' }, + ])).toEqual({ system: 'a\n\nb', turns: [{ role: 'user', content: 'q' }] }) + }) + + it('omits system when there is none', () => { + expect(toChatRequest([{ role: 'user', content: 'q' }])).toEqual({ + turns: [{ role: 'user', content: 'q' }], + }) + }) +}) + +describe('toRoleMessages', () => { + it('round-trips a canonical Ollama message list unchanged', () => { + const messages = [ + { role: 'system', content: 'persona' }, + { role: 'user', content: 'hi' }, + { role: 'assistant', content: 'hello' }, + ] + expect(toRoleMessages(toChatRequest(messages))).toEqual(messages) + }) + + it('puts the system message first', () => { + expect(toRoleMessages({ system: 's', turns: [{ role: 'user', content: 'u' }] })).toEqual([ + { role: 'system', content: 's' }, + { role: 'user', content: 'u' }, + ]) + }) +}) + +describe('fitChatRequest', () => { + const limits = { maxSystemChars: 50, maxMessageChars: 20, maxMessages: 3, maxTotalChars: 40 } + + it('keeps the head of an over-long system prompt within the limit', () => { + const system = `INSTRUCTIONS ${'x'.repeat(200)}` + const fitted = fitChatRequest({ system, turns: [{ role: 'user', content: 'q' }] }, limits) + expect(fitted.system?.length).toBeLessThanOrEqual(limits.maxSystemChars) + expect(fitted.system?.startsWith('INSTRUCTIONS')).toBe(true) + expect(fitted.system?.endsWith(CHAT_TRUNCATION_MARKER)).toBe(true) + }) + + it('drops empty turns such as an empty assistant reply', () => { + const fitted = fitChatRequest({ + turns: [ + { role: 'user', content: 'q1' }, + { role: 'assistant', content: '' }, + { role: 'user', content: 'q2' }, + ], + }, limits) + expect(fitted.turns).toEqual([ + { role: 'user', content: 'q1' }, + { role: 'user', content: 'q2' }, + ]) + }) + + it('keeps the newest turns within count and size limits and never starts with assistant', () => { + const fitted = fitChatRequest({ + turns: [ + { role: 'user', content: 'u1' }, + { role: 'assistant', content: 'a1' }, + { role: 'user', content: 'u2' }, + { role: 'assistant', content: 'a2' }, + { role: 'user', content: 'u3' }, + ], + }, limits) + // maxMessages=3 keeps [u2, a2, u3] + expect(fitted.turns.map((t) => t.content)).toEqual(['u2', 'a2', 'u3']) + + const trimmed = fitChatRequest({ + turns: [ + { role: 'user', content: 'u'.repeat(20) }, + { role: 'assistant', content: 'a'.repeat(15) }, + { role: 'user', content: 'q'.repeat(20) }, + ], + }, limits) + // total 40: [a(15), q(20)] fits, then the leading assistant turn is dropped + expect(trimmed.turns).toEqual([{ role: 'user', content: 'q'.repeat(20) }]) + }) + + it('truncates a single turn over the per-message limit', () => { + const fitted = fitChatRequest({ turns: [{ role: 'user', content: 'z'.repeat(100) }] }, limits) + expect(fitted.turns[0].content.length).toBeLessThanOrEqual(limits.maxMessageChars) + }) + + it('exposes the llm-proxy limits', () => { + expect(LLM_PROXY_CHAT_LIMITS).toEqual({ + maxSystemChars: 8_000, + maxMessageChars: 8_000, + maxMessages: 40, + maxTotalChars: 50_000, + }) + }) +}) diff --git a/packages/core/src/llm-chat.ts b/packages/core/src/llm-chat.ts new file mode 100644 index 0000000..23c2aad --- /dev/null +++ b/packages/core/src/llm-chat.ts @@ -0,0 +1,144 @@ +// packages/core/src/llm-chat.ts +// 채팅 LLM 요청 계약 — 로컬(Ollama)·프리미엄(llm-proxy) 어댑터가 같은 의미로 따르는 모델. +// +// 계약: +// 1. system 지시문은 대화 턴과 분리된다. 호출자가 Ollama 스타일로 넘긴 +// `{ role: 'system' }` 메시지는 어댑터가 버리지 않고 반드시 system 으로 전달한다. +// 2. 스트림은 공급자가 종료 프레임(Ollama `done`, Anthropic `message_stop`)을 보낸 +// 경우에만 정상 종료한다. 그 전에 끊기거나 공급자가 오류 프레임을 보내면 throw 한다. +// 부분 응답을 완료로 돌려주지 않는다. +// 3. signal / timeoutMs / maxTokens 는 호출 단위로 적용된다. + +/** 대화 턴의 역할. system 은 턴이 아니라 ChatRequest.system 으로 분리된다. */ +export type ChatTurnRole = 'user' | 'assistant' + +export interface ChatTurn { + role: ChatTurnRole + content: string +} + +/** system 과 대화 턴이 분리된 채팅 요청. */ +export interface ChatRequest { + system?: string + turns: ChatTurn[] +} + +/** 호출 단위 채팅 스트림 옵션. */ +export interface ChatStreamOptions { + model?: string + temperature?: number + maxTokens?: number + signal?: AbortSignal + timeoutMs?: number +} + +/** 호출자가 쓰는 Ollama 스타일 역할 메시지. */ +export interface RoleMessage { + role: string + content: string +} + +const SYSTEM_JOINER = '\n\n' + +function isChatTurnRole(role: string): role is ChatTurnRole { + return role === 'user' || role === 'assistant' +} + +/** + * Ollama 스타일 역할 메시지를 ChatRequest 로 바꾼다. + * 모든 system 메시지는 순서대로 이어 붙여 system 하나로 합친다. + * user/assistant 이외의 역할은 대화 턴이 아니므로 제외한다. + */ +export function toChatRequest(messages: readonly RoleMessage[]): ChatRequest { + const systemParts: string[] = [] + const turns: ChatTurn[] = [] + for (const message of messages) { + if (message.role === 'system') { + if (message.content.trim().length > 0) systemParts.push(message.content) + continue + } + if (isChatTurnRole(message.role)) { + turns.push({ role: message.role, content: message.content }) + } + } + const request: ChatRequest = { turns } + if (systemParts.length > 0) request.system = systemParts.join(SYSTEM_JOINER) + return request +} + +/** ChatRequest 를 Ollama `/api/chat` 형식(선두 system 메시지 + 턴)으로 되돌린다. */ +export function toRoleMessages(request: ChatRequest): RoleMessage[] { + const messages: RoleMessage[] = [] + if (request.system !== undefined && request.system.length > 0) { + messages.push({ role: 'system', content: request.system }) + } + for (const turn of request.turns) { + messages.push({ role: turn.role, content: turn.content }) + } + return messages +} + +/** 요청 크기 한도. */ +export interface ChatRequestLimits { + maxSystemChars: number + maxMessageChars: number + maxMessages: number + maxTotalChars: number +} + +/** + * llm-proxy 요청 한도. + * server/supabase/functions/_shared/llm-contract.ts 의 MAX_* 상수와 같은 값이어야 한다. + * (계약 일치는 desktop premium-llm 계약 테스트가 parseLlmRequest 로 검증한다.) + */ +export const LLM_PROXY_CHAT_LIMITS: Readonly = Object.freeze({ + maxSystemChars: 8_000, + maxMessageChars: 8_000, + maxMessages: 40, + maxTotalChars: 50_000, +}) + +/** 한도를 넘어 잘린 텍스트 끝에 붙는 모델용 표식 (UI 문자열 아님). */ +export const CHAT_TRUNCATION_MARKER = '\n…[truncated]' + +function truncateText(text: string, maxChars: number): string { + if (text.length <= maxChars) return text + if (maxChars <= CHAT_TRUNCATION_MARKER.length) return text.slice(0, maxChars) + return text.slice(0, maxChars - CHAT_TRUNCATION_MARKER.length).trimEnd() + CHAT_TRUNCATION_MARKER +} + +/** + * 요청을 한도 안으로 맞춘다 (순수 함수). + * - system 은 앞부분을 남기고 잘라 한도를 지킨다 (지시문이 앞에 오므로 지시가 보존된다). + * - 빈 턴은 제거한다. 빈 assistant 턴은 공급자가 거부해 이후 모든 턴을 깨뜨린다. + * - 한 턴이 한도를 넘으면 잘라 낸다. + * - 개수·총량 한도를 넘으면 오래된 턴부터 버리고, 선두 assistant 턴은 제거한다. + */ +export function fitChatRequest(request: ChatRequest, limits: Readonly): ChatRequest { + const system = request.system?.trim() + const fittedSystem = system && system.length > 0 + ? truncateText(system, limits.maxSystemChars) + : undefined + + const normalized: ChatTurn[] = [] + for (const turn of request.turns) { + const content = turn.content.trim() + if (content.length === 0) continue + normalized.push({ role: turn.role, content: truncateText(content, limits.maxMessageChars) }) + } + + const kept: ChatTurn[] = [] + let totalChars = 0 + for (let i = normalized.length - 1; i >= 0; i--) { + const turn = normalized[i] + if (kept.length >= limits.maxMessages) break + if (totalChars + turn.content.length > limits.maxTotalChars) break + totalChars += turn.content.length + kept.unshift(turn) + } + while (kept.length > 0 && kept[0].role === 'assistant') kept.shift() + + const fitted: ChatRequest = { turns: kept } + if (fittedSystem !== undefined) fitted.system = fittedSystem + return fitted +} diff --git a/server/supabase/functions/llm-proxy/index.ts b/server/supabase/functions/llm-proxy/index.ts index 701706c..e5870c0 100644 --- a/server/supabase/functions/llm-proxy/index.ts +++ b/server/supabase/functions/llm-proxy/index.ts @@ -66,6 +66,16 @@ async function issueGenerationReceipt( return parseGenerationReceiptId(data) } +/** 쓰지 않을 공급자 응답 본문을 닫아 연결과 생성을 정리한다. */ +async function discardBody(resp: Response): Promise { + if (!resp.body || resp.bodyUsed) return + try { + await resp.body.cancel() + } catch { + // 이미 닫힌 스트림 — 무시 + } +} + function generationHeaders(generationId: string | null): Record { return generationId === null ? {} : { [GENERATION_ID_HEADER]: generationId } } @@ -147,27 +157,9 @@ Deno.serve(async (req: Request) => { ) } - // 3단계: 쿼터 소비 (원자적 base → overage fallback) - const policy = getQuotaPolicy(tier, quotaKey) - const consume = await consumeQuota(user.id, quotaKey, serviceClient, policy.limit) - if (!consume.allowed) { - return new Response( - JSON.stringify({ - error: 'quota_exceeded', - model: requestedModel, - current: consume.current, - limit: consume.limit, - tier, - overage_credits: consume.overageCredits, - }), - { - status: 429, - headers: { ...corsHeaders, 'Content-Type': 'application/json' }, - }, - ) - } - - // 실제 Anthropic API 호출 (Prompt Caching 2024-07-31 활성화) + // 3단계: 공급자 호출 (Prompt Caching 2024-07-31 활성화) + // 쿼터는 공급자가 요청을 받아들인 뒤에만 소비한다. 공급자 5xx/과부하/타임아웃으로 + // 답을 주지 못한 요청이 쿼터를 깎고, 클라이언트 재시도가 한 번 더 깎는 일을 막는다. const systemPayload = buildAnthropicSystemBlocks(body.system) const anthropicResp = await fetch('https://api.anthropic.com/v1/messages', { @@ -190,12 +182,54 @@ Deno.serve(async (req: Request) => { if (!anthropicResp.ok) { console.error('Anthropic request failed', { status: anthropicResp.status }) + await discardBody(anthropicResp) return new Response(JSON.stringify({ error: 'provider_request_failed' }), { status: 502, headers: { ...corsHeaders, 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, }) } + // 비스트리밍 응답은 형식 검증까지 통과한 뒤에 소비한다. + let nonStreamData: unknown = null + if (!body.stream || !anthropicResp.body) { + nonStreamData = await anthropicResp.json() + if (!hasAssistantText(nonStreamData)) { + console.error('Anthropic returned an invalid response shape') + return new Response(JSON.stringify({ error: 'provider_invalid_response' }), { + status: 502, + headers: { ...corsHeaders, 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, + }) + } + } + + // 4단계: 쿼터 소비 (원자적 base → overage fallback). + // 동시 요청이 checkQuota 를 함께 통과했다면 여기서 초과분이 거절된다. + const policy = getQuotaPolicy(tier, quotaKey) + let consume: Awaited> + try { + consume = await consumeQuota(user.id, quotaKey, serviceClient, policy.limit) + } catch (err) { + await discardBody(anthropicResp) + throw err + } + if (!consume.allowed) { + await discardBody(anthropicResp) + return new Response( + JSON.stringify({ + error: 'quota_exceeded', + model: requestedModel, + current: consume.current, + limit: consume.limit, + tier, + overage_credits: consume.overageCredits, + }), + { + status: 429, + headers: { ...corsHeaders, 'Content-Type': 'application/json' }, + }, + ) + } + if (body.stream && anthropicResp.body) { const generationId = await issueGenerationReceipt( serviceClient, @@ -215,21 +249,13 @@ Deno.serve(async (req: Request) => { }) } - const data: unknown = await anthropicResp.json() - if (!hasAssistantText(data)) { - console.error('Anthropic returned an invalid response shape') - return new Response(JSON.stringify({ error: 'provider_invalid_response' }), { - status: 502, - headers: { ...corsHeaders, 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, - }) - } const generationId = await issueGenerationReceipt( serviceClient, user.id, generationPurpose, requestedModel, ) - return new Response(JSON.stringify(data), { + return new Response(JSON.stringify(nonStreamData), { status: 200, headers: { ...corsHeaders,