agentic-coding-workflow/web/src/composables/useRunStream.ts

// useRunStream 封装 run profile 的日志 WebSocket，结构照搬 useAcpStream：
//   - connect + 自动重连（指数退避 1s/3s，最多 2 次）
//   - ?since=lastLogID 续传
//   - 把日志行累积进 ref 数组；state 帧更新 runStatus
import { ref, onScopeDispose, type Ref } from 'vue'
import { openRunLogsWS, type LogLine, type RunState } from '@/api/runs'
import { useAuthStore } from '@/stores/auth'

export type StreamStatus = 'idle' | 'connecting' | 'streaming' | 'closed' | 'error'

export interface UseRunStreamReturn {
  status: Ref<StreamStatus>
  runStatus: Ref<RunState>
  exitCode: Ref<number | null>
  logs: Ref<LogLine[]>
  reconnect: () => void
  close: () => void
  clear: () => void
}

interface WireFrame {
  id?: number
  kind?: string
  level?: 'info' | 'error'
  text?: string
  ts?: string
  status?: RunState
  exit_code?: number
}

export function useRunStream(profileId: string): UseRunStreamReturn {
  const status = ref<StreamStatus>('idle')
  const runStatus = ref<RunState>('stopped')
  const exitCode = ref<number | null>(null)
  const logs = ref<LogLine[]>([])
  const lastLogID = ref(0)

  let ws: WebSocket | null = null
  let retried = 0
  let manualClose = false
  let reconnectTimer: ReturnType<typeof setTimeout> | null = null

  const auth = useAuthStore()

  function connect() {
    if (ws || manualClose) return
    status.value = 'connecting'
    const token = auth.token ?? ''
    const sock = openRunLogsWS(profileId, lastLogID.value, token)
    ws = sock

    sock.onopen = () => {
      status.value = 'streaming'
      retried = 0
    }

    sock.onmessage = (ev: MessageEvent) => {
      let parsed: WireFrame
      try {
        parsed = JSON.parse(ev.data as string) as WireFrame
      } catch {
        return
      }
      if (!parsed || typeof parsed !== 'object') return

      if (parsed.kind === 'state') {
        if (parsed.status) runStatus.value = parsed.status
        exitCode.value = parsed.exit_code ?? null
        return
      }
      if (parsed.kind === 'slow_consumer_disconnect') {
        status.value = 'error'
        return
      }
      // 普通日志行
      if (parsed.kind === 'log' && typeof parsed.id === 'number') {
        if (parsed.id > lastLogID.value) lastLogID.value = parsed.id
        logs.value.push({
          id: parsed.id,
          level: parsed.level ?? 'info',
          text: parsed.text ?? '',
          ts: parsed.ts ?? '',
        })
      }
    }

    sock.onerror = () => {
      status.value = 'error'
    }

    sock.onclose = () => {
      ws = null
      if (manualClose) {
        status.value = 'closed'
        return
      }
      if (retried < 2) {
        const delay = retried === 0 ? 1000 : 3000
        retried += 1
        reconnectTimer = setTimeout(connect, delay)
      } else {
        status.value = 'closed'
      }
    }
  }

  function reconnect() {
    if (ws) ws.close()
    retried = 0
    manualClose = false
    connect()
  }

  function close() {
    manualClose = true
    if (reconnectTimer) {
      clearTimeout(reconnectTimer)
      reconnectTimer = null
    }
    if (ws) ws.close()
    status.value = 'closed'
  }

  function clear() {
    logs.value = []
  }

  connect()
  onScopeDispose(close)

  return { status, runStatus, exitCode, logs, reconnect, close, clear }
}