从 0 到 1 打造 AI 提示流编排器：条件分支路由、Kahn 图剪枝与 HTTP 节点实战（开源系列 06）

// src/engine/browser-engine.ts (核心规则评估)
export function evaluateCondition(rule: ConditionRule, actualVal: unknown): boolean {
  const op = rule.operator;
  const targetVal = rule.value;

  switch (op) {
    case 'equals':
      return String(actualVal ?? '').trim() === String(targetVal ?? '').trim();
    case 'not_equals':
      return String(actualVal ?? '').trim() !== String(targetVal ?? '').trim();
    case 'contains':
      return String(actualVal ?? '').includes(String(targetVal ?? ''));
    case 'not_contains':
      return !String(actualVal ?? '').includes(String(targetVal ?? ''));
    case 'greater_than':
      return Number(actualVal) > Number(targetVal);
    case 'less_than':
      return Number(actualVal) < Number(targetVal);
    case 'is_empty':
      return actualVal === null || actualVal === undefined || String(actualVal).trim() === '';
    case 'is_not_empty':
      return actualVal !== null && actualVal !== undefined && String(actualVal).trim() !== '';
    case 'regex_match':
      try {
        return new RegExp(String(targetVal)).test(String(actualVal ?? ''));
      } catch {
        return false;
      }
    default:
      return false;
  }
}

// src/engine/browser-engine.ts (分层波次中自上而下的跳过判定)
const incoming = incomingEdgesMap.get(node.id) || [];
let shouldSkip = skippedNodes.has(node.id);

if (!shouldSkip && incoming.length > 0) {
  if (node.data.type === 'aggregator') {
    // 聚合器节点特殊规约：只要有至少一条入度边未被跳过，节点即可正常激活！
    const hasActiveIncoming = incoming.some((edge) => {
      if (skippedNodes.has(edge.source)) return false;
      const srcNode = nodeMap.get(edge.source);
      if (srcNode && srcNode.data.type === 'condition') {
        const activeBranch = nodeActiveBranch.get(edge.source);
        if (activeBranch && edge.sourceHandle && edge.sourceHandle !== activeBranch) {
          return false; // 非命中分支的 Handle 视为 Inactive
        }
      }
      return true;
    });
    if (!hasActiveIncoming) shouldSkip = true;
  } else {
    // 普通节点严格规约：只要存在来自未命中分支或已跳过节点的边，整节点跳过
    const isAnyIncomingInactive = incoming.some((edge) => {
      if (skippedNodes.has(edge.source)) return true;
      const srcNode = nodeMap.get(edge.source);
      if (srcNode && srcNode.data.type === 'condition') {
        const activeBranch = nodeActiveBranch.get(edge.source);
        if (activeBranch && edge.sourceHandle && edge.sourceHandle !== activeBranch) {
          return true;
        }
      }
      return false;
    });
    if (isAnyIncomingInactive) shouldSkip = true;
  }
}

if (shouldSkip) {
  skippedNodes.add(node.id); // 记录跳过节点，自动向下游穿透传播
  eventQueue.push({
    type: 'NODE_SKIPPED',
    payload: { nodeId: node.id, reason: 'Condition branch not matched or upstream node was skipped' },
  });
  return { nodeId: node.id, status: 'success', output: {}, durationMs: 0 };
}

// src/engine/browser-engine.ts (聚合节点的多模式提取)
case 'aggregator': {
  const config = (node.data.config || {}) as unknown as AggregatorNodeConfig;
  const mode = config.mode || 'first_available';
  let aggregatedValue: unknown = null;

  if (mode === 'first_available') {
    // 模式 1：提取首个未被跳过且产生有效输出的分支数据（互斥分支合并的黄金模式）
    for (const edge of incomingEdges) {
      if (!skippedNodes.has(edge.source) && context[edge.source]) {
        const srcOut = context[edge.source]!;
        aggregatedValue = srcOut['promptText'] ?? srcOut['output'] ?? srcOut['result'] ?? srcOut;
        break;
      }
    }
  } else if (mode === 'merge_all') {
    // 模式 2：将所有活跃上游节点的输出汇聚为一个以 nodeId 为 Key 的统一对象字典
    const merged: Record<string, unknown> = {};
    for (const edge of incomingEdges) {
      if (!skippedNodes.has(edge.source) && context[edge.source]) {
        merged[edge.source] = context[edge.source]!;
      }
    }
    aggregatedValue = merged;
  } else if (mode === 'wait_all') {
    // 模式 3：等待所有分支，若某个分支被跳过，则显式填充为 null 保留键位对齐
    const merged: Record<string, unknown> = {};
    for (const edge of incomingEdges) {
      merged[edge.source] = skippedNodes.has(edge.source) ? null : context[edge.source] ?? null;
    }
    aggregatedValue = merged;
  }

  output = { [config.outputKey || 'result']: aggregatedValue };
  break;
}

// src/engine/browser-engine.ts (HTTP 执行器安全校验与退避循环)
// 1. 安全防线：强制校验协议白名单，阻断恶意伪协议
if (rawUrl.startsWith('file:') || rawUrl.startsWith('javascript:') || rawUrl.startsWith('data:')) {
  throw new Error(`Security Exception: Forbidden or unsafe URL protocol "${rawUrl}"`);
}

// 2. 构造鉴权头 (Bearer / Basic / Custom API-Key)
if (authType === 'bearer' && authConfig.token) {
  reqHeaders['Authorization'] = `Bearer ${authConfig.token}`;
} else if (authType === 'basic' && (authConfig.username || authConfig.password)) {
  reqHeaders['Authorization'] = `Basic ${btoa(`${authConfig.username}:${authConfig.password}`)}`;
}

// 3. 面对瞬态故障的指数退避重试 (Exponential Backoff)
const retryConfig = config.retryConfig || { maxRetries: 1, retryDelayMs: 1000, retryOn: [500, 502, 503, 504] };
let attempt = 0;

while (attempt <= retryConfig.maxRetries) {
  if (signal.aborted) throw new Error('Workflow execution aborted by user.');

  try {
    const timeoutCtrl = new AbortController();
    const timeoutId = setTimeout(() => timeoutCtrl.abort(), config.timeout || 30000);

    const res = await fetch(targetUrl, { method, headers: reqHeaders, body, signal });
    clearTimeout(timeoutId);

    // 命中 5xx 临时错误码且未达到最大重试次数时，执行指数级退避延迟
    if (!res.ok && retryConfig.retryOn.includes(res.status) && attempt < retryConfig.maxRetries) {
      attempt++;
      const delay = retryConfig.retryDelayMs * Math.pow(2, attempt - 1); // 1s -> 2s -> 4s...
      await new Promise((r) => setTimeout(r, delay));
      continue;
    }

    const contentType = res.headers.get('content-type') || '';
    const responseData = contentType.includes('application/json') ? await res.json() : await res.text();
    output = { status: res.status, data: responseData, headers: respHeaders };
    break;
  } catch (err) {
    if (attempt >= retryConfig.maxRetries) throw err;
    attempt++;
    await new Promise((r) => setTimeout(r, retryConfig.retryDelayMs * Math.pow(2, attempt - 1)));
  }
}