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.

App owns admission; the executor chooses polling order and returns each admitted completion exactly once or reports failure. App validates membership before reducing results in original child order.

The associated future preserves concrete child and execution types, both of which may be non-Send and non-Unpin. This GAT is the App-specific exception to ordinary RPITIT methods.

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 outcome = async {
    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);
    Ok::<(), Box<dyn std::error::Error>>(())
}.await;
let cleanup = app.shutdown(Shutdown::CancelAndDrain { timeout: std::time::Duration::from_secs(5) }).await;
if let Err(error) = &cleanup { eprintln!("App shutdown: {error}"); }
outcome?;
cleanup?;

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.

See the complete child executor workflow.

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