Trait ChildExecutor 

pub trait ChildExecutor:
    Send
    + Sync
    + 'static {
    type Execution<'a, F>: Future<Output = Result<Vec<ChildCompletion>, AppError>> + 'a
       where Self: 'a,
             F: Future<Output = Result<ChildCompletion, AppError>> + 'a;

    // Required method
    fn execute<'a, F>(&'a self, batch: ChildBatch<F>) -> Self::Execution<'a, F>
       where F: Future<Output = Result<ChildCompletion, AppError>> + 'a;
}

Caller-selected execution of a complete, already-admitted child batch.

The returned future may be local and need not be Unpin. App retains ownership of admission and validates exact returned membership before reducing results in original child order, independently of the executor’s polling order.

The associated GAT future is the App-specific exception to ordinary RPITIT methods. Both the child futures and returned future may be non-Send and non-Unpin; retain their concrete types. App owns admission and validates exact completion membership regardless of custom polling order. A custom executor must return every admitted completion once or report failure.

Example

use agents_kit::Agent;
use app_kit::{App, AppConfig, Shutdown, WorkspaceConfig};
use app_kit::workflow::{CheckpointSelector, ExecutionEvent, ExecutionOptions, RunSelector,
    RunStatus, WorkflowDefinition, WorkflowState};
use storage_kit::StateStore;
use storage_kit::state::config::{LanceStateStoreConfig, StateStoreConfig};
use app_kit::{AppError, workflow::{ActionMode, ChildBatch, ChildCompletion, ChildExecutor}};
use futures_util::FutureExt;
use std::future::Future;
struct InspectChildren;
impl InspectChildren {
    fn inspect(result: Result<Vec<ChildCompletion>, AppError>) -> Result<Vec<ChildCompletion>, AppError> {
        let completed = result?;
        assert_eq!(completed.len(), 1);
        assert_eq!(completed[0].ordinal(), 0);
        Ok(completed)
    }
}
impl ChildExecutor for InspectChildren {
    type Execution<'a, F> = futures_util::future::Map<ChildBatch<F>,
        fn(Result<Vec<ChildCompletion>, AppError>) -> Result<Vec<ChildCompletion>, AppError>>
    where Self: 'a, F: Future<Output = Result<ChildCompletion, AppError>> + 'a;
    fn execute<'a, F>(&'a self, batch: ChildBatch<F>) -> Self::Execution<'a, F>
    where F: Future<Output = Result<ChildCompletion, AppError>> + 'a {
        assert_eq!(batch.mode(), ActionMode::Asynchronous);
        assert_eq!(batch.len(), 1);
        assert!(!batch.is_empty());
        batch.map(Self::inspect as fn(_) -> _)
    }
}
let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
let definition = WorkflowDefinition::parse(r#"{
 "schema":"app-kit.workflow/v2","id":"counter","revision":"r1",
 "state":{"sum":0},"required":["sum"],"reducers":{"sum":"sum_integer"},
 "entry":"child","nodes":[{"id":"child","step":{"kind":"child","workflow":"child",
 "input":{"count":{"kind":"literal","value":4}},
 "reductions":[{"source":"count","target":"sum","reducer":"sum_integer"}]}}],"routes":{}
}"#)?;
let child = WorkflowDefinition::parse(r#"{
    "schema":"app-kit.workflow/v2", "id":"child", "revision":"r1",
    "state":{"count":0}, "required":["count"],
    "reducers":{"count":"reject_concurrent"}, "entry":"increment",
    "nodes":[{"id":"increment","step":{"kind":"state","mutations":[
        {"kind":"increment","target":"count","value":{"kind":"literal","value":1}}
    ]}}], "routes":{}
}"#)?;
let config = AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition, child]).with_child_executor(InspectChildren);
let mut app = App::open(config).await?;
let mut workspace = app.create_workspace(WorkspaceConfig {
    name: Some("local counter".into()), root: std::env::current_dir()?,
}).await?;
let workflow = workspace.workflow("counter".parse()?).await?;
let mut run = workflow.prepare(WorkflowState::default()).await?;
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
assert_eq!(terminal.state.values()["sum"], 5);
app.shutdown(Shutdown::CancelAndDrain { timeout: std::time::Duration::from_secs(5) }).await?;

Required Associated Types

type Execution<'a, F>: Future<Output = Result<Vec<ChildCompletion>, AppError>> + 'a where Self: 'a, F: Future<Output = Result<ChildCompletion, AppError>> + 'a

The exact caller future for this batch, without future erasure.

Required Methods

fn execute<'a, F>(&'a self, batch: ChildBatch<F>) -> Self::Execution<'a, F>

where F: Future<Output = Result<ChildCompletion, AppError>> + 'a,

Selects how the admitted child futures are polled and awaited.

App supplies already admitted work; this method chooses polling and collection, not admission or new child identity. Pipeline nodes use their own synchronous state handoff through the same kernel and bypass this seam.

Errors

Any Err returned by the execution future becomes WorkflowFailure::Child, regardless of the AppError variant. Input and admission operations may classify the same variant differently. A panic while constructing or polling the future is contained as WorkflowFailure::Execution. Neither failure reduces the parent from partial child results. Existing descendant cleanup retains ownership before terminal publication.

Example

use app_kit::{AppError, workflow::{ActionMode, ChildBatch, ChildCompletion, ChildExecutor}};
use futures_util::FutureExt;
use std::future::Future;
struct InspectChildren;
impl InspectChildren {
    fn inspect(result: Result<Vec<ChildCompletion>, AppError>) -> Result<Vec<ChildCompletion>, AppError> {
        let completed = result?;
        assert_eq!(completed.len(), 1);
        assert_eq!(completed[0].ordinal(), 0);
        Ok(completed)
    }
}
impl ChildExecutor for InspectChildren {
    type Execution<'a, F> = futures_util::future::Map<ChildBatch<F>,
        fn(Result<Vec<ChildCompletion>, AppError>) -> Result<Vec<ChildCompletion>, AppError>>
    where Self: 'a, F: Future<Output = Result<ChildCompletion, AppError>> + 'a;
    fn execute<'a, F>(&'a self, batch: ChildBatch<F>) -> Self::Execution<'a, F>
    where F: Future<Output = Result<ChildCompletion, AppError>> + 'a {
        assert_eq!(batch.mode(), ActionMode::Asynchronous);
        assert_eq!(batch.len(), 1);
        assert!(!batch.is_empty());
        batch.map(Self::inspect as fn(_) -> _)
    }
}
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
assert_eq!(terminal.state.values()["sum"], 5);

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl ChildExecutor for InlineExecutor

type Execution<'a, F> = ChildBatch<F> where Self: 'a, F: Future<Output = Result<ChildCompletion, AppError>> + 'a