websocket-streaming

WebSocket ストリーミング

DAG 実行エンジンとダッシュボード間のリアルタイム双方向通信です。型付きイベントプロトコル、接続管理、React フックとの連携を扱います。

---

使用する場面

✅ 次の用途に使用します：

• 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 フック：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 が切断され、ダッシュボードに古いデータが永久に表示されます。 正しい方法：指数バックオフで再接続します（1 秒、2 秒、4 秒、8 秒……最大 30 秒）。再接続時に状態を再同期します。

イベントごとに状態全体を送信する

誤り：ノードが更新されるたびに、DAG の状態全体をブロードキャストします。 正しい方法：差分だけを送信します。どのノードが、どのステータスに変わったかを伝えます。クライアントはローカルストアに更新を適用します。

型付きプロトコルがない

誤り：型なしの JSON オブジェクトを送信し、any で解析します。 正しい方法：ServerEvent と ClientEvent のユニオン型を定義します。msg.type に対する switch 文で、すべてのケースを網羅します。