CMDCOrchestrator.Execution (cmdc_orchestrator v0.5.0)

Executor 与 dry_run 共用的信号分支、上下文合并工具。

Summary

Functions

edge_active?(edge, results, pruned)

判断一条边是否激活。

edge_signal(edge)

读取 edge 的 signal/when/branch 匹配值。

get_path(data, path)

从 context / dep_results 读取 dotted path。

incoming_edges(dag, node_id)

返回节点依赖边。

layer_signals(layer_results)

把当前层结果转换成 signal history。

merge_layer_outputs(context, dag, layer_results, ordered_ids)

按一个执行层的结果合并 context。

merge_node_output(context, node, output)

按 output_key 或 node_id 合并节点输出到运行上下文。

node_active?(node_id, dag, results, pruned)

判断节点是否应在当前 results/pruned 下激活。

signal(output)

从节点输出推导 signal。

Functions

edge_active?(edge, results, pruned)

@spec edge_active?(map(), map(), MapSet.t()) :: boolean()

判断一条边是否激活。

edge_signal(edge)

@spec edge_signal(map()) :: String.t() | nil

读取 edge 的 signal/when/branch 匹配值。

get_path(data, path)

@spec get_path(map(), [String.t()]) :: term()

从 context / dep_results 读取 dotted path。

incoming_edges(dag, node_id)

@spec incoming_edges(CMDCOrchestrator.DAG.t(), String.t()) :: [map()]

返回节点依赖边。

layer_signals(layer_results)

@spec layer_signals(map()) :: map()

把当前层结果转换成 signal history。

merge_layer_outputs(context, dag, layer_results, ordered_ids)

@spec merge_layer_outputs(map(), CMDCOrchestrator.DAG.t(), map(), [String.t()]) ::
  map()

按一个执行层的结果合并 context。

merge_node_output(context, node, output)

@spec merge_node_output(map(), map(), term()) :: map()

按 output_key 或 node_id 合并节点输出到运行上下文。

node_active?(node_id, dag, results, pruned)

@spec node_active?(String.t(), CMDCOrchestrator.DAG.t(), map(), MapSet.t()) ::
  boolean()

判断节点是否应在当前 results/pruned 下激活。

signal(output)

@spec signal(term()) :: String.t()

从节点输出推导 signal。