agent-management/skills/agent-management-external-workflow/references/platform/composables/useWorkflowRunner.js

import { ref, reactive } from 'vue'
import { useMessage } from 'naive-ui'
import { runWorkflow, reconnectRunStream, getRunState, submitClarifyAnswer } from '../api/workflow'
import { stripHermesThinkingDecoration } from '../utils/hermesThinking'
import { logTypeLabel as sharedLogTypeLabel } from '../utils/logLabels'
import { parseVisitDecision } from '../utils/visitChain'

/**
 * 判断对象是否为节点输出 envelope `{status, message, data}`。
 * 旧版直接输出 Map 的记录没有 status/data 字段，返回 false 以走兼容分支。
 */
export function isEnvelope(o) {
  return !!(o && typeof o === 'object' && 'status' in o && 'data' in o)
}

/**
 * 解包节点输出：envelope 返回 data，否则原样返回。
 * 兜底 data 缺失时返回空对象，避免下游 v-for 报错。
 */
export function unwrapNodeOutput(o) {
  return isEnvelope(o) ? (o.data || {}) : (o || {})
}

/**
 * 获取节点输出 envelope.status（200=成功，400=失败）。
 * 非 envelope 返回 null（视为成功）。
 */
export function nodeOutputStatus(o) {
  return isEnvelope(o) ? o.status : null
}

/**
 * 获取节点输出 envelope.message（含失败原因、重试次数等提示）。
 * 非 envelope 返回空串。
 */
export function nodeOutputMessage(o) {
  return isEnvelope(o) ? (o.message || '') : ''
}

/**
 * 工作流运行时 composable
 * 抽取自 WorkflowEditor.vue，承担 SSE 流解析、节点状态更新、运行结果管理
 *
 * <p>v2.0 卡片统一：DAG / AGENTIC 模式均使用 Visit 作为单一卡片数据源。
 * 所有 SSE 事件（node_status / node_stream / visit_recorded）按 visitUuid 路由到 visits 数组；
 * 同节点多次访问时各 visit 拥有独立 visitUuid，互不冲突。</p>
 *
 * @param {Object} options
 * @param {import('vue').Ref} options.getNodes 画布节点（用于按 nodeId 找节点 + 设置 class）
 * @param {Object} options.nodeTypeInfo 节点类型 → { color, typeLabel } 映射
 * @param {import('vue').Ref} options.workflowIdRef 当前工作流 name 或 id（仅用于 lastRunWorkflowId 记录）
 */
export function useWorkflowRunner({ getNodes, nodeTypeInfo, workflowIdRef }) {
  const message = useMessage()

  const running = ref(false)
  const showRunDialog = ref(false)
  const showResult = ref(false)
  const workflowResult = ref(null)
  /**
   * 当前运行元数据 { runId, workingDirFiles }。
   * 这两个字段不再注入 output 节点 envelope，而是作为 workflow-run 的属性由后端 SSE 推送 /
   * 持久化到 WorkflowRun 实体。前端在"下载工作空间"按钮旁独立展示，与下载入口出现时机一致。
   */
  const runMeta = ref(null)
  // 画布节点状态（按 nodeId 索引，用于画布节点着色 / 徽章）
  const nodeStatuses = reactive({})
  const activeTab = ref('props')
  /**
   * v2.0：统一卡片列表（DAG / AGENTIC 模式均使用）。
   * 每个 visit 携带完整执行上下文：
   *  - 元数据：visitUuid / nodeId / nodeType / nodeLabel / iterSeq / fromVisitUuid / fromNodeId / sortOrder
   *  - 结果：status / output / logs / contextSnapshot / variablesJson / outputJson / errorMsg
   *  - 流式：streamingThinking / currentStatus / currentTool / toolHistory / todos
   *  - UI：color / typeLabel
   */
  const visits = reactive([])
  const expandedContext = reactive({})
  // 思考过程折叠状态：key=visitUuid，value=true/false。未设置默认折叠（避免运行结果页过长）。
  // 折叠时仅隐藏 UI，streamingThinking 仍持续累加，展开后可继续看到完整内容。
  const expandedThinking = reactive({})
  // 执行日志折叠状态：同 expandedThinking。Visit running 时强制展开，完成时自动收起（清空 key）。
  const expandedLogs = reactive({})
  const lastRunId = ref(null)
  const lastRunWorkflowId = ref(null)
  // v1.1：断线重连相关状态
  // lastEventId：从 SSE 帧的 id: 字段提取，用于断线重连续接
  // reconnecting：是否正在重连（控制 UI 提示）
  const lastEventId = ref(null)
  const reconnecting = ref(false)
  // 重连失败次数上限：超过则退化为 GET /state 轮询
  const MAX_RECONNECT_ATTEMPTS = 5
  // v1.1：全局唯一 Askfor 暂停点 {clarifyId, question, choices}
  const pendingAsk = ref(null)
  // v1.1：轮询定时器（重连耗尽后启动）
  const pollTimer = ref(null)

  function resetState() {
    Object.keys(nodeStatuses).forEach(k => delete nodeStatuses[k])
    visits.splice(0, visits.length)
  }

  function clearNodeStatusClasses() {
    getNodes.value.forEach(n => { n.class = '' })
  }

  /**
   * 启动工作流运行
   * @param {string|number} nameOrId 工作流 name（kebab-case）或旧数字 id
   * @param {Object} inputs 表单输入值
   * @param {Array<{file:File, relativePath:string}>} uploadFiles 上传文件列表
   */
  async function startRun(nameOrId, inputs, uploadFiles) {
    showRunDialog.value = false
    running.value = true
    showResult.value = false
    workflowResult.value = null
    runMeta.value = null
    resetState()
    clearNodeStatusClasses()
    // 清空折叠状态（不影响本轮首次默认展开）
    Object.keys(expandedThinking).forEach(k => delete expandedThinking[k])
    Object.keys(expandedLogs).forEach(k => delete expandedLogs[k])
    // 重置断线重连状态
    lastEventId.value = null
    reconnecting.value = false
    // H-7：清理上一轮可能残留的轮询定时器（用户未关闭结果页直接发起下次运行时，旧 pollTimer 会与新 run 并存）
    if (pollTimer.value) {
      clearInterval(pollTimer.value)
      pollTimer.value = null
    }
    // 启动时单次切到「执行结果」Tab；后续不再强制切换，允许用户自由回到「节点属性」
    activeTab.value = 'result'

    let response
    try {
      response = await runWorkflow(
        nameOrId,
        inputs,
        uploadFiles.map(f => f.file),
        uploadFiles.map(f => f.relativePath)
      )
      if (!response.ok) throw new Error(`HTTP ${response.status}`)
    } catch (e) {
      message.error('工作流启动失败: ' + e.message)
      running.value = false
      return
    }

    // 进入 SSE 消费循环（H-6/H-7：改为迭代式 runSSELoop，避免递归导致的重连计数重置与栈累积）
    await runSSELoop(response)
  }

  /**
   * H-6/H-7：消费单段 SSE 流（不递归、不自动重连）。
   * 只负责：解析事件 → 调 handleSSEEvent → 返回 { completed, receivedAny }。
   * 重连策略由 runSSELoop 统一管理，避免原 consumeSSEStream → tryReconnect → consumeSSEStream
   * 递归调用导致的 attempts 计数重置（每次递归都从 0 开始，实际重连次数无上限）与栈累积。
   *
   * @returns {{ completed: boolean, receivedAny: boolean }}
   *          completed=true 表示工作流已终止或不可重连（外层循环应退出）
   *          receivedAny=true 表示曾收到至少一个事件（连接健康，外层应清零重连计数）
   */
  async function consumeSSEStream(response) {
    const reader = response.body.getReader()
    const decoder = new TextDecoder()
    let buffer = ''
    let receivedAny = false

    try {
      while (true) {
        const { done, value } = await reader.read()
        if (done) break

        buffer += decoder.decode(value, { stream: true })
        const parts = buffer.split('\n\n')
        buffer = parts.pop()

        for (const part of parts) {
          const lines = part.split('\n')
          let data = null
          let eventId = null
          for (const line of lines) {
            if (line.startsWith('id:')) {
              eventId = line.substring(3).trim()
            } else if (line.startsWith('data:')) {
              try { data = JSON.parse(line.substring(5).trim()) } catch { /* skip */ }
            }
          }
          if (eventId != null) {
            lastEventId.value = Number(eventId)
            receivedAny = true
          }
          // comment 帧心跳（":keep-alive"）已被 split('\n\n') 切割掉，正常忽略
          if (data) handleSSEEvent(data)
        }
      }
      // 流正常结束：可能后端早完成但 SSE 提前断，由 runSSELoop 决定是否重连
      return { completed: isTerminal() || !lastRunId.value, receivedAny }
    } catch (e) {
      // reader.read() 抛错：若已无 runId 或已终止，直接当作完成；否则交回 runSSELoop 重连
      if (!lastRunId.value || isTerminal()) {
        message.error('工作流执行失败: ' + e.message)
        return { completed: true, receivedAny }
      }
      return { completed: false, receivedAny }
    }
  }

  /**
   * 判断当前工作流是否已终止（complete / error），用于决定是否值得重连。
   */
  function isTerminal() {
    return showResult.value || workflowResult.value !== null
  }

  /**
   * H-6/H-7：迭代式 SSE 消费 + 重连驱动。
   */
  async function runSSELoop(initialResponse) {
    let response = initialResponse
    let attempts = 0

    while (true) {
      const result = await consumeSSEStream(response)
      if (result.receivedAny) {
        attempts = 0
      }
      if (result.completed) {
        reconnecting.value = false
        running.value = false
        return
      }

      if (attempts >= MAX_RECONNECT_ATTEMPTS) {
        reconnecting.value = false
        message.warning('重连失败次数过多，回退到状态轮询')
        await pollRunState()
        return
      }

      attempts++
      reconnecting.value = true
      message.warning(`连接中断，尝试重连 (${attempts}/${MAX_RECONNECT_ATTEMPTS})`)
      try {
        await new Promise(r => setTimeout(r, 3000))
        const resp = await reconnectRunStream(lastRunId.value, lastEventId.value)
        if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
        reconnecting.value = false
        message.success('已重新连接')
        response = resp
      } catch (e) {
        reconnecting.value = false
      }
    }
  }

  /**
   * v1.1：回退方案 —— 每 30 秒轮询 GET /state。
   */
  async function pollRunState() {
    if (!lastRunId.value) {
      running.value = false
      return
    }
    const poll = async () => {
      try {
        const res = await getRunState(lastRunId.value)
        const state = res?.data || res
        if (state.currentPause && state.status === 'RUNNING') {
          if (!pendingAsk.value) {
            // 重连场景：按 nodeId 找到对应 Visit（取最近一条），让 Askfor 卡片能按 visitUuid 匹配
            const pausedVisits = visits.filter(v => v.nodeId === state.currentPause.nodeId)
            const matchedVisit = pausedVisits[pausedVisits.length - 1]
            pendingAsk.value = {
              nodeId: state.currentPause.nodeId || null,
              visitUuid: matchedVisit?.visitUuid || null,
              clarifyId: state.currentPause.clarifyId,
              question: state.currentPause.question,
              choices: safeParseChoices(state.currentPause.choices)
            }
          }
        }
        if (state.status === 'SUCCESS' || state.status === 'FAILED') {
          running.value = false
          if (state.status === 'SUCCESS') {
            message.success('工作流执行完成（来自轮询）')
          } else {
            message.error('工作流执行失败（来自轮询）')
          }
          return true
        }
        return false
      } catch (e) {
        return false
      }
    }
    const terminated = await poll()
    if (terminated || !running.value) return
    pollTimer.value = setInterval(async () => {
      const done = await poll()
      if (done || !running.value) {
        clearInterval(pollTimer.value)
        pollTimer.value = null
      }
    }, 30000)
  }

  function safeParseChoices(s) {
    if (!s) return []
    try {
      const v = typeof s === 'string' ? JSON.parse(s) : s
      return Array.isArray(v) ? v : []
    } catch { return [] }
  }

  /**
   * v2.0：按 visitUuid 确保 visits 中存在对应条目。
   * <p>路由策略：
   * <ol>
   *   <li>携带 visitUuid 时严格按 visitUuid 路由：找到则复用，找不到则新建（绝不 fallback 到 nodeId，
   *       否则同节点多次访问会被合并到同一张卡片——例如 A→B→B 实际只显示 A→B(第2次)，B 第 1 次被覆盖）</li>
   *   <li>visitUuid 缺失时按 nodeId 查最近一条（兼容老数据 / 事件乱序场景）</li>
   *   <li>都查不到则创建新条目（nodeType/label 优先取事件携带值，其次按 nodeId 从 getNodes 推断）</li>
   * </ol>
   * </p>
   * <p>v1.2：新增 eventMeta 可选参数（node_status RUNNING 事件自身）。
   * 后端在事件中携带 nodeType/nodeLabel，外部集成场景无画布数据（getNodes 为空）时，
   * 优先使用事件值推断类型，避免运行期间显示"未知"。</p>
   */
  function ensureVisit(visitUuid, nodeId, eventMeta) {
    if (visitUuid) {
      const v = visits.find(r => r.visitUuid === visitUuid)
      if (v) return v
      // 携带了 visitUuid 但 visits 里没有 → 必须新建，不能 fallback 到同节点的旧 visit
    } else if (nodeId) {
      const byNode = visits.filter(r => r.nodeId === nodeId)
      if (byNode.length > 0) return byNode[byNode.length - 1]
    }
    const node = nodeId ? getNodes.value.find(n => n.id === nodeId) : null
    const nodeType = eventMeta?.nodeType || node?.type || null
    const nodeLabel = eventMeta?.nodeLabel || node?.data?.label || nodeId || '节点'
    const info = nodeTypeInfo[nodeType] || { color: '#666', typeLabel: '未知' }
    const newVisit = {
      visitUuid: visitUuid || `temp-${nodeId || 'unknown'}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
      nodeId: nodeId || null,
      nodeType: nodeType,
      nodeLabel: nodeLabel,
      iterSeq: 1,
      fromVisitUuid: null,
      fromNodeId: null,
      // offCanvas：等 visit_recorded 到达时按 decisionJson 同步；初始为 false（保守不显示）
      offCanvas: false,
      sortOrder: visits.length,
      status: 'running',
      errorMsg: null,
      output: null,
      logs: [],
      contextSnapshot: null,
      variablesJson: '{}',
      outputJson: null,
      streamingThinking: '',
      currentStatus: '',
      currentTool: null,
      toolHistory: [],
      todos: [],
      color: info.color,
      typeLabel: info.typeLabel
    }
    visits.push(newVisit)
    return newVisit
  }

  function handleSSEEvent(event) {
    // v1.1：从任意事件提取 runId，便于断线重连
    if (event.runId && !lastRunId.value) {
      lastRunId.value = event.runId
      lastRunWorkflowId.value = workflowIdRef.value
    }

    if (event.type === 'node_status' && event.nodeId) {
      const status = event.status?.toLowerCase() || 'running'
      nodeStatuses[event.nodeId] = status
      const node = getNodes.value.find(n => n.id === event.nodeId)
      if (node) node.class = `node-status-${status}`

      // 传入 event 本身：RUNNING 事件携带 nodeType/nodeLabel（v1.2），
      // 无画布数据时 ensureVisit 优先用事件值建卡片，避免类型显示"未知"
      const visit = ensureVisit(event.visitUuid, event.nodeId, event)
      // 节点完成（success/failed/skipped/no_run）时自动收起思考过程、清空实时状态
      // 仅 running → finished 的首次状态变更才触发折叠/清空；重连或重复事件不会反复折叠
      const FINISHED_STATES = ['success', 'failed', 'skipped', 'no_run']
      const wasFinished = FINISHED_STATES.includes(visit.status)
      const isFinished = FINISHED_STATES.includes(status)
      if (isFinished && !wasFinished) {
        delete expandedThinking[visit.visitUuid]
        delete expandedLogs[visit.visitUuid]
      }
      // 更新 visit 字段（保留 streamingThinking / toolHistory / todos 等流式累加字段）
      visit.status = status
      if (event.output !== undefined && event.output !== null) visit.output = event.output
      if (event.logs) visit.logs = filterNonThinkingLogs(event.logs)
      if (event.contextSnapshot) visit.contextSnapshot = event.contextSnapshot
      if (isFinished) visit.currentStatus = ''
      if (status === 'failed' && event.error) visit.errorMsg = event.error
    } else if (event.type === 'node_stream' && event.nodeId) {
      // 按 visitUuid 路由到对应 Visit 卡片；visitUuid 缺失时 fallback 到 nodeId
      const visit = ensureVisit(event.visitUuid, event.nodeId)
      if (!visit) return

      const kind = event.kind
      const content = event.content || ''
      const toolName = event.toolName || null

      if (kind === 'thinking') {
        if (content) visit.streamingThinking = (visit.streamingThinking || '') + content
      } else if (kind === 'status') {
        visit.currentStatus = stripHermesThinkingDecoration(content)
      } else if (kind === 'step') {
        try {
          const obj = JSON.parse(content)
          const iteration = obj.iteration ?? '?'
          visit.currentStatus = `第 ${iteration} 轮 LLM 调用`
        } catch { /* ignore */ }
      } else if (kind === 'tool_progress') {
        try {
          const obj = JSON.parse(content)
          if (obj.event_type === 'tool.started' && obj.name) {
            const tn = obj.name ?? '工具'
            visit.currentStatus = `正在调用 ${tn}`
          }
        } catch { /* ignore */ }
      } else if (kind === 'tool_call') {
        if (visit.currentTool) {
          visit.toolHistory.unshift(visit.currentTool)
        }
        visit.currentTool = { name: toolName, argsSummary: content, result: null }
      } else if (kind === 'tool_result') {
        if (visit.currentTool) {
          visit.currentTool.result = content
          visit.toolHistory.unshift(visit.currentTool)
          visit.currentTool = null
        }
      } else if (kind === 'todo_update') {
        try {
          const parsed = JSON.parse(content)
          const arr = Array.isArray(parsed) ? parsed : (parsed?.todos || [])
          visit.todos = Array.isArray(arr) ? arr : []
        } catch { /* ignore */ }
      } else if (kind === 'clarify_request') {
        // Askfor：全局唯一 pendingAsk（同 run 同时只有一个 pending）
        try {
          const obj = JSON.parse(content)
          if (obj.clarify_id) {
            pendingAsk.value = {
              nodeId: event.nodeId || null,
              visitUuid: event.visitUuid || null,
              clarifyId: obj.clarify_id,
              question: obj.question || '',
              choices: Array.isArray(obj.choices) ? obj.choices : []
            }
          }
        } catch { /* ignore */ }
      }
    } else if (event.type === 'visit_recorded' && event.visit) {
      // 更新对应 Visit 的元数据（visitUuid / iterSeq / fromNodeId / sortOrder / variablesJson / outputJson）
      const v = event.visit
      const existing = ensureVisit(v.visitUuid, v.nodeId)
      existing.visitUuid = v.visitUuid
      existing.nodeId = v.nodeId
      existing.nodeType = v.nodeType || existing.nodeType
      existing.nodeLabel = v.nodeLabel || v.nodeId || existing.nodeLabel
      existing.iterSeq = v.iterSeq || 1
      existing.fromVisitUuid = v.fromVisitUuid || null
      existing.fromNodeId = v.fromNodeId || null
      // offCanvas：仅当 decisionJson 显式标记为 true（按画布走 / 起始节点 均为 false）
      existing.offCanvas = parseVisitDecision(v.decisionJson)?.offCanvas === true
      existing.variablesJson = v.variablesJson || '{}'
      existing.outputJson = v.outputJson || null
      existing.status = (v.status || existing.status).toLowerCase()
      existing.errorMsg = v.errorMsg || existing.errorMsg
      existing.sortOrder = v.sortOrder ?? existing.sortOrder
      // 同步 color / typeLabel（如果 nodeType 变了）
      const node = getNodes.value.find(n => n.id === v.nodeId)
      const info = nodeTypeInfo[existing.nodeType] || nodeTypeInfo[node?.type] || { color: '#666', typeLabel: '未知' }
      existing.color = info.color
      existing.typeLabel = info.typeLabel
    } else if (event.type === 'workflow_complete') {
      workflowResult.value = event.output
      showResult.value = true
      if (event.runId) {
        lastRunId.value = event.runId
      }
      runMeta.value = {
        runId: event.runId || null,
        workingDirFiles: Array.isArray(event.workingDirFiles) ? event.workingDirFiles : null,
      }
      lastRunWorkflowId.value = workflowIdRef.value
      message.success('工作流执行完成')
    } else if (event.type === 'workflow_error') {
      message.error('执行失败: ' + (event.error || '未知错误'))
    }
  }

  /**
   * v1.1：用户提交 Askfor 答案。
   */
  async function submitClarify(answer) {
    if (!pendingAsk.value || !lastRunId.value) return
    const { clarifyId } = pendingAsk.value
    try {
      const res = await submitClarifyAnswer(lastRunId.value, clarifyId, answer)
      const resolved = res?.resolved ?? res?.data?.resolved ?? false
      if (resolved) {
        pendingAsk.value = null
        message.success('已提交回答，工作流继续执行')
      } else {
        message.warning('回答未生效，可能 clarify 已过期')
        pendingAsk.value = null
      }
    } catch (e) {
      message.error('提交回答失败: ' + e.message)
    }
  }

  function closeResult() {
    showResult.value = false
    resetState()
    clearNodeStatusClasses()
    activeTab.value = 'props'
    pendingAsk.value = null
    lastEventId.value = null
    reconnecting.value = false
    if (pollTimer.value) {
      clearInterval(pollTimer.value)
      pollTimer.value = null
    }
    Object.keys(expandedLogs).forEach(k => delete expandedLogs[k])
  }

  function openRunDialog(inputFields, formData) {
    Object.keys(formData).forEach(k => delete formData[k])
    inputFields.value.forEach(f => { formData[f.name] = '' })
    showRunDialog.value = true
  }

  function cancelRun() {
    showRunDialog.value = false
  }

  function statusText(s) {
    return { running: '执行中', success: '成功', failed: '失败', skipped: '跳过', no_run: '未运行' }[s] || s
  }

  function logTypeLabel(type) {
    return sharedLogTypeLabel(type)
  }

  function formatTimestamp(ts) {
    if (!ts) return ''
    const d = new Date(ts)
    return d.toLocaleTimeString('zh-CN', { hour12: false })
  }

  function filterNonThinkingLogs(logs) {
    if (!Array.isArray(logs)) return logs
    return logs.filter(l => l.type !== 'THINKING')
  }

  function toggleContext(visitUuid) {
    expandedContext[visitUuid] = !expandedContext[visitUuid]
  }

  function toggleThinking(visitUuid) {
    expandedThinking[visitUuid] = !expandedThinking[visitUuid]
  }

  function toggleLogs(visitUuid) {
    expandedLogs[visitUuid] = !expandedLogs[visitUuid]
  }

  /**
   * 格式化 contextSnapshot / variablesJson 中的值：
   * - 对象/数组：JSON.stringify（缩进 2）
   * - 字符串：原样返回，超长由 CSS 控制最大高度
   * - 其他：String(value)
   */
  function formatContextValue(value) {
    if (value === null || value === undefined) return 'null'
    if (typeof value === 'object') return JSON.stringify(value, null, 2)
    return String(value)
  }

  return {
    running,
    showRunDialog,
    showResult,
    workflowResult,
    runMeta,
    nodeStatuses,
    activeTab,
    // v2.0：单一卡片列表（替代旧版 nodeResults + visits 双数组）
    visits,
    expandedContext,
    expandedThinking,
    expandedLogs,
    lastRunId,
    lastRunWorkflowId,
    lastEventId,
    reconnecting,
    pendingAsk,
    startRun,
    submitClarify,
    openRunDialog,
    cancelRun,
    closeResult,
    statusText,
    logTypeLabel,
    formatTimestamp,
    toggleContext,
    toggleThinking,
    toggleLogs,
    formatContextValue
  }
}