mateclaw/mateclaw-ui/src/composables/chat/useStream.ts

/**
 * SSE 流处理 Composable
 * 参考 @agentscope-ai/chat 的 Stream 实现，提供标准的 SSE 解析
 */
import { ref, computed } from 'vue'
import { handleAuthFailure, updateTokenFromHeader } from '@/utils/auth'
import { classifyHttpError, classifyNetworkError, type ChatErrorInfo } from '@/types/chatError'

export type SSEEventType =
  | 'content_delta'
  | 'thinking_delta'
  | 'message_complete'
  | 'done'
  | 'error'
  | 'session'
  | 'message_start'
  // Agent 事件
  | 'tool_call_started'
  | 'tool_call_completed'
  | 'phase'
  | 'plan_created'
  | 'plan_step_started'
  | 'plan_step_completed'
  // 审批事件
  | 'tool_approval_requested'
  | 'tool_approval_resolved'
  // 恢复/警告事件
  | 'warning'
  // Interrupt + Queue 事件
  | 'heartbeat'
  | 'turn_interrupt_requested'
  | 'turn_interrupted'
  | 'queued_input_accepted'
  | 'queued_input_started'
  // 异步任务事件
  | 'async_task_progress'
  | 'async_task_completed'
  // TTS 事件
  | 'tts_ready'
  // 浏览器执行事件
  | 'browser_action'
  // Agent 委派事件
  | 'delegation_start'
  | 'delegation_progress'
  | 'delegation_end'
  | 'delegation_child_complete'

export interface SSEEvent {
  type: SSEEventType
  data: any
}

export interface UseStreamOptions {
  /** API 端点 */
  url: string
  /** 请求方法 */
  method?: 'POST' | 'GET'
  /** 请求头 */
  headers?: Record<string, string>
  /** 请求体 */
  body?: any
  /** 是否自动开始 */
  autoStart?: boolean
}

export interface UseStreamReturn {
  /** 是否连接中 */
  isConnected: import('vue').Ref<boolean>
  /** 是否正在接收数据 */
  isReceiving: import('vue').Ref<boolean>
  /** 当前错误 */
  error: import('vue').Ref<Error | null>
  /** 连接流 */
  connect: (body?: any) => Promise<void>
  /** 断开连接 */
  disconnect: () => void
  /** 中止请求 */
  abort: () => void
  /** 注册事件处理器 */
  on: (event: SSEEventType, handler: (data: any) => void) => () => void
  /** 注册所有事件处理器 */
  onEvent: (handler: (event: SSEEvent) => void) => () => void
}

// SSE 解析器 - 使用 TransformStream 风格
class SSEParser {
  private buffer = ''
  private readonly separator = '\n\n'

  parse(chunk: string): SSEEvent[] {
    this.buffer += chunk
    const events: SSEEvent[] = []
    
    // 分割事件块
    const parts = this.buffer.split(this.separator)
    
    // 保留最后一个不完整的部分
    this.buffer = parts.pop() || ''
    
    // 处理完整的事件块
    for (const part of parts) {
      const event = this.parseEvent(part)
      if (event) {
        events.push(event)
      }
    }
    
    return events
  }

  flush(): SSEEvent[] {
    if (!this.buffer.trim()) return []
    const event = this.parseEvent(this.buffer)
    this.buffer = ''
    return event ? [event] : []
  }

  private parseEvent(part: string): SSEEvent | null {
    const lines = part.split('\n')
    let eventType: SSEEventType = 'content_delta'
    let data: any = {}
    let hasData = false

    for (const line of lines) {
      if (!line.trim()) continue
      
      const colonIndex = line.indexOf(':')
      if (colonIndex === -1) continue

      const key = line.slice(0, colonIndex).trim()
      const value = line.slice(colonIndex + 1).trim()

      if (key === 'event') {
        eventType = value as SSEEventType
      } else if (key === 'data') {
        hasData = true
        try {
          data = JSON.parse(value)
        } catch {
          data = value
        }
      }
    }

    return hasData ? { type: eventType, data } : null
  }
}

export function useStream(options: UseStreamOptions): UseStreamReturn {
  const { url, method = 'POST', headers = {}, autoStart = false } = options

  const isConnected = ref(false)
  const isReceiving = ref(false)
  const error = ref<Error | null>(null)

  let abortController: AbortController | null = null
  let parser = new SSEParser()
  let streamTimeoutTimer: ReturnType<typeof setTimeout> | null = null
  // 提高到 120 秒（后端心跳每 10 秒一次，任何心跳都会重置此计时器）
  const STREAM_TIMEOUT_MS = 120_000
  
  // 事件处理器存储
  const eventHandlers = new Map<SSEEventType, Set<(data: any) => void>>()
  const globalHandlers = new Set<(event: SSEEvent) => void>()

  // 触发事件
  const emit = (event: SSEEvent) => {
    // 全局处理器
    globalHandlers.forEach(handler => {
      try {
        handler(event)
      } catch (e) {
        console.error('Stream event handler error:', e)
      }
    })

    // 特定类型处理器
    const handlers = eventHandlers.get(event.type)
    if (handlers) {
      handlers.forEach(handler => {
        try {
          handler(event.data)
        } catch (e) {
          console.error('Stream event handler error:', e)
        }
      })
    }
  }

  // 处理 SSE 数据
  const processChunk = (chunk: string) => {
    const events = parser.parse(chunk)
    events.forEach(emit)
  }

  // 连接流
  const connect = async (body?: any) => {
    // 断开已有连接
    disconnect()
    
    parser = new SSEParser()
    error.value = null
    
    try {
      abortController = new AbortController()
      
      // 自动从 localStorage 读取 token（与 http.ts 拦截器保持一致）
      const authHeaders: Record<string, string> = {}
      const token = localStorage.getItem('token')
      if (token) {
        authHeaders['Authorization'] = `Bearer ${token}`
      }

      const response = await fetch(url, {
        method,
        headers: {
          'Content-Type': 'application/json',
          'Accept': 'text/event-stream',
          ...authHeaders,
          ...headers,
        },
        body: body ? JSON.stringify(body) : undefined,
        signal: abortController.signal,
      })

      if (!response.ok) {
        // 尝试解析错误响应体（后端返回 JSON 格式错误信息）
        let errorMsg = `HTTP ${response.status}: ${response.statusText}`
        let errorBody: any = null
        try {
          errorBody = await response.json()
          if (errorBody.msg || errorBody.message) {
            errorMsg = errorBody.msg || errorBody.message
          }
        } catch {
          // 非 JSON 响应，使用默认错误信息
        }
        // 构建结构化错误信息
        const errorInfo = classifyHttpError(response.status, errorBody)
        errorInfo.requestId = response.headers.get('X-Request-Id')
          || errorBody?.requestId
          || errorInfo.requestId
        // 401 = authentication failure → log out (consistent with http.ts).
        // 403 = authorization failure (e.g. workspace permission denied) → keep session.
        if (response.status === 401) {
          handleAuthFailure()
        }
        throw Object.assign(new Error(errorMsg), { errorInfo })
      }

      // 滑动窗口续期：从响应头获取新 Token
      updateTokenFromHeader(response.headers)

      if (!response.body) {
        throw new Error('Response body is null')
      }

      isConnected.value = true
      isReceiving.value = true

      // 流级超时：若长时间无数据到达则中止
      const resetStreamTimeout = () => {
        if (streamTimeoutTimer) clearTimeout(streamTimeoutTimer)
        streamTimeoutTimer = setTimeout(() => {
          if (isReceiving.value && abortController) {
            abortController.abort()
            const timeoutInfo: ChatErrorInfo = {
              category: 'timeout',
              rawMessage: 'Stream timeout: no data received',
              retryable: true,
              timestamp: Date.now(),
            }
            error.value = Object.assign(new Error('Stream timeout'), { errorInfo: timeoutInfo })
            emit({ type: 'error', data: { message: 'Stream timeout', errorInfo: timeoutInfo } })
          }
        }, STREAM_TIMEOUT_MS)
      }
      resetStreamTimeout()

      const reader = response.body.getReader()
      const decoder = new TextDecoder('utf-8')

      // 读取循环
      const read = async () => {
        try {
          while (true) {
            const { done, value } = await reader.read()

            if (done) {
              break
            }

            resetStreamTimeout()
            const chunk = decoder.decode(value, { stream: true })
            processChunk(chunk)
          }

          // 处理剩余数据
          const remaining = decoder.decode()
          if (remaining) {
            processChunk(remaining)
          }
          
          // flush 剩余事件
          const flushEvents = parser.flush()
          flushEvents.forEach(emit)
          
        } catch (e) {
          if (e instanceof Error && e.name === 'AbortError') {
            // 用户主动中止，不是错误
            return
          }
          throw e
        } finally {
          isReceiving.value = false
          isConnected.value = false
          reader.releaseLock()
        }
      }

      await read()
      
    } catch (e) {
      if (e instanceof Error && e.name === 'AbortError') {
        return
      }
      error.value = e instanceof Error ? e : new Error(String(e))
      const errorInfo: ChatErrorInfo = (e as any)?.errorInfo
        || classifyNetworkError(error.value)
      emit({ type: 'error', data: { message: error.value.message, errorInfo } })
    } finally {
      if (streamTimeoutTimer) {
        clearTimeout(streamTimeoutTimer)
        streamTimeoutTimer = null
      }
      isReceiving.value = false
      isConnected.value = false
    }
  }

  // 断开连接
  const disconnect = () => {
    if (streamTimeoutTimer) {
      clearTimeout(streamTimeoutTimer)
      streamTimeoutTimer = null
    }
    if (abortController) {
      abortController.abort()
      abortController = null
    }
    isConnected.value = false
    isReceiving.value = false
  }

  // 中止请求（别名）
  const abort = disconnect

  // 注册事件处理器
  const on = (event: SSEEventType, handler: (data: any) => void) => {
    if (!eventHandlers.has(event)) {
      eventHandlers.set(event, new Set())
    }
    eventHandlers.get(event)!.add(handler)

    // 返回取消订阅函数
    return () => {
      eventHandlers.get(event)?.delete(handler)
    }
  }

  // 注册全局事件处理器
  const onEvent = (handler: (event: SSEEvent) => void) => {
    globalHandlers.add(handler)
    return () => {
      globalHandlers.delete(handler)
    }
  }

  return {
    isConnected,
    isReceiving,
    error,
    connect,
    disconnect,
    abort,
    on,
    onEvent,
  }
}

export default useStream