Struct ExecutionSession 

pub struct ExecutionSession<S: WorkflowState = State, M: MergeStrategy<S> = StateMerge>
where
    S::Checkpoint: Debug,
{ /* private fields */ }

执行会话 — 持有 State 所有权 + FrameStack + Graph 引用。

职责

• 持有 State 所有权（Engine 只是借用）
• 管理 FrameStack（Subgraph 执行时 push/pop）
• 创建和恢复 SessionCheckpoint

设计原则

Graph 是 Immutable 的，多个 Session 共享同一个 Graph 实例。 Session 不拥有 Graph，只持有 Arc<Graph> 引用。

Runtime
└── Arc<Graph>

Session1 ──┐
Session2 ──┼── Arc<Graph>
Session3 ──┘

Implementations

impl<S: WorkflowState, M: MergeStrategy<S>> ExecutionSession<S, M>

where S::Checkpoint: Debug,

pub fn new(state: S, graph: Arc<Graph<S, M>>) -> Self

创建新的执行会话。

pub fn restore( checkpoint: SessionCheckpoint<S>, graph: Arc<Graph<S, M>>, ) -> Result<Self, SessionError>

从 Checkpoint 恢复。

P0-1: 使用 restore() 恢复 State

S::restore(checkpoint.state) 从 checkpoint snapshot 恢复完整 Runtime State。

Graph 参数

调用方负责提供 Arc<Graph>（从 Runtime 获取）， Session 不负责存储或查找 Graph。

错误

如果 checkpoint.graph_hash 与 graph.canonical_hash() 不匹配， 返回 SessionError::GraphMismatch，拒绝恢复。

pub fn checkpoint(&self) -> SessionCheckpoint<S>

创建 checkpoint — 保存当前执行位置 + 状态投影。

P0-1: 使用 snapshot() 进行投影

state.snapshot() 返回 S::Checkpoint，只序列化必要字段。

P0-2: 使用 canonical_hash

graph.canonical_hash() 从 DSL 层计算，不依赖 HashMap 顺序。

pub fn state(&self) -> &S

获取状态引用。

pub fn state_mut(&mut self) -> &mut S

获取状态可变引用。

pub fn frame_stack(&self) -> &FrameStack<S>

获取帧栈引用。

pub fn frame_stack_mut(&mut self) -> &mut FrameStack<S>

获取帧栈可变引用（用于 Subgraph 执行时 push/pop）。

pub fn graph(&self) -> &Graph<S, M>

获取图引用。

pub fn graph_arc(&self) -> Arc<Graph<S, M>>

获取图的 Arc 引用（用于共享）。

pub fn into_state(self) -> S

消费会话，返回最终状态。

pub fn into_parts(self) -> (S, FrameStack<S>)

消费会话，返回 (状态, 帧栈)。

impl<S: WorkflowState, M: MergeStrategy<S>> ExecutionSession<S, M>

where S::Checkpoint: Debug,

pub async fn run_with( &mut self, engine: &mut ExecutionEngine<'_, S>, ) -> Result<(), GraphError>

使用指定的 Engine 执行。

Session 不知道 Stream，Engine 才知道 Stream。 职责分离：Session 负责 state + frame_stack，Engine 负责执行 + stream。

示例

// 创建 Checkpoint Sink（可选）
let mut sink = SessionCheckpointSink::new(
    session.frame_stack_mut(),
    session.graph().name(),
);

let mut engine = ExecutionEngine::new(
    session.state_mut(),
    Some(stream),       // Stream 由调用者提供
    cancel,
    Some(&mut sink),    // 启用自动 checkpoint
    None,               // 不需要 Barrier Sink
);
session.run_with(&mut engine).await?;

Trait Implementations

impl<S, M: MergeStrategy<S>> Default for ExecutionSession<S, M>

where S: Default + WorkflowState, S::Checkpoint: Debug,

fn default() -> Self

Returns the “default value” for a type.