Struct ExecutionState 

pub struct ExecutionState {

    pub persistence: Arc<dyn WorkflowPersistence>,
    pub action_registry: Arc<ActionRegistry>,
    pub script_env_provider: Arc<dyn ScriptEnvProvider>,
    pub workflow_run_id: String,
    pub workflow_name: String,
    pub run_ctx: Arc<dyn RunContext>,
    pub extra_plugin_dirs: Vec<String>,
    pub model: Option<String>,
    pub exec_config: WorkflowExecConfig,
    pub inputs: HashMap<String, String>,
    pub parent_run_id: String,
    pub depth: u32,
    pub target_label: Option<String>,
    pub step_results: HashMap<String, StepResult>,
    pub contexts: Vec<ContextEntry>,
    pub position: i64,
    pub all_succeeded: bool,
    pub total_cost: f64,
    pub total_turns: i64,
    pub total_duration_ms: i64,
    pub total_input_tokens: i64,
    pub total_output_tokens: i64,
    pub total_cache_read_input_tokens: i64,
    pub total_cache_creation_input_tokens: i64,
    pub has_llm_metrics: bool,
    pub last_gate_feedback: Option<String>,
    pub block_output: Option<String>,
    pub block_with: Vec<String>,
    pub resume_ctx: Option<ResumeContext>,
    pub default_as_identity: Option<String>,
    pub triggered_by_hook: bool,
    pub schema_resolver: Option<Arc<dyn Fn(&str) -> Result<OutputSchema> + Send + Sync>>,
    pub child_runner: Option<Arc<dyn ChildWorkflowRunner>>,
    pub last_heartbeat_at: Arc<AtomicI64>,
    pub registry: Arc<ItemProviderRegistry>,
    pub event_sinks: Arc<[Arc<dyn EventSink>]>,
    pub cancellation: CancellationToken,
    pub current_execution_id: Arc<Mutex<Option<(String, String)>>>,
    pub owner_token: Option<String>,
    pub lease_generation: Option<i64>,
}

Mutable runtime state for a workflow execution — no conductor-core deps.

Fields

persistence: Arc<dyn WorkflowPersistence>

action_registry: Arc<ActionRegistry>

script_env_provider: Arc<dyn ScriptEnvProvider>

workflow_run_id: String

workflow_name: String

run_ctx: Arc<dyn RunContext>

Shared per-run context carrying injected variables and working directory. Arc (not Box) because ExecutionState derives Clone for fork_child.

extra_plugin_dirs: Vec<String>

Extra plugin directories for the executor. Not part of RunContext because Vec<String> doesn’t fit the HashMap<&'static str, String> injected-variables contract, and only executor code reads it.

model: Option<String>

exec_config: WorkflowExecConfig

inputs: HashMap<String, String>

parent_run_id: String

depth: u32

target_label: Option<String>

step_results: HashMap<String, StepResult>

contexts: Vec<ContextEntry>

position: i64

all_succeeded: bool

total_cost: f64

total_turns: i64

total_duration_ms: i64

total_input_tokens: i64

total_output_tokens: i64

total_cache_read_input_tokens: i64

total_cache_creation_input_tokens: i64

has_llm_metrics: bool

last_gate_feedback: Option<String>

block_output: Option<String>

block_with: Vec<String>

resume_ctx: Option<ResumeContext>

default_as_identity: Option<String>

triggered_by_hook: bool

schema_resolver: Option<Arc<dyn Fn(&str) -> Result<OutputSchema> + Send + Sync>>

Schema resolver callback — (schema_name) → OutputSchema. The host closes over working_dir and repo_path at construction time.

child_runner: Option<Arc<dyn ChildWorkflowRunner>>

Runner for child workflows (call workflow nodes).

last_heartbeat_at: Arc<AtomicI64>

registry: Arc<ItemProviderRegistry>

event_sinks: Arc<[Arc<dyn EventSink>]>

Event sinks — slice shared cheaply across sub-workflow states.

cancellation: CancellationToken

Cancellation token for this run. Checked at each step boundary.

current_execution_id: Arc<Mutex<Option<(String, String)>>>

The executor label and step_id of the currently executing action, if any. Written by execute_call before dispatch; read by FlowEngine::cancel_run() to fire-and-forget executor.cancel().

owner_token: Option<String>

Lease token held by this engine instance after a successful acquire_lease(). Used by PRs 3–5 for refresh and generation checks.

lease_generation: Option<i64>

Implementations

impl ExecutionState

pub fn new_heartbeat() -> Arc<AtomicI64>

Create a fresh heartbeat counter, initialized to 0 so the first tick fires immediately.

pub fn expect_lease_generation(&self) -> i64

Read the current lease generation, panicking with a consistent message if the lease was never acquired. Every executor update_step call site requires a generation, and FlowEngine::run/resume is the single entry point that sets it — so a None here is a programmer error in engine construction, not a runtime condition.

pub fn check_cancellation_throttled(&self) -> Result<()>

Throttled heartbeat tick + external cancel check.

Bumps last_heartbeat in persistence at most once every 5 seconds and polls for cross-process cancellation via persistence.is_run_cancelled. On external cancel, sets self.cancellation and returns Err(EngineError::Cancelled).

Callers that own the engine main loop use ? to propagate cancellation up. Wait loops that need to drain in-flight work (parallel, foreach) can call this best-effort and rely on self.cancellation.is_cancelled() for their controlled exit — the cancellation token is set by this helper before the Err is returned.

Without this being called from inside long-running wait loops (parallel blocks, foreach fan-out), the cancellation check is skipped during multi-minute waits — see #2731.

NOTE (#2731/#2796): lease refresh (refresh_lease_loop in flow_engine.rs) is the load-bearing ownership mechanism. detect_stuck_workflow_run_ids falls back to started_at when last_heartbeat is NULL (new runs).

pub fn child_workflow_context(&self) -> ChildWorkflowContext

Project this state into the narrow surface a ChildWorkflowRunner implementation needs to spawn a child run.

pub fn fork_child(&self, cancellation: CancellationToken) -> ExecutionState

Fork a child execution state from this parent.

Copies shared configuration (persistence, registries, workflow identity) and resets all runtime accumulators so the child starts with a clean slate.

pub fn accumulate_metrics( &mut self, cost: Option<f64>, turns: Option<i64>, duration: Option<i64>, input_tokens: Option<i64>, output_tokens: Option<i64>, cache_read: Option<i64>, cache_create: Option<i64>, ) -> bool

Accumulate individual metrics into this execution state.

Returns true if at least one metric was present and added.

Trait Implementations

impl Clone for ExecutionState

fn clone(&self) -> ExecutionState

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.