Struct SyncExecution 

pub struct SyncExecution<'a, CS: StateStoreBackend, A: AgentBackend, X: ActionBackend = NoActions, C: ConditionBackend = NoConditions, T: StateTyping = JsonState, H: LifecycleHooks = NoHooks, E: ChildExecutor = InlineExecutor> { /* private fields */ }

Blocking advancement of the same caller-owned execution source.

Calls require a thread outside any entered Tokio runtime context. The App’s captured runtime and caller-provided lower runtimes must remain alive and driven elsewhere. Source polling and destruction remain on the caller thread; dropping this owner delegates nonblocking revocation to Execution.

Example

let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(2).enable_all().build()?;
// Runtime workers stay alive; this thread is outside any entered Tokio context.
let mut execution = SyncExecution::open(&mut run, ExecutionOptions::default())?;
while execution.try_next()?.is_some() {}
assert_eq!(execution.finish()?.state.values()["count"], 1);

Implementations

impl<'a, CS: StateStoreBackend, A: AgentBackend, X: ActionBackend, C: ConditionBackend, T: StateTyping, H: LifecycleHooks, E: ChildExecutor> SyncExecution<'a, CS, A, X, C, T, H, E>

pub fn open( run: &'a mut WorkflowRun<CS, A, X, C, T, H, E>, options: ExecutionOptions, ) -> Result<Self, AppError>

Acquires lazy synchronous advancement on a thread outside Tokio.

The App’s captured runtime must remain alive and driven elsewhere. Acquisition validates options before callbacks or source construction. The returned owner borrows the run exclusively through its last use.

A live synchronous owner prevents overlapping access to the run:

use agents_kit::AgentBackend;
use app_kit::{AppError, workflow::{ExecutionOptions, RunSnapshot, SyncExecution, WorkflowRun}};
use storage_kit::StateStoreBackend;
fn overlapping<CS: StateStoreBackend, A: AgentBackend>(
run: &mut WorkflowRun<CS, A>,
) -> Result<RunSnapshot, AppError> {
let execution = SyncExecution::open(run, ExecutionOptions::default())?;
let _ = run.id();
execution.abort()
}

Errors

Returns InvalidInput inside an entered Tokio runtime context before blocking. Otherwise propagates the corresponding asynchronous execution errors, including stale/revoked authority, invalid options, durable failures and timeouts. The captured runtime and lower runtimes must stay alive and driven elsewhere.

Example

let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(2).enable_all().build()?;
// Runtime workers stay alive; this thread is outside any entered Tokio context.
let execution = SyncExecution::open(&mut run, ExecutionOptions::default())?;
assert_eq!(execution.finish()?.state.values()["count"], 1);

pub fn try_next(&mut self) -> Result<Option<ExecutionEvent>, AppError>

Blocks until one partial result or committed boundary is available.

An entered runtime context is rejected before advancing this owner.

Errors

Returns InvalidInput inside an entered Tokio runtime context before blocking. Otherwise propagates the corresponding asynchronous execution errors, including stale/revoked authority, invalid options, durable failures and timeouts. The captured runtime and lower runtimes must stay alive and driven elsewhere.

Example

let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(2).enable_all().build()?;
// Runtime workers stay alive; this thread is outside any entered Tokio context.
let mut execution = SyncExecution::open(&mut run, ExecutionOptions::default())?;
assert!(matches!(execution.try_next()?, Some(ExecutionEvent::Committed { .. })));
execution.finish()?;

pub fn step(&mut self) -> Result<Option<ExecutionEvent>, AppError>

Drains partial results and blocks until one committed boundary.

An entered runtime context is rejected before advancing this owner.

Errors

Returns InvalidInput inside an entered Tokio runtime context before blocking. Otherwise propagates the corresponding asynchronous execution errors, including stale/revoked authority, invalid options, durable failures and timeouts. The captured runtime and lower runtimes must stay alive and driven elsewhere.

Example

let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(2).enable_all().build()?;
// Runtime workers stay alive; this thread is outside any entered Tokio context.
let mut execution = SyncExecution::open(&mut run, ExecutionOptions::default())?;
assert!(matches!(execution.step()?, Some(ExecutionEvent::Committed { .. })));
assert_eq!(execution.finish()?.state.values()["count"], 1);

pub fn finish(self) -> Result<RunSnapshot, AppError>

Drains execution to a halt or exhaustion and finalizes its observation.

Rejection inside an entered runtime drops this consumed owner; existing execution cleanup retains any previously admitted publication.

Errors

Returns InvalidInput inside an entered Tokio runtime context before blocking. Otherwise propagates the corresponding asynchronous execution errors, including stale/revoked authority, invalid options, durable failures and timeouts. The captured runtime and lower runtimes must stay alive and driven elsewhere.

Example

let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(2).enable_all().build()?;
// Runtime workers stay alive; this thread is outside any entered Tokio context.
let execution = SyncExecution::open(&mut run, ExecutionOptions::default())?;
let snapshot = execution.finish()?;
assert!(matches!(snapshot.status, RunStatus::Succeeded));

pub fn abort(self) -> Result<RunSnapshot, AppError>

Finalizes the current uncommitted boundary without cancelling the run.

Rejection inside an entered runtime drops this consumed owner; it does not dispatch a new operation or discard accepted publication ownership.

Errors

Returns InvalidInput inside an entered Tokio runtime context before blocking. Otherwise propagates the corresponding asynchronous execution errors, including stale/revoked authority, invalid options, durable failures and timeouts. The captured runtime and lower runtimes must stay alive and driven elsewhere.

Example

let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(2).enable_all().build()?;
// Runtime workers stay alive; this thread is outside any entered Tokio context.
let execution = SyncExecution::open(&mut run, ExecutionOptions::default())?;
let parked = execution.abort()?;
assert_eq!(parked.state.values()["count"], 0);
assert_eq!(SyncExecution::open(&mut run, ExecutionOptions::default())?.finish()?.state.values()["count"], 1);