import type { ChatMessage } from './chat'; import { API_BASE_URL } from './http'; export interface StreamEvents { onMeta?: (data: any) => void; onReasoningDelta?: (text: string) => void; onDelta?: (text: string) => void; onRetry?: (data: any) => void; onToolCall?: (data: { id: string; name: string; args: any }) => void; onToolResult?: (data: { id: string; name: string; result: any }) => void; onDone?: (data: { user: ChatMessage; assistant: ChatMessage }) => void; onAborted?: (data: { assistant: ChatMessage }) => void; onError?: (msg: string) => void; } export interface ModelOverrides { model?: string; model_id?: string; temperature?: number; topP?: number; maxTokens?: number; } export async function streamChat( roomId: string, targetAgentId: string, content: string, handlers: StreamEvents, signal?: AbortSignal, model?: string, modelId?: string, imageUrls?: string[] ) { const resp = await fetch(`${API_BASE_URL}/rooms/${roomId}/messages/stream`, { method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' }, body: JSON.stringify({ content, targetAgentId, model, model_id: modelId, imageUrls: imageUrls ?? [] }), signal, credentials: 'include' }); return await consumeSSE(resp, handlers, signal); } export async function regenerateMessage( agentId: string, messageId: string, handlers: StreamEvents, signal?: AbortSignal, overrides?: ModelOverrides, attachmentsText?: string ) { const resp = await fetch(`${API_BASE_URL}/chat/${agentId}/messages/${messageId}/regenerate`, { method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' }, body: JSON.stringify({ overrides, attachmentsText }), signal, credentials: 'include' }); return await consumeSSE(resp, handlers, signal); } async function consumeSSE(resp: Response, h: StreamEvents, signal?: AbortSignal) { if (!resp.ok || !resp.body) { const txt = await resp.text().catch(() => ''); h.onError?.(`HTTP ${resp.status}: ${txt}`); return; } const reader = resp.body.getReader(); const decoder = new TextDecoder('utf-8'); let buf = ''; let hasError = false; try { while (true) { const { value, done } = await reader.read(); if (done || hasError) break; buf += decoder.decode(value, { stream: true }); buf = buf.replace(/\r\n/g, '\n'); let idx; while ((idx = buf.indexOf('\n\n')) !== -1 && !hasError) { const raw = buf.slice(0, idx); buf = buf.slice(idx + 2); if (!raw.trim() || raw.startsWith(':')) continue; let event = 'message'; let dataStr = ''; for (const line of raw.split('\n')) { if (line.startsWith('event:')) event = line.slice(6).trim(); else if (line.startsWith('data:')) { let part = line.slice(5); if (part.startsWith(' ')) part = part.slice(1); dataStr += (dataStr ? '\n' : '') + part; } } if (!dataStr) continue; let data: any; try { data = JSON.parse(dataStr); } catch { continue; } switch (event) { case 'meta': h.onMeta?.(data); break; case 'retry': h.onRetry?.(data); break; case 'reasoning_delta': h.onReasoningDelta?.(data.content || ''); break; case 'delta': h.onDelta?.(data.content || ''); break; case 'tool_call': h.onToolCall?.(data); break; case 'tool_result': h.onToolResult?.(data); break; case 'done': if (!hasError) { h.onDone?.(data); } break; case 'aborted': h.onAborted?.(data); break; case 'error': hasError = true; h.onError?.(data.message || 'stream error'); break; } } } } catch (e: any) { if (signal?.aborted || e?.name === 'AbortError') { return; } h.onError?.(e?.message ?? String(e)); } finally { reader.cancel().catch(() => {}); } }