企业级 AI Agent“反应式重规划与动态执行图剪枝”（Reactive Re-Planning & Dynamic DAG Mutation）架构实战：破解长调用链路外部状态突变、局部执行图动态剪枝与增量热替换

interface MutationEvent {
  // 触发故障的节点唯一标识
  failedNodeId: string;
  // 突变错误类别：业务前置不满足、外部资源枯竭、超时降级或人工驳回
  reason: 'PRECONDITION_FAILED' | 'RESOURCE_EXHAUSTED' | 'EXTERNAL_REJECTION' | 'SAFETY_GUARD_TRIPPED';
  // 外部环境增量快照数据（供局部重规划使用）
  environmentDelta: Record;
  // 故障节点原定期望输出的数据槽位清单
  expectedOutputKeys: string[];
  // 时间戳与上下文追踪 ID
  timestamp: number;
  traceId: string;
}

遍历队列 Q = [failedNodeId]
受损节点集 PrunedSet = {}

While Q 非空:
    currentNode = Q.pop()
    For each targetNode in currentNode.outgoingEdges:
        If targetNode.status != 'COMMITTED':
            PrunedSet.add(targetNode)
            Q.push(targetNode)
            // 取消正在挂起等待的 Promise 调度

/**
 * 企业级 AI Agent 反应式动态 DAG 重规划核心引擎
 * 实现环境状态守卫、前向可达性剪枝、增量步骤热替换与数据流重绑定
 */

// 节点执行状态枚举
export type NodeStatus = 'PENDING' | 'RUNNING' | 'COMMITTED' | 'FAILED' | 'PRUNED';

// 反应式突变原因枚举
export type MutationReason =
  | 'PRECONDITION_FAILED'
  | 'RESOURCE_EXHAUSTED'
  | 'ENVIRONMENT_STATE_DRIFT'
  | 'BUSINESS_GUARD_REJECTED';

// DAG 节点定义接口
export interface DagNode {
  id: string;                                   // 节点唯一标识
  name: string;                                 // 步骤业务名称
  dependencies: string[];                       // 依赖的前序节点 ID 列表
  inputSlots: Record;           // 数据流槽位绑定：[入参字段名] -> "[提供节点ID].[出参字段名]"
  outputKeys: string[];                         // 本节点产出的字段名称列表
  preGuard?: (context: ExecutionContext) => Promise; // 前置环境状态守卫
  postGuard?: (output: Record) => Promise; // 后置业务不变式守卫
  executor: (inputs: Record) => Promise>; // 业务工具调用执行器
  status: NodeStatus;                           // 当前执行状态
  result?: Record;             // 执行输出结果存储
}

// 全局执行上下文
export interface ExecutionContext {
  globalOutputs: Map>; // 已提交节点的输出字典
  environmentState: Record;           // 外部环境即时快照
}

// 动态 DAG 反应式重规划引擎
export class ReactiveDagEngine {
  private nodes: Map = new Map();
  private context: ExecutionContext = {
    globalOutputs: new Map(),
    environmentState: {}
  };

  /**
   * 注册节点到初始执行图
   * @param node 待加入的 DAG 节点
   */
  public registerNode(node: DagNode): void {
    this.nodes.set(node.id, { ...node, status: 'PENDING' });
  }

  /**
   * 更新外部环境状态快照
   * @param state 外部系统状态增量
   */
  public updateEnvironmentState(state: Record): void {
    this.context.environmentState = { ...this.context.environmentState, ...state };
  }

  /**
   * 驱动执行完整的 DAG
   */
  public async execute(): Promise>> {
    while (true) {
      // 1. 查找所有依赖均已 COMMITTED 且自身处于 PENDING 的可执行就绪节点
      const readyNodes = Array.from(this.nodes.values()).filter(node => {
        if (node.status !== 'PENDING') return false;
        return node.dependencies.every(depId => {
          const dep = this.nodes.get(depId);
          return dep && dep.status === 'COMMITTED';
        });
      });

      // 如果没有就绪节点且无运行节点，检查是否所有非剪枝节点均已完成
      if (readyNodes.length === 0) {
        const remainingPending = Array.from(this.nodes.values()).filter(n => n.status === 'PENDING' || n.status === 'RUNNING');
        if (remainingPending.length === 0) {
          break; // 全图正常执行完毕
        }
        throw new Error('DAG 执行停滞：存在无法解析依赖的死锁节点');
      }

      // 2. 依次并发调度就绪节点
      for (const node of readyNodes) {
        node.status = 'RUNNING';
        try {
          // 执行前置环境状态守卫检查
          if (node.preGuard) {
            const guardPass = await node.preGuard(this.context);
            if (!guardPass) {
              await this.handleReactiveMutation(node.id, 'PRECONDITION_FAILED', '前置环境守卫断言不满足');
              break; // 触发局部变异后退出当前内层循环，重新检索就绪节点
            }
          }

          // 解析数据流槽位输入
          const resolvedInputs = this.resolveInputSlots(node);

          // 执行核心业务工具调用
          const output = await node.executor(resolvedInputs);

          // 执行后置业务不变式守卫检查
          if (node.postGuard) {
            const invariantPass = await node.postGuard(output);
            if (!invariantPass) {
              await this.handleReactiveMutation(node.id, 'BUSINESS_GUARD_REJECTED', '后置业务不变式校验失败');
              break;
            }
          }

          // 节点成功完成，标记为已提交不可变
          node.status = 'COMMITTED';
          node.result = output;
          this.context.globalOutputs.set(node.id, output);
        } catch (error) {
          // 捕获不可预见的底层系统异常并触发反应式突变
          await this.handleReactiveMutation(node.id, 'ENVIRONMENT_STATE_DRIFT', (error as Error).message);
          break;
        }
      }
    }

    return this.context.globalOutputs;
  }

  /**
   * 核心算法：反应式局部子图剪枝与增量热替换处理
   */
  private async handleReactiveMutation(failedNodeId: string, reason: MutationReason, detail: string): Promise {
    const failedNode = this.nodes.get(failedNodeId);
    if (!failedNode) return;
    failedNode.status = 'FAILED';

    console.warn(`[反应式重规划] 节点 [${failedNodeId}] 触发环境突变！原因: ${reason}, 详情: ${detail}`);

    // 1. 前向可达性分析：收集所有依赖该失败节点的下游受损节点
    const prunedNodeIds = new Set();
    const queue: string[] = [failedNodeId];

    while (queue.length > 0) {
      const currentId = queue.shift()!;
      for (const [id, node] of this.nodes.entries()) {
        if (node.dependencies.includes(currentId) && node.status !== 'COMMITTED') {
          if (!prunedNodeIds.has(id)) {
            prunedNodeIds.add(id);
            queue.push(id);
          }
        }
      }
    }

    // 2. 将受损节点状态标记为 PRUNED（剪枝作废）
    for (const pruneId of prunedNodeIds) {
      const targetNode = this.nodes.get(pruneId);
      if (targetNode) {
        targetNode.status = 'PRUNED';
        console.warn(`[局部剪枝] 已精确剪枝下游受损节点: [${pruneId}]`);
      }
    }

    // 3. 模拟局部决策模型（Sub-DAG Planner）生成增量替代补丁
    // 仅基于局部故障上下文生成，替代失败节点的契约输出
    const replacementNodes = await this.synthesizePatchNodes(failedNode, reason);

    // 4. 将替代节点热嫁接（Hot-Splice）进图结构中
    for (const patchNode of replacementNodes) {
      this.nodes.set(patchNode.id, patchNode);
      console.info(`[热替换嫁接] 成功注入增量补丁节点: [${patchNode.id}]`);
    }

    // 5. 数据流槽位重绑定：更新被剪枝节点原下游的输入指向
    const terminalPatchNode = replacementNodes[replacementNodes.length - 1];
    for (const [id, node] of this.nodes.entries()) {
      if (node.status === 'PENDING') {
        // 如果该就绪节点原来依赖失败节点，重绑定其依赖到补丁链终端节点
        if (node.dependencies.includes(failedNodeId)) {
          node.dependencies = node.dependencies.map(d => d === failedNodeId ? terminalPatchNode.id : d);
        }
        // 重写输入槽位映射
        for (const [paramKey, slotPath] of Object.entries(node.inputSlots)) {
          if (slotPath.startsWith(`${failedNodeId}.`)) {
            const field = slotPath.split('.')[1];
            node.inputSlots[paramKey] = `${terminalPatchNode.id}.${field}`;
            console.info(`[槽位重绑定] 节点 [${id}] 参数 [${paramKey}] 已重映射至: [${terminalPatchNode.id}.${field}]`);
          }
        }
      }
    }
  }

  /**
   * 模拟局部决策大模型生成拓扑替代补偿子图
   */
  private async synthesizePatchNodes(failedNode: DagNode, reason: MutationReason): Promise {
    // 在真实生产环境中，此处将调用轻量 LLM 进行 300 Token 的局部决策
    const patchId = `patch_${failedNode.id}_alt`;
    return [
      {
        id: patchId,
        name: `应急替代工序:针对[${failedNode.name}]自愈执行`,
        dependencies: [...failedNode.dependencies],
        inputSlots: { ...failedNode.inputSlots },
        outputKeys: [...failedNode.outputKeys],
        status: 'PENDING',
        executor: async (inputs: Record) => {
          // 执行备用机床调度与补偿逻辑
          return {
            allocatedStationId: 'CNC_BACKUP_02',
            stationAuthToken: 'AUTH_HOT_SWAP_9921',
            status: 'HOT_SWAPPED_OK'
          };
        }
      }
    ];
  }

  /**
   * 根据槽位配置从已提交节点输出中解析实际输入参数
   */
  private resolveInputSlots(node: DagNode): Record {
    const resolved: Record = {};
    for (const [paramKey, slotPath] of Object.entries(node.inputSlots)) {
      const [depNodeId, outputField] = slotPath.split('.');
      const depOutput = this.context.globalOutputs.get(depNodeId);
      if (!depOutput || !(outputField in depOutput)) {
        throw new Error(`节点 [${node.id}] 数据流解析失败: 依赖输出 [${slotPath}] 不存在`);
      }
      resolved[paramKey] = depOutput[outputField];
    }
    return resolved;
  }
}