Struct ExecutionEngine 

pub struct ExecutionEngine<'a, S: WorkflowState> { /* private fields */ }

执行引擎 — 借用 State，持有 Mutation 缓冲、流发射器等运行时资源。

不对节点开发者公开。节点通过 NodeContext 或 LeafContext 能力视图交互。

状态所有权

Engine 借用 &'a mut S，不拥有 State。调用方在 Engine 生命周期外持有所有权。 这使得 Subgraph 组合成为可能 — 外层 Engine 借用 Outer，内层 Engine 借用 Inner。

Sink 注入

所有高级能力通过 Sink 注入：

• stream — 数据面流式输出
• checkpoint — 恢复边界通知
• barrier — Barrier 等待 + 决策注入

三层 API

• Leaf Execution API: build_leaf_context() — 构建只读 + emit 视图
• Composite Execution API: clone_state(), replace_state() — 执行控制
• Runtime Control Plane: stream, cancel, commit() — 运行时管理

Implementations

impl<'a, S: WorkflowState> ExecutionEngine<'a, S>

pub fn new( state: &'a mut S, stream: Option<Arc<dyn StreamSink>>, cancel: CancellationToken, checkpoint: Option<&'a mut dyn CheckpointSink<S>>, barrier: Option<&'a mut dyn BarrierSink>, ) -> Self

创建新的 ExecutionEngine。

Engine 借用 state，不拥有它。调用方在 Engine 外保持所有权。

• checkpoint — 可选的 Checkpoint Sink（None = 不需要自动 checkpoint）
• barrier — 可选的 Barrier Sink（None = 遇到 Barrier 直接 Approve）

pub fn take_mutations(&mut self) -> Vec<S::Mutation>

消费 Mutation 缓冲（Executor 调用）。

pub fn take_control(&mut self) -> (NextAction, Option<ExecutionSignal>)

消费控制信号（Executor 调用）。

pub fn take_metadata(&mut self) -> NodeMetadata

获取元数据（Executor 调用）。

pub fn state(&self) -> &S

获取状态引用。

pub fn stream(&self) -> Option<&dyn StreamSink>

获取数据面发射器引用。

pub fn cancel_token(&self) -> &CancellationToken

获取取消令牌引用（Composite 节点用于 child_token）。

pub fn stream_sink(&self) -> Option<Arc<dyn StreamSink>>

获取 stream 的 Arc 引用（Parallel 子分支 clone 用）。

pub fn take_commit_batch(&mut self) -> Vec<S::Mutation>

取出 mutation batch（Executor 调用）。

这是 commit 流水线的第一段：

take_commit_batch() → TraceSink/MutationLog 消费 → apply_batch_to_state()

调用方可以在此处插入 Trace 记录、MutationLog 持久化等扩展点。

pub fn apply_batch_to_state(&mut self, mutations: Vec<S::Mutation>)

将 mutation batch 应用到状态（Executor 调用）。

commit 流水线的最后一段。与 take_commit_batch() 配合使用。

pub fn commit(&mut self)

完整的 commit 流水线（便捷方法）。

等价于 apply_batch_to_state(take_commit_batch())。 内部调用使用此方法即可；需要扩展 Trace/MutationLog 的场景 应手动调用 take_commit_batch() + 扩展点 + apply_batch_to_state()。

Trait Implementations

impl<'a, S: WorkflowState> ExecutionView<S> for ExecutionEngine<'a, S>

fn state(&self) -> &S

fn emit(&self, chunk: StreamChunk)

fn is_cancelled(&self) -> bool

impl<'a, S: WorkflowState> ExecutorState<S> for ExecutionEngine<'a, S>

fn build_node_context(&mut self) -> NodeContext<'_, S>

fn build_leaf_context(&mut self) -> LeafContext<'_, S>

fn clone_state(&self) -> S

fn replace_state(&mut self, state: S)

fn apply_batch(&mut self, mutations: impl IntoIterator<Item = S::Mutation>)

fn take_control(&mut self) -> (NextAction, Option<ExecutionSignal>)

fn take_metadata(&mut self) -> NodeMetadata