|
@@ -8,18 +8,19 @@ import java.util.concurrent.ConcurrentHashMap;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* 工作流执行上下文
|
|
* 工作流执行上下文
|
|
|
- * 携带节点间传递的变量、工作目录和累积节点输出
|
|
|
|
|
|
|
+ *
|
|
|
|
|
+ * <p>变量作用域采用节点命名空间隔离:每个节点的输出存于 {@code nodeScopedOutputs[nodeId]},
|
|
|
|
|
+ * 同名变量互不覆盖。后继节点通过 {@link NodeInputResolver} 的显式 mapping 或四级隐式回退
|
|
|
|
|
+ * 从前驱命名空间取值,参见 docs/workflow-variable-scope-design.md。</p>
|
|
|
*/
|
|
*/
|
|
|
@Slf4j
|
|
@Slf4j
|
|
|
public class WorkflowContext {
|
|
public class WorkflowContext {
|
|
|
|
|
|
|
|
private final Long workflowId;
|
|
private final Long workflowId;
|
|
|
private final String runId;
|
|
private final String runId;
|
|
|
- /** 扁平变量表,所有节点的输出变量合并于此(向后兼容 {{varName}} 模板渲染) */
|
|
|
|
|
- private final Map<String, Object> variables = new ConcurrentHashMap<>();
|
|
|
|
|
/**
|
|
/**
|
|
|
* 节点级隔离输出表:nodeId -> {varName -> value}。
|
|
* 节点级隔离输出表:nodeId -> {varName -> value}。
|
|
|
- * 用于实现变量作用域隔离,避免不同分支的同名输出相互覆盖。
|
|
|
|
|
|
|
+ * 同名变量按 nodeId 隔离,互不覆盖。
|
|
|
*/
|
|
*/
|
|
|
private final Map<String, Map<String, Object>> nodeScopedOutputs = new ConcurrentHashMap<>();
|
|
private final Map<String, Map<String, Object>> nodeScopedOutputs = new ConcurrentHashMap<>();
|
|
|
/** 初始运行输入,与节点输出隔离保存 */
|
|
/** 初始运行输入,与节点输出隔离保存 */
|
|
@@ -44,14 +45,10 @@ public class WorkflowContext {
|
|
|
this.runId = runId;
|
|
this.runId = runId;
|
|
|
this.workingDir = workingDir;
|
|
this.workingDir = workingDir;
|
|
|
this.initialInputs = initialInputs == null ? Map.of() : new LinkedHashMap<>(initialInputs);
|
|
this.initialInputs = initialInputs == null ? Map.of() : new LinkedHashMap<>(initialInputs);
|
|
|
- if (initialInputs != null) {
|
|
|
|
|
- this.variables.putAll(initialInputs);
|
|
|
|
|
- }
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
public Long getWorkflowId() { return workflowId; }
|
|
public Long getWorkflowId() { return workflowId; }
|
|
|
public String getRunId() { return runId; }
|
|
public String getRunId() { return runId; }
|
|
|
- public Map<String, Object> getVariables() { return variables; }
|
|
|
|
|
public Path getWorkingDir() { return workingDir; }
|
|
public Path getWorkingDir() { return workingDir; }
|
|
|
public List<NodeOutput> getNodeOutputs() { return Collections.unmodifiableList(nodeOutputs); }
|
|
public List<NodeOutput> getNodeOutputs() { return Collections.unmodifiableList(nodeOutputs); }
|
|
|
public Map<String, Object> getSections() { return sections; }
|
|
public Map<String, Object> getSections() { return sections; }
|
|
@@ -67,28 +64,13 @@ public class WorkflowContext {
|
|
|
return Collections.unmodifiableMap(initialInputs);
|
|
return Collections.unmodifiableMap(initialInputs);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- public void setVariable(String key, Object value) {
|
|
|
|
|
- variables.put(key, value);
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- public Object getVariable(String key) {
|
|
|
|
|
- return variables.get(key);
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 记录节点输出:同时更新扁平变量、节点级隔离输出和累积输出列表。
|
|
|
|
|
- * 变量覆盖检测:若 key 已存在(被上游节点写过),记录 WARN 日志,便于 debug。
|
|
|
|
|
|
|
+ * 记录节点输出:写入节点级隔离输出表与累积输出列表。
|
|
|
|
|
+ * 不再维护扁平变量表,同名输出按 nodeId 隔离,互不覆盖。
|
|
|
*/
|
|
*/
|
|
|
public void setNodeOutput(String nodeId, Map<String, Object> output) {
|
|
public void setNodeOutput(String nodeId, Map<String, Object> output) {
|
|
|
if (output != null) {
|
|
if (output != null) {
|
|
|
setNodeScopedOutput(nodeId, output);
|
|
setNodeScopedOutput(nodeId, output);
|
|
|
- for (Map.Entry<String, Object> e : output.entrySet()) {
|
|
|
|
|
- Object existing = variables.put(e.getKey(), e.getValue());
|
|
|
|
|
- if (existing != null) {
|
|
|
|
|
- log.warn("[WorkflowContext] 变量 {} 被节点 {} 覆盖(旧值类型={})",
|
|
|
|
|
- e.getKey(), nodeId, existing.getClass().getSimpleName());
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
nodeOutputs.add(new NodeOutput(nodeId, output));
|
|
nodeOutputs.add(new NodeOutput(nodeId, output));
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
@@ -123,9 +105,18 @@ public class WorkflowContext {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 计算当前 variables 的全量快照(深拷贝),用于 debug 视图。
|
|
|
|
|
|
|
+ * 全量快照(debug 视图与运行历史回放)。
|
|
|
|
|
+ * 格式:初始输入(扁平形式) + 各节点命名空间输出({nodeId}__{varName} 形式)。
|
|
|
*/
|
|
*/
|
|
|
- public Map<String, Object> snapshotVariables() {
|
|
|
|
|
- return new LinkedHashMap<>(variables);
|
|
|
|
|
|
|
+ public Map<String, Object> snapshotAllOutputs() {
|
|
|
|
|
+ Map<String, Object> snapshot = new LinkedHashMap<>(initialInputs);
|
|
|
|
|
+ for (Map.Entry<String, Map<String, Object>> nodeEntry : nodeScopedOutputs.entrySet()) {
|
|
|
|
|
+ String nodeId = nodeEntry.getKey();
|
|
|
|
|
+ for (Map.Entry<String, Object> varEntry : nodeEntry.getValue().entrySet()) {
|
|
|
|
|
+ snapshot.put(nodeId + NodeWorkspaceBuilder.NODE_FIELD_SEPARATOR + varEntry.getKey(),
|
|
|
|
|
+ varEntry.getValue());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return snapshot;
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|