websocket-streaming

WebSocket 流式传输

DAG 执行引擎与仪表板之间的实时双向通信。包含类型化事件协议、连接管理和 React Hook 集成。

---

适用场景

✅ 适用于：

• 将 DAG 节点状态变更流式传输到可视化仪表板
• 将人工审核节点的决策从仪表板发送到执行引擎
• 执行期间实时更新成本计数器和进度
• 双向通信（不只是服务端 → 客户端）

❌ 不适用于：

• 单向服务端 → 客户端更新（可考虑更简单的 SSE）
• REST API 设计（使用 api-architect）
• 基于轮询的状态检查（WebSocket 可替代轮询）

---

事件协议

服务端 → 客户端事件

type ServerEvent =
  | { type: 'node_state'; node_id: string; status: NodeStatus; output?: any; metrics?: NodeMetrics }
  | { type: 'edge_active'; from: string; to: string }
  | { type: 'dag_mutated'; mutation: DAGMutation }
  | { type: 'cost_update'; spent: number; budget: number; remaining: number }
  | { type: 'execution_complete'; results: Record<string, any> }
  | { type: 'human_gate_waiting'; node_id: string; presentation: GatePresentation }
  | { type: 'error'; node_id?: string; message: string };

客户端 → 服务端事件

type ClientEvent =
  | { type: 'human_decision'; node_id: string; decision: 'approve' | 'reject' | 'modify'; feedback?: string }
  | { type: 'pause_execution' }
  | { type: 'resume_execution' }
  | { type: 'cancel_execution' };

---

React Hook：useDAGStream

import { useEffect, useRef, useCallback } from 'react';

export function useDAGStream(dagId: string, store: DAGStore) {
  const wsRef = useRef<WebSocket | null>(null);
  const reconnectAttempt = useRef(0);

  const connect = useCallback(() => {
    const ws = new WebSocket(`/api/dags/${dagId}/stream`);

    ws.onopen = () => { reconnectAttempt.current = 0; };

    ws.onmessage = (event) => {
      const msg = JSON.parse(event.data) as ServerEvent;
      switch (msg.type) {
        case 'node_state':
          store.updateNodeData(msg.node_id, {
            status: msg.status, output: msg.output, metrics: msg.metrics,
          });
          break;
        case 'cost_update':
          store.setCostState({ spent: msg.spent, budget: msg.budget });
          break;
        case 'dag_mutated':
          store.applyMutation(msg.mutation);
          break;
        case 'execution_complete':
          store.setExecutionComplete(msg.results);
          break;
      }
    };

    ws.onclose = () => {
      // Reconnect with exponential backoff (max 30s)
      const delay = Math.min(1000 * 2 ** reconnectAttempt.current, 30000);
      reconnectAttempt.current++;
      setTimeout(connect, delay);
    };

    wsRef.current = ws;
  }, [dagId, store]);

  useEffect(() => { connect(); return () => wsRef.current?.close(); }, [connect]);

  // Send client events
  const send = useCallback((event: ClientEvent) => {
    wsRef.current?.send(JSON.stringify(event));
  }, []);

  return { send };
}

---

服务端实现（Node.js）

import { WebSocketServer } from 'ws';

const wss = new WebSocketServer({ noServer: true });

// Per-DAG rooms
const rooms = new Map<string, Set<WebSocket>>();

function broadcast(dagId: string, event: ServerEvent) {
  const clients = rooms.get(dagId);
  if (!clients) return;
  const msg = JSON.stringify(event);
  for (const ws of clients) {
    if (ws.readyState === ws.OPEN) ws.send(msg);
  }
}

// Usage in execution engine:
function onNodeComplete(dagId: string, nodeId: string, result: any) {
  broadcast(dagId, {
    type: 'node_state',
    node_id: nodeId,
    status: 'completed',
    output: result.output,
    metrics: result.metrics,
  });
}

---

反模式

没有重连逻辑

错误：WebSocket 关闭后，仪表板永远显示过时数据。 正确：使用指数退避进行重连（1s、2s、4s、8s……最长 30s）。重连时重新同步状态。

每次事件都发送完整状态

错误：每次节点更新时都广播整个 DAG 状态。 正确：仅发送增量：哪个节点发生了变化，以及变更后的状态。客户端将更新应用到本地存储中。

没有类型化协议

错误：发送无类型的 JSON 对象，并使用 any 解析。 正确：定义 ServerEvent 和 ClientEvent 联合类型。针对 msg.type 使用穷尽式 switch。