js-implementation.md 42 KB
JS/Vue 参考实现
可直接复用的运行展示参考实现：框架无关的核心层（SSE 客户端、状态容器、总控）+ Vue 3 视图层。接口契约见 external-workflow-api.md；展示规范（六面板、折叠策略、multi-visit）见 run-history-ui.md。
本实现以服务端实际行为为准，以下几点容易踩坑，已在代码中正确处理：
- 成功判断用
resp.code === 'OK'（字符串，不是数字 0）。 X-API-Key只支持请求头 → 不能用原生EventSource（无法自定义请求头），必须用fetch+ReadableStream手动解析 SSE。- 断线重连通过
Last-Event-ID请求头携带（不是 query 参数）。 - SSE 事件 payload 是扁平结构：
node_stream的增量在content字符串字段上，status是大写枚举（RUNNING/SUCCESS/FAILED/SKIPPED）。 tool_call/tool_result无 callId，按同 Visit 内到达顺序配对（Hermes 工具串行执行，安全）。- envelope 的
status是数字（200 成功 / 400 失败）；工作流最终输出同样带 envelope 壳。
模块总览
┌─────────────────────────────────────────────────────────┐
│ 视图层（Vue 3，可替换为任意框架） │
│ WorkflowRunPanel.vue（运行详情 + Visit 卡片列表） │
│ AskForPanel.vue（用户确认面板） │
│ MarkdownView.vue（markdown 渲染，DOMPurify 消毒） │
└──────────────▲──────────────────────────────────────────┘
│ 订阅变更（version 计数器触发重渲染）
┌──────────────┴──────────────────────────────────────────┐
│ VisitStore（状态容器，框架无关） │
│ 按 visitUuid 聚合事件 → 六面板数据（visit 对象） │
└──────────────▲──────────────────────────────────────────┘
│ onEvent(evt) / loadTerminalResult(run)
┌──────────────┴──────────────────────────────────────────┐
│ RunController（总控） │
│ 创建运行 → 启动 → 订阅 SSE；5 次重连失败降级轮询 │
│ ┌────────────────────────────────────────────┐ │
│ │ WorkflowStreamClient（fetch + ReadableStream│ │
│ │ 手动解析 SSE 帧，指数退避重连，Last-Event-ID）│ │
│ └────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
1. 配置与 REST 封装 agent-client.js
// ============ 运行时配置 ============
// baseUrl 与 apiKey 必须从环境变量/后端配置注入，不要硬编码进源码、不要提交到仓库。
// 推荐生产环境：业务方后端反代 /agent-api/* → agent-management /api/v1/* 并注入密钥，
// 前端走同源（此时 baseUrl 为空串、apiKey 由反代注入，可省略 X-API-Key）。
export const config = {
baseUrl: import.meta.env.VITE_AGENT_API_BASE || 'http://127.0.0.1:2438',
apiKey: import.meta.env.VITE_AGENT_API_KEY || '',
workflowName: 'military-analyze-agent', // kebab-case 唯一标识
};
export class ApiError extends Error {
constructor(message, code) {
super(message);
this.name = 'ApiError';
this.code = code; // 如 UNAUTHORIZED / WORKFLOW_NOT_FOUND / VALIDATION_FAILED
}
}
// ============ REST 请求封装 ============
// 统一注入 X-API-Key、解包 {code, message, data} 响应。
export async function apiRequest(path, options = {}) {
const headers = { 'X-API-Key': config.apiKey, ...(options.headers || {}) };
if (options.body !== undefined) headers['Content-Type'] = 'application/json';
const resp = await fetch(config.baseUrl + path, {
method: options.method || 'GET',
headers,
body: options.body !== undefined ? JSON.stringify(options.body) : undefined,
signal: options.signal,
});
if (!resp.ok) {
let code = 'HTTP_' + resp.status;
let message = resp.statusText;
try {
const err = await resp.json(); // 错误包：{code, message, data: null}
code = err.code || code;
message = err.message || message;
} catch (_) { /* 响应体不是 JSON */ }
throw new ApiError(message, code);
}
const json = await resp.json();
if (json.code !== 'OK') { // 注意：code 是字符串 'OK'
throw new ApiError(json.message || '请求失败', json.code);
}
return json.data; // 直接返回业务数据
}
// ============ 工作空间下载（二进制，不走 JSON 解析） ============
export async function downloadWorkspace(workflowName, runId) {
const resp = await fetch(
`${config.baseUrl}/api/v1/workflows/${workflowName}/runs/${runId}/workspace`,
{ headers: { 'X-API-Key': config.apiKey } },
);
if (!resp.ok) throw new ApiError('工作空间下载失败', 'HTTP_' + resp.status);
const blob = await resp.blob(); // 二进制 zip
const url = URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = `workspace-${runId}.zip`;
a.click();
URL.revokeObjectURL(url);
}
2. SSE 客户端 sse-client.js
import { apiRequest, config } from './agent-client.js';
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
/**
* 工作流 SSE 客户端：fetch + ReadableStream 手动解析。
*
* 为什么不用原生 EventSource：X-API-Key 只支持请求头，
* EventSource 无法自定义请求头（见 external-workflow-api.md「断线重连机制」）。
*
* 职责：
* - 解析标准 SSE 帧（event:/id:/data:，空行为帧边界，: 开头为心跳注释帧）
* - 断线后指数退避重连（1s/2s/4s/8s/16s，共 5 次），重连携带 Last-Event-ID 请求头
* - 5 次失败降级轮询 GET /runs/{runId}（30s/次），SSE 恢复后立即停止轮询
* - 收到 workflow_complete / workflow_error 终态事件后停止一切重连
*/
export class WorkflowStreamClient {
/**
* @param {{onEvent:function(Object),onStateChange:function(string,any):void}} handlers
* onEvent 收到已 JSON.parse 的事件对象（{type, runId, ...}）
* onStateChange 状态变化：'connected' | 'reconnecting'(delayMs) | 'polling' | 'closed'
*/
constructor({ onEvent, onStateChange }) {
this.onEvent = onEvent;
this.onStateChange = onStateChange || (() => {});
this.lastEventId = null; // 最近一条 SSE id，重连时作为 Last-Event-ID 请求头
this.stopped = false; // 用户主动停止
this.finished = false; // 已收到终态事件（workflow_complete / workflow_error）
this.sseAlive = false; // 当前是否有活跃 SSE 连接
this.pollTimer = null; // 降级轮询定时器
this.abort = null; // 当前连接的 AbortController
}
/** 订阅运行事件流。不传 lastEventId 时从头订阅（服务端补发全部缓存事件，实现完整回放）。 */
subscribe(workflowName, runId, { lastEventId = null } = {}) {
this.lastEventId = lastEventId;
this._loop(workflowName, runId);
}
/** 主动停止：中断连接 + 停止轮询。 */
stop() {
this.stopped = true;
this._stopPolling();
if (this.abort) this.abort.abort();
}
// ============ 内部：重连主循环 ============
async _loop(workflowName, runId) {
const backoff = [1000, 2000, 4000, 8000, 16000]; // 指数退避，5 次
let attempts = 0;
while (!this.stopped && !this.finished) {
try {
await this._open(workflowName, runId);
// _open 正常 return：流被服务端正常关闭（少见），同样进入重连
attempts = 0;
this.sseAlive = false;
} catch (err) {
this.sseAlive = false;
if (this.stopped || this.finished) return;
if (attempts >= backoff.length) {
this._degradeToPolling(workflowName, runId); // 5 次失败 → 轮询降级
return;
}
const delay = backoff[attempts++];
this.onStateChange('reconnecting', delay);
await sleep(delay);
}
}
}
// ============ 内部：打开连接并解析帧 ============
async _open(workflowName, runId) {
const headers = {
Accept: 'text/event-stream',
'X-API-Key': config.apiKey,
};
if (this.lastEventId !== null) {
headers['Last-Event-ID'] = String(this.lastEventId); // 续传：请求头，不是 query
}
this.abort = new AbortController();
const resp = await fetch(
`${config.baseUrl}/api/v1/workflows/${workflowName}/runs/${runId}/stream`,
{ headers, signal: this.abort.signal },
);
if (!resp.ok || !resp.body) {
// 404 可能是运行不存在；交给上层重连/降级逻辑
throw new Error('stream HTTP ' + resp.status);
}
this.onStateChange('connected', undefined);
this.sseAlive = true;
const reader = resp.body.getReader();
const decoder = new TextDecoder('utf-8');
let buffer = '';
let eventName = '';
let eventId = null;
let dataLines = [];
// 空行 = 帧边界：派发一帧
const dispatch = () => {
if (dataLines.length === 0) return; // 纯注释帧（:keep-alive 心跳，30s 一次）
const payloadText = dataLines.join('\n');
if (eventId !== null) this.lastEventId = eventId; // 记录最新 id 供重连
let payload;
try {
payload = JSON.parse(payloadText);
} catch (_) {
return; // 忽略无法解析的帧
}
// 终态事件：标记结束，停止一切重连
if (payload.type === 'workflow_complete' || payload.type === 'workflow_error') {
this.finished = true;
}
this.onEvent(payload);
eventName = '';
eventId = null;
dataLines = [];
};
while (true) {
const { done, value } = await reader.read();
if (done) return; // 流结束（服务端关闭）→ 上层重连
buffer += decoder.decode(value, { stream: true });
let nl;
while ((nl = buffer.indexOf('\n')) >= 0) {
const line = buffer.slice(0, nl).replace(/\r$/, ''); // 兼容 CRLF
buffer = buffer.slice(nl + 1);
if (line === '') {
dispatch();
} else if (line.startsWith(':')) {
continue; // 注释帧（心跳）
} else if (line.startsWith('event:')) {
eventName = line.slice(6).trim(); // 兼容 event:name 与 event: name
} else if (line.startsWith('id:')) {
eventId = line.slice(3).trim();
} else if (line.startsWith('data:')) {
const rest = line.slice(5);
dataLines.push(rest.startsWith(' ') ? rest.slice(1) : rest);
}
}
}
}
// ============ 内部：轮询降级 ============
_degradeToPolling(workflowName, runId) {
if (this.pollTimer || this.stopped || this.finished) return;
this.onStateChange('polling', undefined);
this.pollTimer = setInterval(async () => {
// SSE 已恢复 → 立即停止轮询
if (this.sseAlive) {
this._stopPolling();
return;
}
try {
const run = await apiRequest(
`/api/v1/workflows/${workflowName}/runs/${runId}`,
);
if (run.status === 'SUCCESS' || run.status === 'FAILED') {
this.finished = true; // 终态：停止轮询与重连
this._stopPolling();
// 用查询结果合成终态事件，由 store 幂等合并。
// outputs 带 envelope 壳，与 workflow_complete.output 行为一致；
// 时间/节点等更多细节可由视图层在终态后调 GET /runs/{runId} 补齐。
this.onEvent({
type: run.status === 'SUCCESS' ? 'workflow_complete' : 'workflow_error',
runId,
output: run.outputs ?? null,
error: run.error ?? null,
});
return;
}
} catch (_) { /* 网络仍不可用，下个周期重试 */ }
// 后台尝试恢复 SSE（恢复后事件按 visitUuid 幂等合并，不会产生重复卡片）
if (!this.sseAlive && !this.stopped && !this.finished) {
this._loop(workflowName, runId);
}
}, 30_000);
}
_stopPolling() {
if (this.pollTimer) {
clearInterval(this.pollTimer);
this.pollTimer = null;
}
}
}
3. Visit 状态容器 visit-store.js
框架无关。按 visitUuid 聚合事件（不能按 nodeId 合并，AGENTIC 模式同节点可多次访问），派生六面板数据。
// ============ 容错工具 ============
export function parseJson(str) {
if (str === null || str === undefined) return null;
if (typeof str === 'object') return str; // 已是对象
try {
return JSON.parse(str);
} catch (_) {
return null;
}
}
// 兼容 ISO 字符串与 Spring/Java 的数组式日期（LocalDateTime 序列化成 [2026,7,1,8,30,5]）
export function formatTime(t) {
if (t === null || t === undefined) return '';
if (Array.isArray(t)) {
const [y, mo, d, h = 0, mi = 0, s = 0] = t;
const p = (n) => String(n).padStart(2, '0');
return `${y}-${p(mo)}-${p(d)} ${p(h)}:${p(mi)}:${p(s)}`;
}
const d = new Date(t);
if (Number.isNaN(d.getTime())) return String(t);
return d.toLocaleString('zh-CN', { hour12: false });
}
export function durationText(startedAt, completedAt) {
if (!startedAt || !completedAt) return '';
const ms = new Date(completedAt) - new Date(startedAt);
if (Number.isNaN(ms) || ms < 0) return '';
const sec = Math.round(ms / 1000);
if (sec < 60) return `${sec}秒`;
if (sec < 3600) return `${Math.floor(sec / 60)}分${sec % 60}秒`;
return `${Math.floor(sec / 3600)}时${Math.floor((sec % 3600) / 60)}分`;
}
// ============ envelope 解包（status 是数字：200 成功 / 400 失败） ============
export function isEnvelope(o) {
return o && typeof o === 'object' && !Array.isArray(o)
&& ('status' in o) && ('data' in o); // 必须同时含 status 与 data
}
export function unwrapOutput(o) {
if (!isEnvelope(o)) return o;
return o.data ?? {};
}
export function isFailedEnvelope(o) {
return isEnvelope(o) && Number(o.status) === 400;
}
const TERMINAL_NODE_STATUS = new Set(['SUCCESS', 'FAILED', 'SKIPPED']);
/**
* Visit 状态容器。
* visit 对象（六面板 + 元数据）：
* { visitUuid, nodeId, nodeLabel, nodeType, iterSeq, fromNodeId, fromVisitUuid,
* offCanvas, sortOrder, status, // RUNNING/SUCCESS/FAILED/SKIPPED
* collapsed, manuallyToggled, // 折叠状态 + 用户手动标记
* liveStatus, // ① 实时状态（覆盖式；终态后渲染层隐藏）
* thinking, // ② 思考过程（追加式）
* todos, // ③ 待办清单（覆盖式；null 表示无）
* toolCalls, // ④ 工具调用（顺序配对）
* logs, // ⑤ 执行日志（追加式）
* output, error, // ⑥ 最终输出 / 失败原因
* clarify, clarifyAnswered } // AskFor 面板
*/
export class VisitStore {
constructor() {
this.visits = new Map(); // visitUuid → visit
this.version = 0; // 变更计数器，视图层据此触发重渲染
this.run = {
runId: null, status: 'CREATED', error: null,
finalOutput: null, workingDirFiles: null,
startedAt: null, completedAt: null, // SSE 终态事件不带时间，可在终态后从 GET /runs/{runId} 补齐
};
this.decisions = []; // planner_decision 决策链（AGENTIC 模式）
}
_bump() { this.version++; }
_ensureVisit(visitUuid, nodeId = null) {
let v = this.visits.get(visitUuid);
if (!v) { // 幂等：补发/重放场景不产生重复卡片
v = {
visitUuid, nodeId, nodeLabel: nodeId || visitUuid, nodeType: null,
iterSeq: 1, fromNodeId: null, fromVisitUuid: null, offCanvas: false,
sortOrder: null, status: 'RUNNING',
collapsed: false, manuallyToggled: false,
liveStatus: '', thinking: '', todos: null,
toolCalls: [], logs: [],
output: null, error: null,
clarify: null, clarifyAnswered: false,
};
this.visits.set(visitUuid, v);
}
return v;
}
/** 按展示顺序返回 visit 列表：有 sortOrder 的按其升序，其余按到达顺序排在其后 */
orderedVisits() {
const withOrder = [];
const withoutOrder = [];
let seq = 0;
for (const v of this.visits.values()) {
if (v.sortOrder !== null && v.sortOrder !== undefined) withOrder.push(v);
else { v._seq = seq++; withoutOrder.push(v); }
}
withOrder.sort((a, b) => a.sortOrder - b.sortOrder);
return [...withoutOrder, ...withOrder];
}
/** SSE 事件总入口（6 种 type，见 external-workflow-api.md） */
handleEvent(evt) {
if (!evt || !evt.type) return;
switch (evt.type) {
case 'node_status': this._onNodeStatus(evt); break;
case 'node_stream': this._onNodeStream(evt); break;
case 'visit_recorded': this._onVisitRecorded(evt); break;
case 'planner_decision': this._onPlannerDecision(evt); break;
case 'workflow_complete':
this.run.status = 'SUCCESS';
this.run.finalOutput = evt.output ?? null; // 带 envelope 壳，渲染前 unwrap
this.run.workingDirFiles = evt.workingDirFiles ?? null;
this._bump();
break;
case 'workflow_error':
this.run.status = 'FAILED';
this.run.error = evt.error ?? '工作流执行失败';
this._bump();
break;
}
}
// ---------- node_status：卡片创建 / 终态 ----------
_onNodeStatus(evt) {
const v = this._ensureVisit(evt.visitUuid, evt.nodeId);
if (evt.nodeId && !v.nodeId) v.nodeId = evt.nodeId;
v.status = evt.status; // 大写枚举
if (TERMINAL_NODE_STATUS.has(evt.status)) {
v.output = evt.output ?? null; // envelope：{status:200|400, message, data}
v.error = evt.error ?? v.error;
v.liveStatus = '';
// 自动折叠：用户手动操作过的卡片不自动折叠（手动优先）
if (!v.manuallyToggled) v.collapsed = true;
}
this._bump();
}
// ---------- node_stream：8 种 kind 分流到六面板 ----------
_onNodeStream(evt) {
if (!evt.visitUuid) return;
const v = this._ensureVisit(evt.visitUuid, evt.nodeId);
const content = evt.content ?? '';
switch (evt.kind) {
case 'thinking': // ② 思考过程：追加，markdown 渲染
v.thinking += content;
break;
case 'status': // ① 实时状态：覆盖式，只留最新一条；终态后渲染层隐藏
v.liveStatus = content;
break;
case 'step': { // ⑤ 执行日志：每轮 LLM 调用进度
const step = parseJson(content);
v.logs.push(`[第 ${step?.iteration ?? '?'} 轮] 已用工具: ${JSON.stringify(step?.prev_tools ?? [])}`);
break;
}
case 'tool_call': // ④ 工具调用开始（content 已截断 500 字符，仅作预览）
v.toolCalls.push({
tool: evt.toolName || '?', preview: content,
status: 'running', result: null,
});
v.logs.push(`调用工具 ${evt.toolName || '?'}`);
break;
case 'tool_result': { // ④ 工具调用结束：无 callId，按顺序关闭最早未关闭的同名调用
const open = v.toolCalls.find(
(t) => t.status === 'running' && t.tool === (evt.toolName || '?'),
);
if (open) {
open.status = 'done';
open.result = content; // 截断到 500 字符的摘要
}
v.logs.push(`工具完成 ${evt.toolName || '?'}`);
break;
}
case 'tool_progress': { // ④ 工具细粒度进度：更新运行中卡片的当前动作
const prog = parseJson(content);
const open = v.toolCalls.find(
(t) => t.status === 'running' && t.tool === evt.toolName,
);
if (open && prog) open.progress = `${prog.event_type ?? ''} ${prog.name ?? ''}`.trim();
break;
}
case 'todo_update': { // ③ 待办清单：整体覆盖（content 是 JSON 字符串）
const todos = parseJson(content);
if (Array.isArray(todos)) v.todos = todos;
break;
}
case 'clarify_request': { // AskFor：content 是 JSON 字符串（snake_case 键）
const clarify = parseJson(content);
if (clarify && clarify.clarify_id) {
v.clarify = {
clarifyId: clarify.clarify_id,
question: clarify.question ?? '',
choices: Array.isArray(clarify.choices) ? clarify.choices : [],
};
}
break;
}
}
this._bump();
}
// ---------- visit_recorded：补齐卡片元数据并按 sortOrder 重排 ----------
_onVisitRecorded(evt) {
const visit = evt.visit;
if (!visit || !visit.visitUuid) return;
const v = this._ensureVisit(visit.visitUuid, visit.nodeId);
v.nodeId = visit.nodeId ?? v.nodeId;
v.nodeLabel = visit.nodeLabel ?? v.nodeLabel;
v.nodeType = visit.nodeType ?? v.nodeType;
v.iterSeq = visit.iterSeq ?? 1;
v.fromNodeId = visit.fromNodeId ?? null;
v.fromVisitUuid = visit.fromVisitUuid ?? null;
v.offCanvas = visit.decisionJson === '{"offCanvas":true}';
v.sortOrder = visit.sortOrder ?? v.sortOrder;
if (visit.status) v.status = visit.status;
if (visit.errorMsg) v.error = visit.errorMsg;
this._bump();
}
// ---------- planner_decision：AI 决策链（可选渲染） ----------
_onPlannerDecision(evt) {
const d = evt.decision;
if (d && d.nextNodeId) {
this.decisions.push({ nextNodeId: d.nextNodeId, reason: d.reason ?? '' });
this._bump();
}
}
// ---------- 用户交互 ----------
toggleCollapse(visitUuid) {
const v = this.visits.get(visitUuid);
if (!v) return;
v.collapsed = !v.collapsed;
v.manuallyToggled = true; // 手动操作后不再自动折叠
this._bump();
}
markClarifyAnswered(visitUuid) {
const v = this.visits.get(visitUuid);
if (v) {
v.clarifyAnswered = true;
this._bump();
}
}
// ---------- 回放场景：从持久化结果构建（刷新页面的兜底路径） ----------
// 首选方案是 SSE 重放（subscribe 不带 Last-Event-ID，服务端补发全部缓存事件，
// 事件会自然流经 handleEvent，无需本方法）。仅当缓存过期（>25h）时用查询结果兜底：
loadTerminalResult(run) {
this.run.runId = run.runId;
this.run.status = run.status;
this.run.error = run.error ?? null;
this.run.finalOutput = run.outputs ?? null;
this.run.startedAt = run.startedAt ?? null;
this.run.completedAt = run.completedAt ?? null;
// DAG 模式：nodes[] 构建 Visit（每节点一张卡）；AGENTIC 模式 nodes 可能为空，
// 此时过程细节已不可恢复（只有 SSE 缓存内有），只展示最终输出。
for (const n of run.nodes ?? []) {
const key = n.nodeId; // DAG 模式节点只执行一次，用 nodeId 充当 visit 键
const v = this._ensureVisit(key, n.nodeId);
v.nodeLabel = n.label ?? n.nodeId;
v.nodeType = n.nodeType;
v.status = n.status;
v.output = n.output ?? null;
v.error = n.error ?? null;
v.collapsed = true; // 回放默认折叠
v.manuallyToggled = false;
}
this._bump();
}
// ---------- 回放场景：节点日志分流到六面板 ----------
// 日志元素 {type, message, detail, timestamp}，来自节点 logs 接口或 visit_recorded 的 visit.logsJson
loadVisitLogs(visitUuid, logs, nodeMeta = {}) {
const v = this._ensureVisit(visitUuid, nodeMeta.nodeId);
if (nodeMeta.nodeLabel) v.nodeLabel = nodeMeta.nodeLabel;
if (nodeMeta.nodeType) v.nodeType = nodeMeta.nodeType;
for (const log of logs ?? []) {
switch (log.type) {
case 'THINKING':
v.thinking += log.message ?? '';
break;
case 'TOOL_CALL':
v.toolCalls.push({ tool: log.message, preview: log.detail, status: 'running', result: null });
break;
case 'TOOL_RESULT': {
const open = v.toolCalls.find((t) => t.status === 'running' && t.tool === log.message);
if (open) { open.status = 'done'; open.result = log.detail; }
break;
}
case 'TODO_UPDATE': {
const todos = parseJson(log.detail);
if (Array.isArray(todos)) v.todos = todos;
break;
}
case 'STATUS':
case 'STEP':
case 'TOOL_PROGRESS':
case 'CLARIFY_REQUEST':
case 'ERROR':
v.logs.push(log.message ?? '');
break;
}
}
// 回放场景无实时状态面板
v.liveStatus = '';
this._bump();
}
}
4. 总控 run-controller.js
import { apiRequest, config } from './agent-client.js';
import { WorkflowStreamClient } from './sse-client.js';
import { VisitStore } from './visit-store.js';
/**
* 一次运行的生命周期管理：创建 → 启动 → 订阅 SSE →（降级轮询）→ 终态。
* 视图层只需订阅 store 的 version 变化 + 读取 store.orderedVisits()。
*/
export class RunController {
constructor() {
this.store = new VisitStore();
this.stream = null;
this.connectionState = 'idle'; // idle|connected|reconnecting|polling
}
/** 创建并启动一次运行，随后订阅事件流 */
async startWorkflow(variables = {}) {
// 1. 创建运行（async: true → 仅创建；响应含 runId 与 links）
const created = await apiRequest(
`/api/v1/workflows/${config.workflowName}/runs`,
{ method: 'POST', body: { variables, async: true } },
);
const runId = created.runId;
this.store.run.runId = runId;
// 2. 启动（运行启动信号由本接口的 HTTP 响应承担，SSE 中没有 run_started 事件）
await apiRequest(
`/api/v1/workflows/${config.workflowName}/runs/${runId}/start`,
{ method: 'POST' },
);
this.store.run.status = 'RUNNING';
// 3. 订阅 SSE
this._subscribe(runId);
return runId;
}
/** 页面刷新 / 查看历史运行：直接从头订阅，服务端补发全部缓存事件（完整回放） */
reconnectRun(runId) {
this.store.run.runId = runId;
this.store.run.status = 'RUNNING';
this._subscribe(runId);
}
_subscribe(runId) {
this.stream = new WorkflowStreamClient({
onEvent: (evt) => this.store.handleEvent(evt),
onStateChange: (state, delay) => {
this.connectionState = state;
// 视图层可据此展示横幅：「网络波动，正在恢复执行进度…」（reconnecting/polling）
},
});
this.stream.subscribe(config.workflowName, runId);
}
/** 组件卸载时调用 */
destroy() {
if (this.stream) this.stream.stop();
}
/**
* AskFor 提交（外部端点，v1.2 已开放）。
*
* 走 POST /api/v1/runs/{runId}/resume（路径不带 workflowName 段）。
* 同一 clarifyId 重复提交返回 409 CLARIFY_ALREADY_ANSWERED，按幂等约定视为成功。
*/
async submitAsk(runId, clarifyId, answer) {
try {
await apiRequest(`/api/v1/runs/${runId}/resume`, {
method: 'POST',
body: { clarifyId, answer },
});
// 响应 data: {accepted, runId, clarifyId}；accepted=false 表示已落库但 Bridge 唤醒失败
} catch (e) {
// 幂等：同一 clarifyId 重复提交返回 409，视为提交成功
if (e.code !== 'CLARIFY_ALREADY_ANSWERED') throw e;
}
// 成功后工作流自动继续，后续事件继续经 SSE 到达
}
/** AskFor 暂停点查询（外部端点，断线重连后恢复 AskFor UI 用） */
async fetchPauseState(runId) {
const state = await apiRequest(`/api/v1/runs/${runId}/state`);
// state: {runId, status, currentPause}；currentPause 为 null 表示无待答暂停点
// currentPause.choices 已是数组（区别于内部 API 的 JSON 字符串），无需解析
return state.currentPause || null;
}
}
5. markdown 渲染 markdown.js
// package.json 依赖（推荐版本）：
// marked ^12、marked-highlight ^2.2、highlight.js ^11.12、dompurify ^3.2、katex ^0.16（可选）
import { marked } from 'marked';
import { markedHighlight } from 'marked-highlight';
import hljs from 'highlight.js/lib/common'; // 常用语言子集，控制包体
import DOMPurify from 'dompurify';
marked.use(
markedHighlight({
langPrefix: 'hljs language-',
highlight(code, lang) {
const language = hljs.getLanguage(lang) ? lang : 'plaintext';
return hljs.highlight(code, { language }).value;
},
}),
);
// LLM 输出与用户输入都可能携带恶意脚本：渲染 HTML 前必须过 DOMPurify
export function renderMarkdown(src) {
const html = marked.parse(src || '');
return DOMPurify.sanitize(html, { ADD_ATTR: ['target'] });
}
配套 CSS 要点（代码块横向滚动不换行、限高滚动）：
.md-body pre { overflow-x: auto; }
.md-body pre code { white-space: pre; }
.md-body table { display: block; max-width: 100%; overflow-x: auto; }
6. Vue 组件
6.1 运行面板 WorkflowRunPanel.vue
六面板顺序固定：「实时状态 → 思考过程 → 待办清单 → 工具调用 → 执行日志」，最终输出在面板区之后单独渲染（始终可见，不随卡片折叠）。
<script setup>
import { computed, onMounted, onUnmounted, ref } from 'vue';
import { RunController } from '../core/run-controller.js';
import { downloadWorkspace } from '../core/agent-client.js';
import {
durationText, isFailedEnvelope, unwrapOutput,
} from '../core/visit-store.js';
import MarkdownView from './MarkdownView.vue';
import AskForPanel from './AskForPanel.vue';
const props = defineProps({
workflowName: { type: String, required: true },
variables: { type: Object, default: () => ({}) },
runId: { type: String, default: null }, // 传入则回放已有运行，否则新起运行
});
const ctrl = new RunController();
const tick = ref(0); // 订阅 store.version，手动触发重渲染
// 重连提示横幅（对应 RunController.connectionState）
const connBanner = computed(() => {
void tick.value;
if (ctrl.connectionState === 'reconnecting') return '网络波动，正在恢复执行进度…';
if (ctrl.connectionState === 'polling') return '网络不稳定，已切换为轮询模式（30 秒/次）…';
return '';
});
const visits = computed(() => {
void tick.value; // 依赖 tick，version 变化 → 重新取列表
return ctrl.store.orderedVisits();
});
const run = computed(() => {
void tick.value;
return ctrl.store.run;
});
onMounted(async () => {
// 简易轮询 version（也可在 store 内实现发布-订阅）
const timer = setInterval(() => {
if (tick.value !== ctrl.store.version) tick.value = ctrl.store.version;
}, 200);
onUnmounted(() => clearInterval(timer));
if (props.runId) {
ctrl.reconnectRun(props.runId); // 回放：从头订阅，服务端补发全部缓存事件
} else {
await ctrl.startWorkflow(props.variables); // 新运行：创建 + 启动 + 订阅
}
});
onUnmounted(() => ctrl.destroy());
// ---- 节点类型/状态映射（run-history-ui.md） ----
const NODE_LABELS = {
userInput: '用户输入', llm: '大模型', skill: '技能', agent: '智能体',
smartAction: '智能操作', condition: '条件', output: '输出',
};
const STATUS_TEXT = { running: '执行中', success: '成功', failed: '失败', skipped: '跳过' };
const normStatus = (s) => String(s || '').toLowerCase();
const statusText = (s) => STATUS_TEXT[normStatus(s)] ?? '未知';
const finalOutputData = computed(() => unwrapOutput(run.value.finalOutput));
const hasFinalOutput = computed(() => {
const d = finalOutputData.value;
return d !== null && d !== undefined && Object.keys(d ?? {}).length > 0;
});
</script>
<template>
<div class="run-panel">
<!-- 连接状态横幅（重连提示） -->
<div v-if="connBanner" class="conn-banner">{{ connBanner }}</div>
<!-- 运行头部 -->
<header class="run-header">
<span class="dot" :class="`status-${normStatus(run.status)}`" />
<span class="run-id">运行 #{{ run.runId }}</span>
<span class="badge" :class="`status-${normStatus(run.status)}`">
{{ statusText(run.status) }}
</span>
<span v-if="run.startedAt">
耗时 {{ durationText(run.startedAt, run.completedAt) }}
</span>
</header>
<div v-if="run.error" class="run-error">{{ run.error }}</div>
<!-- Visit 卡片列表（AGENTIC 模式同节点多卡，纵向堆叠） -->
<section v-for="v in visits" :key="v.visitUuid" class="visit-card"
:class="`status-${normStatus(v.status)}`">
<!-- 卡片头部：点击折叠/展开 -->
<header class="visit-head" @click="ctrl.store.toggleCollapse(v.visitUuid)">
<span class="badge" :class="`status-${normStatus(v.status)}`">
{{ statusText(v.status) }}
</span>
<span class="node-type">{{ NODE_LABELS[v.nodeType] ?? v.nodeType ?? '' }}</span>
<span class="node-label">{{ v.nodeLabel }}</span>
<span v-if="v.iterSeq > 1" class="chip">第 {{ v.iterSeq }} 次访问</span>
<span v-if="v.offCanvas && v.fromNodeId" class="chip">来自 {{ v.fromNodeId }}</span>
<span class="spacer" />
<span class="fold-arrow">{{ v.collapsed ? '▸' : '▾' }}</span>
</header>
<div v-show="!v.collapsed" class="visit-body">
<!-- ① 实时状态：覆盖式；节点结束后隐藏 -->
<p v-if="normStatus(v.status) === 'running' && v.liveStatus" class="live-status">
{{ v.liveStatus }}
</p>
<!-- ② 思考过程：追加式 markdown，限高内部滚动 -->
<details v-if="v.thinking" class="panel" open>
<summary>思考过程</summary>
<div class="md-body scroll-y">
<MarkdownView :source="v.thinking" />
</div>
</details>
<!-- ③ 待办清单：整体覆盖；无数据不渲染 -->
<div v-if="v.todos && v.todos.length" class="panel">
<div class="panel-title">待办清单</div>
<ul class="todo-list">
<li v-for="(t, i) in v.todos" :key="i" :data-status="t.status">
<span class="todo-icon">{{ t.status === 'completed' ? '✓'
: t.status === 'in_progress' ? '◐' : '○' }}</span>
{{ t.content }}
</li>
</ul>
</div>
<!-- ④ 工具调用：每调用一张小卡片；无调用不渲染 -->
<div v-if="v.toolCalls.length" class="panel">
<div class="panel-title">工具调用</div>
<div v-for="(t, i) in v.toolCalls" :key="i" class="tool-call"
:data-status="t.status">
<div class="tool-head">
<code>{{ t.tool }}</code>
<span>{{ t.status === 'running' ? '运行中' : '完成' }}</span>
</div>
<pre v-if="t.preview" class="tool-preview">{{ t.preview }}</pre>
<pre v-if="t.result" class="tool-result">{{ t.result }}</pre>
</div>
</div>
<!-- ⑤ 执行日志：追加式纯文本，与思考过程严格分开 -->
<details v-if="v.logs.length" class="panel">
<summary>执行日志（{{ v.logs.length }}）</summary>
<ol class="log-list scroll-y">
<li v-for="(line, i) in v.logs" :key="i">{{ line }}</li>
</ol>
</details>
</div>
<!-- ⑥ 最终输出：envelope 解包后渲染，始终展开（不随卡片折叠隐藏） -->
<div v-if="v.output" class="visit-output" :class="{ failed: isFailedEnvelope(v.output) }">
<div class="panel-title">
{{ isFailedEnvelope(v.output) ? '失败原因' : '最终结果' }}
</div>
<!-- 失败时优先展示 envelope.message；成功时渲染 unwrap 后的 data -->
<MarkdownView v-if="isFailedEnvelope(v.output)"
:source="v.output.message || v.error || ''" />
<MarkdownView v-else-if="typeof unwrapOutput(v.output) === 'string'"
:source="unwrapOutput(v.output)" />
<dl v-else class="kv">
<template v-for="(val, key) in unwrapOutput(v.output)" :key="key">
<dt>{{ key }}</dt>
<dd>
<MarkdownView v-if="typeof val === 'string'" :source="val" />
<pre v-else>{{ JSON.stringify(val, null, 2) }}</pre>
</dd>
</template>
</dl>
</div>
<div v-else-if="v.error" class="visit-output failed">
<div class="panel-title">失败原因</div>
<MarkdownView :source="v.error" />
</div>
<!-- AskFor 面板（clarify_request 到达时出现） -->
<AskForPanel v-if="v.clarify" :clarify="v.clarify" :answered="v.clarifyAnswered"
@submit="(answer) => ctrl.submitAsk(run.runId, v.clarify.clarifyId, answer)
.then(() => ctrl.store.markClarifyAnswered(v.visitUuid))" />
</section>
<!-- 工作流最终输出（workflow_complete.output，envelope 解包后渲染） -->
<section v-if="run.status === 'SUCCESS' && hasFinalOutput" class="final-output">
<div class="panel-title">最终结果</div>
<dl class="kv">
<template v-for="(val, key) in finalOutputData" :key="key">
<dt>{{ key }}</dt>
<dd>
<MarkdownView v-if="typeof val === 'string'" :source="val" />
<pre v-else>{{ JSON.stringify(val, null, 2) }}</pre>
</dd>
</template>
</dl>
</section>
<!-- 工作空间下载（成功终态后） -->
<footer v-if="run.status === 'SUCCESS' && run.workingDirFiles?.length" class="run-footer">
<button class="primary" @click="downloadWorkspace(props.workflowName, run.runId)">
下载工作空间（{{ run.workingDirFiles.length }} 个文件）
</button>
</footer>
</div>
</template>
6.2 AskFor 面板 AskForPanel.vue
<script setup>
import { ref } from 'vue';
const props = defineProps({
clarify: { type: Object, required: true }, // {clarifyId, question, choices}
answered: { type: Boolean, default: false },
});
const emit = defineEmits(['submit']);
const customText = ref('');
const submitting = ref(false);
async function submit(value) {
if (props.answered || submitting.value) return;
submitting.value = true;
try {
await emit('submit', value); // 由父组件调用 RunController.submitAsk
} finally {
submitting.value = false;
}
}
</script>
<template>
<div class="askfor-panel">
<div class="panel-title">需要您的确认</div>
<p class="question">{{ clarify.question }}</p>
<div class="choices">
<button v-for="c in clarify.choices" :key="c" :disabled="answered || submitting"
class="choice-btn" @click="submit(c)">
{{ c }}
</button>
</div>
<!-- 自定义输入：多行文本 -->
<textarea v-model="customText" rows="3" :disabled="answered || submitting"
placeholder="输入自定义回答... (Enter 提交，Shift+Enter 换行)" />
<button class="primary" :disabled="answered || submitting || !customText.trim()"
@click="submit(customText.trim())">
{{ answered ? '已提交' : '提交' }}
</button>
</div>
</template>
注意：
choices若为对象数组（如{value, label}），需在 store 解析处适配——本系统 clarify_request 的choices是字符串数组。另外emit('submit', value)不会返回 Promise，严谨写法是父组件传回调 prop 或使用defineExpose；参考实现以简洁为主，业务方按自身约定调整。
6.3 markdown 视图 MarkdownView.vue
<script setup>
import { computed } from 'vue';
import { renderMarkdown } from '../core/markdown.js';
const props = defineProps({ source: { type: String, default: '' } });
const html = computed(() => renderMarkdown(props.source));
</script>
<template>
<!-- 内容已过 DOMPurify 消毒 -->
<div class="md-body" v-html="html" />
</template>
7. 回放（历史运行）数据源
| 场景 | 路径 |
|---|---|
| 运行结束 ≤25h（SSE 缓存期内） | reconnectRun(runId) 从头订阅：服务端补发全部缓存事件，事件自然流经 handleEvent，卡片按 visitUuid 幂等重建，默认全部折叠（终态 node_status 自动折叠） |
| AGENTIC 模式 / 需要 Visit 粒度 | SSE 重放即可获得 visit_recorded；或从查询结果的 visit 数据（visit.logsJson）经 loadVisitLogs 分流到六面板 |
| 缓存过期（>25h / 服务重启） | GET /runs/{runId} → loadTerminalResult(run)；DAG 模式可再逐节点拉 logsUrl → loadVisitLogs 补全六面板 |
| 断线重连补发 | Last-Event-ID 请求头续传，补发事件按 visitUuid 幂等合并，不产生重复卡片 |
8. 集成要点清单
- 鉴权：所有请求带
X-API-Key请求头；不要把 key 暴露在前端源码，生产走后端反代。 - SSE：必须用
fetch+ReadableStream（EventSource无法带自定义请求头）；不要给 read 设置超时，依赖 30s 心跳 + 重连。 - 重连：指数退避 5 次（1s/2s/4s/8s/16s）→ 轮询降级（30s/次
GET /runs/{runId}）；workflow_complete/workflow_error后停止一切重连。 - 幂等：卡片以
visitUuid为主键；_ensureVisit的「不存在才创建」保证补发/重放不重复。 - 折叠：终态自动折叠、手动操作优先、最终输出始终可见；回放默认全折叠。
- 空面板不渲染：todos / toolCalls / thinking / logs 为空时整块不输出。
- 二进制下载：工作空间 zip 走 blob，不进 JSON 拦截器。
- AskFor：外部 API 已开放（v1.2）
POST /api/v1/runs/{runId}/resume+GET /api/v1/runs/{runId}/state（路径不带 workflowName 段）；409 重复提交按幂等约定视为成功。提交成功后 SSE 会继续推送后续事件，无需手动刷新。 - AGENTIC 模式：绝不能按
nodeId合并卡片；iterSeq > 1渲染「第 N 次访问」徽章；可选渲染planner_decision决策链。