import type { ChatErrorInfo } from '../types/workbench'; import type { MessageExportCapability } from '../types/replyExport'; import { fallbackErrorClass } from './chatRetry'; import { mergeAuthHeaders } from './authHeaders'; import { apiFetch } from './client'; export type StreamEvent = { id?: string; type: string; phase?: { phase_id?: string; message?: string; step_index?: number; total_steps?: number; progress?: number; }; status?: { message?: string; intent?: string }; delta?: string; error?: string; deliverable?: { task_key?: string; title?: string; file_path?: string; word_count?: number; quality_score?: number; }; governance?: { doc_id?: string; action?: string; current_state?: string; is_locked?: boolean; author_claimed?: boolean; notification_title?: string; notification_msg?: string; }; error_info?: { user_message?: string; code?: string; class?: string; phase_id?: string; upstream?: string; request_id?: string; session_id?: string; workspace_id?: string; actions?: string[]; guide?: string; }; turn?: { client_turn_id?: string; user_message_id?: number; status?: string; duplicate?: boolean; }; message_id?: number; user_message_id?: number; content_revision?: number; content_sha256?: string; export_capability?: MessageExportCapability; done?: { session_id?: string; duration_ms?: number; total_events?: number; workspace_id?: string; message_id?: number; user_message_id?: number; content_revision?: number; content_sha256?: string; export_capability?: MessageExportCapability; }; }; type ErrorAction = ChatErrorInfo['actions'][number]; const VALID_ACTIONS: readonly ErrorAction[] = ['retry', 'continue', 'restart']; function isErrorClass(value: unknown): value is ChatErrorInfo['class'] { return ( value === 'transient' || value === 'resumable' || value === 'fatal' || value === 'cancelled' ); } function defaultActionsForClass(cls: ChatErrorInfo['class']): ChatErrorInfo['actions'] { switch (cls) { case 'transient': return ['retry']; case 'resumable': return ['continue', 'restart']; case 'fatal': return ['restart']; case 'cancelled': return ['retry']; default: { const _exhaustive: never = cls; return _exhaustive; } } } function parseActions(raw?: string[]): ChatErrorInfo['actions'] { if (!raw?.length) return []; return raw.filter((a): a is ErrorAction => (VALID_ACTIONS as readonly string[]).includes(a)); } function parseFrame(frame: string): { event: StreamEvent | null; seq?: number } { if (!frame.trim()) return { event: null }; let dataLine: string | undefined; let seq: number | undefined; for (const line of frame.split(/\r?\n/)) { if (line.startsWith('id:')) { const n = Number(line.slice(3).trim()); if (Number.isFinite(n)) seq = n; } if (line.startsWith('data:')) { dataLine = line.slice(5).trim(); } } // SSE comment / heartbeat frames (": ...") have no data — ignore. if (!dataLine) return { event: null, seq }; try { return { event: JSON.parse(dataLine) as StreamEvent, seq }; } catch (err) { console.warn('[sse] 忽略无法解析的帧:', dataLine.slice(0, 200), err); return { event: null, seq }; } } /** Split complete SSE frames; accept both LF and CRLF blank-line delimiters. */ function takeSSEFrames( buffer: string, flush?: boolean, ): { frames: string[]; rest: string } { const frames: string[] = []; let start = 0; while (start < buffer.length) { const lfIdx = buffer.indexOf('\n\n', start); const crlfIdx = buffer.indexOf('\r\n\r\n', start); let delimAt = -1; let delimLen = 0; if (lfIdx === -1 && crlfIdx === -1) { break; } if (crlfIdx !== -1 && (lfIdx === -1 || crlfIdx <= lfIdx)) { delimAt = crlfIdx; delimLen = 4; } else { delimAt = lfIdx; delimLen = 2; } frames.push(buffer.slice(start, delimAt)); start = delimAt + delimLen; } if (flush && start < buffer.length) { frames.push(buffer.slice(start)); return { frames, rest: '' }; } return { frames, rest: buffer.slice(start) }; } export function parseSSEChunk( buffer: string, options?: { flush?: boolean }, ): { events: StreamEvent[]; rest: string; lastSeq?: number } { const events: StreamEvent[] = []; if (!buffer) return { events, rest: '' }; const { frames, rest } = takeSSEFrames(buffer, options?.flush); let lastSeq: number | undefined; for (const frame of frames) { const parsed = parseFrame(frame); if (parsed.seq !== undefined) lastSeq = parsed.seq; if (parsed.event) events.push(parsed.event); } return { events, rest, lastSeq }; } export async function streamChat( body: Record, handlers: { onEvent: (ev: StreamEvent) => void; onRequestId?: (id: string) => void; }, signal?: AbortSignal, options?: { idempotencyKey?: string }, ): Promise { const headers = mergeAuthHeaders( { Accept: 'text/event-stream', ...(options?.idempotencyKey ? { 'Idempotency-Key': options.idempotencyKey } : {}), }, 'application/json', ); const resp = await apiFetch( '/api/v1/chat/stream', { method: 'POST', headers, body: JSON.stringify(body), signal, }, { retry: false }, ); const rid = resp.headers.get('X-Request-Id') || resp.headers.get('x-request-id'); if (rid) handlers.onRequestId?.(rid); if (!resp.ok || !resp.body) { throw Object.assign(new Error(`HTTP ${resp.status}`), { httpStatus: resp.status, requestId: rid, }); } const reader = resp.body.getReader(); const decoder = new TextDecoder(); let buf = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buf += decoder.decode(value, { stream: true }); const parsed = parseSSEChunk(buf); buf = parsed.rest; for (const ev of parsed.events) handlers.onEvent(ev); } if (buf) { const flushed = parseSSEChunk(buf, { flush: true }); for (const ev of flushed.events) handlers.onEvent(ev); } } export function mapErrorInfo(ev: StreamEvent): ChatErrorInfo { const ei = ev.error_info ?? {}; const userMessage = ei.user_message ?? ev.error ?? '发生未知错误'; const code = ei.code ?? 'UNKNOWN_ERROR'; const cls: ChatErrorInfo['class'] = isErrorClass(ei.class) ? ei.class : fallbackErrorClass({ workspaceId: ei.workspace_id }); let actions = parseActions(ei.actions); if (!Array.isArray(ei.actions)) { actions = defaultActionsForClass(cls); } return { userMessage, code, class: cls, phaseId: ei.phase_id, upstream: ei.upstream, requestId: ei.request_id, sessionId: ei.session_id, workspaceId: ei.workspace_id, guide: ei.guide, actions, }; }