Struct AppConfig 

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

Complete application configuration.

App does not offer a separate built-in storage construction path:

let _ = app_kit::AppState::<storage_kit::StateStore>::Ephemeral;

Owns caller-built storage, a homogeneous agent inventory, and workflow definitions.

The consuming with_* methods change their concrete App behavior parameter. They do not erase backends or alter already admitted runs. Heterogeneous agent inventories use a caller-owned exhaustive hierarchy. Dropping configuration before open does not perform asynchronous Agent shutdown.

Configure behavior

Start with ready storage, ready Agents, and parsed definitions. Apply the consuming configuration methods before App::open; they retain exact implementation types, including associated lower children and transaction ownership. None registers a runtime callback or erases a backend.

Behavior	Complete implementation example	Attach before open
Produce lazy action events and writes	workflow::ActionBackend	Self::with_actions
Evaluate declared routing predicates	workflow::ConditionBackend	Self::with_conditions
Construct and validate application state	workflow::StateTyping	Self::with_state_typing
Observe authoritative lifecycle events	workflow::LifecycleHooks	Self::with_hooks
Drive concrete local child futures	workflow::ChildExecutor	Self::with_child_executor

The with_* examples exercise the complete path from a concrete implementation to opening App. Action and condition declarations must match the workflow’s exact logical keys and revisions. A custom child executor must return every admitted completion exactly once; it never creates child identities. App’s built-in NoActions, NoConditions, JsonState, NoHooks and InlineExecutor are suitable defaults only where their actual behavior satisfies the workflow.

Lower composition is independent: A::Workspace fixes its Conversation, while CS::Transaction fixes consuming commit/rollback. For heterogeneous Agent inventories implement matching caller-owned exhaustive hierarchies; do not substitute public trait objects or independent child parameters.

Example

let mut app = App::open(config).await?;
app.shutdown(Shutdown::RejectActive).await?;

Implementations

impl<CS: StateStoreBackend, A: AgentBackend> AppConfig<CS, A>

pub fn new( state: CS, agents: Vec<AgentSpec<A>>, workflows: Vec<WorkflowDefinition>, ) -> Self

Constructs configuration from a caller-owned control-plane store and agents.

The store is required even for process-local operation. App preserves its concrete transaction type and never constructs lower products. Ownership transfers to App when App::open is called.

Example

let config = AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition]);
let mut app = App::open(config).await?;
app.shutdown(Shutdown::RejectActive).await?;

Examples found in repository

examples/history_and_recovery.rs (lines 55-59)

    async fn open_app(&self) -> Result<App<StateStore, Agent>, AppError> {
        let definition = WorkflowDefinition::parse(
            &json!({
                "schema":"app-kit.workflow/v2", "id":"history-counter", "revision":"r1",
                "state":{"count":0}, "required":["count"],
                "reducers":{"count":"reject_concurrent"}, "entry":"first",
                "nodes":[
                    {"id":"first", "step":{"kind":"state", "mutations":[{
                        "kind":"increment", "target":"count",
                        "value":{"kind":"literal", "value":1}
                    }]}},
                    {"id":"second", "step":{"kind":"state", "mutations":[{
                        "kind":"increment", "target":"count",
                        "value":{"kind":"literal", "value":1}
                    }]}}
                ],
                "routes":{"first":{"mode":"first", "branches":[{
                    "target":{"kind":"node", "id":"second"}
                }]}}
            })
            .to_string(),
        )?;
        // Agent supplies the static type for an empty inventory. No Agent value
        // is constructed, and these native State nodes need no lower service.
        App::open(AppConfig::<_, Agent>::new(
            self.store.clone(),
            Vec::new(),
            vec![definition],
        ))
        .await
    }

examples/actions_and_inputs.rs (line 40)

    async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
        let definition = WorkflowDefinition::parse(
            &json!({
                "schema":"app-kit.workflow/v2", "id":"invoice", "revision":"r1",
                "state":{"unit_price":7,"total":0}, "required":["unit_price","total"],
                "reducers":{"unit_price":"reject_concurrent","total":"reject_concurrent"},
                "entry":"calculate",
                "nodes":[{"id":"calculate","step":{
                    "kind":"action","action":"invoice","revision":"r1","project_result":false
                }}], "routes":{}
            })
            .to_string(),
        )?;
        let store =
            StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
        // Native caller actions need no Agent value or provider service.
        let mut app = App::open(
            AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition])
                .with_actions(vec![InvoiceAction::new()]),
        )
        .await?;
        let observed = tokio::time::timeout(Duration::from_secs(15), async {
            let workspace = app
                .create_workspace(WorkspaceConfig {
                    name: Some("invoice example".into()),
                    root: PathBuf::from(env!("CARGO_MANIFEST_DIR")),
                })
                .await?;
            let workflow = workspace.workflow("invoice".parse()?).await?;
            let mut run = workflow.prepare(WorkflowState::default()).await?;
            let mut execution = run
                .execution(ExecutionOptions {
                    inputs: Map::from_iter([
                        ("quantity".into(), json!(6)),
                        ("trace".into(), json!("sample")),
                    ]),
                    ..ExecutionOptions::default()
                })
                .await?;
            let Some(ExecutionEvent::Partial { result, .. }) = execution.next().await? else {
                return Err(AppError::Integration(
                    "expected the real action partial".into(),
                ));
            };
            if result.get("progress") != Some(&json!("calculated")) {
                return Err(AppError::Integration("partial changed".into()));
            }
            let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
                return Err(AppError::Integration(
                    "expected one committed boundary".into(),
                ));
            };
            let expected =
                json!({"quantity":6,"label":"order","currency":"USD","trace":"sample","total":42});
            if outcomes.len() != 1 || Value::Object(outcomes[0].result.clone()) != expected {
                return Err(AppError::Integration(
                    "resolved inputs or action result changed".into(),
                ));
            }
            let terminal = execution.finish().await?;
            if !matches!(terminal.status, RunStatus::Succeeded)
                || terminal.state.values()
                    != json!({"unit_price":7,"total":42}).as_object().unwrap()
            {
                return Err(AppError::Integration(
                    "only the explicit total write should persist".into(),
                ));
            }
            println!(
                "Partial: calculated; committed total: 42; transient inputs stayed out of state"
            );
            Ok::<(), AppError>(())
        })
        .await;
        let closed = tokio::time::timeout(
            Duration::from_secs(15),
            app.shutdown(Shutdown::CancelAndDrain {
                timeout: Duration::from_secs(10),
            }),
        )
        .await;
        drop(app);
        observed.map_err(|_| AppError::Timeout("invoice execution timed out".into()))??;
        closed.map_err(|_| AppError::Timeout("invoice cleanup timed out".into()))??;
        Ok(())
    }

examples/performance.rs (line 1107)

    async fn run(mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
        let manifest = if matches!(self.command, Command::ReopenHistory { .. }) {
            self.root = self.root.canonicalize()?;
            Some(self.read_manifest()?)
        } else {
            fs::create_dir(&self.root)?;
            self.root = self.root.canonicalize()?;
            fs::create_dir(self.root.join("workspace"))?;
            let mut marker = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(self.root.join(".performance-root"))?;
            marker.write_all(MANIFEST_SCHEMA.as_bytes())?;
            marker.sync_all()?;
            None
        };
        let actions = match self.command {
            Command::Linear => 1_000,
            Command::SeedHistory { rows } => (rows - 4) / 3,
            Command::Ready => 32 * 32,
            Command::Map => 4_096,
            Command::Repeated => 100,
            Command::NestedCancel => 16,
            Command::DelayedObserver => 1,
            Command::ReopenHistory { .. } => {
                manifest.as_ref().unwrap()["actions"].as_u64().unwrap()
            }
        };
        let linear = matches!(self.command, Command::Linear);
        let (definitions, definition_bytes) = self.definitions(actions)?;
        let probe = match self.command {
            Command::Ready => Some(WorkloadProbe::new(SourceMode::Permuted, 1_024)),
            Command::Map | Command::Repeated => {
                Some(WorkloadProbe::new(SourceMode::Parallel, actions as usize))
            }
            Command::NestedCancel => Some(WorkloadProbe::new(SourceMode::Held, 16)),
            Command::DelayedObserver => Some(WorkloadProbe::new(SourceMode::Partials, 1)),
            Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => None,
        };
        if let Some(manifest) = &manifest {
            assert_eq!(manifest["definition_bytes"], json!(definition_bytes));
        }
        let mut app = None;
        let started = Instant::now();
        let observed = AssertUnwindSafe(tokio::time::timeout(WORKLOAD_BUDGET, async {
            let connecting = Instant::now();
            let config = LanceStateStoreConfig::directory(self.root.join("state"))
                .with_operation_timeout(Duration::from_secs(30))?;
            let store = StateStore::connect(StateStoreConfig::Lance(config)).await?;
            let connect = connecting.elapsed();
            let opening = Instant::now();
            app = Some(
                App::open(
                    AppConfig::<_, Agent>::new(store, Vec::new(), definitions).with_actions(
                        probe
                            .as_ref()
                            .map(WorkloadAction::new)
                            .into_iter()
                            .collect(),
                    ),
                )
                .await?,
            );
            let open = opening.elapsed();
            let mut result = match self.command {
                Command::Linear | Command::SeedHistory { .. } => {
                    self.seed(app.as_ref().unwrap(), actions, linear).await?
                }
                Command::Ready => {
                    self.ready(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::Map => {
                    self.mapped(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::Repeated => {
                    self.repeated(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::NestedCancel => {
                    self.nested_cancel(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::DelayedObserver => {
                    self.delayed_observer(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::ReopenHistory { open_only } => {
                    self.reopen(
                        app.as_ref().unwrap(),
                        manifest.as_ref().unwrap(),
                        actions,
                        open_only,
                    )
                    .await?
                }
            };
            result["store_connect_ns"] = json!(connect.as_nanos());
            result["app_open_ns"] = json!(open.as_nanos());
            result["workload_elapsed_ns"] = json!(started.elapsed().as_nanos());
            Ok::<_, Box<dyn Error + Send + Sync>>(result)
        }))
        .catch_unwind()
        .await;
        let success = matches!(&observed, Ok(Ok(Ok(_))));
        let closing = Instant::now();
        let cleanup = if let Some(mut app) = app.take() {
            let closed = AssertUnwindSafe(Self::close(&mut app, success))
                .catch_unwind()
                .await;
            drop(app);
            Some(closed)
        } else {
            None
        };
        if !success {
            eprintln!(
                "performance workload failed; cleanup completed={}",
                matches!(&cleanup, Some(Ok(Ok(()))))
            );
        }
        let mut result = match observed {
            Ok(Ok(result)) => result?,
            Ok(Err(_)) => {
                return Err(
                    AppError::Timeout("performance workload budget exhausted".into()).into(),
                );
            }
            Err(panic) => resume_unwind(panic),
        };
        match cleanup {
            Some(Ok(result)) => result?,
            Some(Err(panic)) => resume_unwind(panic),
            None => return Err("successful workload did not retain its App cleanup owner".into()),
        }
        // All run/workflow/App owners and the caller-created store have dropped.
        // A cleanup timeout remains a failure even if reattachment later joined.
        result["schema"] = json!(MANIFEST_SCHEMA);
        result["settings"] = Self::settings(actions);
        if let Some(probe) = &probe {
            probe.assert_idle();
            result["caller_sources"] = probe.report();
            result["settings"]
                .as_object_mut()
                .unwrap()
                .remove("history_row_ceiling");
            result["settings"]["private_task_counts"] =
                json!("not exposed; App shutdown proves joined ownership");
            result["settings"]["lower_agents_and_tool_handlers"] = json!("none constructed");
        }
        result["definition_bytes"] = json!(definition_bytes);
        result["process_id"] = json!(std::process::id());
        result["os"] = json!(std::env::consts::OS);
        result["architecture"] = json!(std::env::consts::ARCH);
        result["package_version"] = json!(env!("CARGO_PKG_VERSION"));
        result["cleanup_joined"] = json!(true);
        result["cleanup_ns"] = json!(closing.elapsed().as_nanos());
        let encoded = serde_json::to_vec(&result)?;
        assert!(
            encoded.len() <= 16_384,
            "result stays within the manifest bound"
        );
        if matches!(self.command, Command::SeedHistory { .. }) {
            let mut file = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(self.root.join("manifest.json"))?;
            file.write_all(&encoded)?;
            file.sync_all()?;
        }
        println!("{}", std::str::from_utf8(&encoded)?);
        Ok(())
    }

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

pub fn with_actions<N: ActionBackend>( self, actions: Vec<N>, ) -> AppConfig<CS, A, N, C, T, H, E>

Replaces the complete caller-owned action inventory before opening App.

Values retain their concrete type and need not implement Clone. App validates their stable declarations before accepting any workflow work. Replacing an inventory drops its previous values without invoking them.

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::{ActionBackend, ActionDeclaration, ActionEvent,
    ActionMode, ActionRequest, StateChange}};
use serde_json::{Map, Value};
struct Increment { declaration: ActionDeclaration }
impl ActionBackend for Increment {
    type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
    fn declaration(&self) -> &ActionDeclaration { &self.declaration }
    fn start(&self, request: ActionRequest) -> Self::Stream {
        let count = request.state.values()["count"].as_i64().unwrap() + 1;
        futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
            result: Map::from_iter([("new_count".into(), Value::from(count))]),
            writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
        })])
    }
}
let action = Increment { declaration: ActionDeclaration {
    key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
    writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
    bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
    source: Some("increment count by one".into()), mode: ActionMode::Both,
} };
let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
let definition = WorkflowDefinition::parse(r#"{
 "schema":"app-kit.workflow/v2","id":"counter","revision":"r1",
 "state":{"count":0},"required":["count"],"reducers":{"count":"reject_concurrent"},
 "entry":"increment","nodes":[{"id":"increment","step":{"kind":"action",
 "action":"increment","revision":"r1"}}],"routes":{}
}"#)?;
let config = AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition]);
let config = config.with_actions(vec![action]);
let mut app = App::open(config).await?;
let workspace = app.create_workspace(WorkspaceConfig { name: None, root: std::env::current_dir()? }).await?;
let workflow = workspace.workflow("counter".parse()?).await?;
let run = workflow.start(ExecutionOptions::default()).await?;
assert_eq!(run.wait().await?.state.values()["count"], 1);
app.shutdown(Shutdown::Drain { timeout: std::time::Duration::from_secs(5) }).await?;

Examples found in repository

examples/actions_and_inputs.rs (line 41)

    async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
        let definition = WorkflowDefinition::parse(
            &json!({
                "schema":"app-kit.workflow/v2", "id":"invoice", "revision":"r1",
                "state":{"unit_price":7,"total":0}, "required":["unit_price","total"],
                "reducers":{"unit_price":"reject_concurrent","total":"reject_concurrent"},
                "entry":"calculate",
                "nodes":[{"id":"calculate","step":{
                    "kind":"action","action":"invoice","revision":"r1","project_result":false
                }}], "routes":{}
            })
            .to_string(),
        )?;
        let store =
            StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
        // Native caller actions need no Agent value or provider service.
        let mut app = App::open(
            AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition])
                .with_actions(vec![InvoiceAction::new()]),
        )
        .await?;
        let observed = tokio::time::timeout(Duration::from_secs(15), async {
            let workspace = app
                .create_workspace(WorkspaceConfig {
                    name: Some("invoice example".into()),
                    root: PathBuf::from(env!("CARGO_MANIFEST_DIR")),
                })
                .await?;
            let workflow = workspace.workflow("invoice".parse()?).await?;
            let mut run = workflow.prepare(WorkflowState::default()).await?;
            let mut execution = run
                .execution(ExecutionOptions {
                    inputs: Map::from_iter([
                        ("quantity".into(), json!(6)),
                        ("trace".into(), json!("sample")),
                    ]),
                    ..ExecutionOptions::default()
                })
                .await?;
            let Some(ExecutionEvent::Partial { result, .. }) = execution.next().await? else {
                return Err(AppError::Integration(
                    "expected the real action partial".into(),
                ));
            };
            if result.get("progress") != Some(&json!("calculated")) {
                return Err(AppError::Integration("partial changed".into()));
            }
            let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
                return Err(AppError::Integration(
                    "expected one committed boundary".into(),
                ));
            };
            let expected =
                json!({"quantity":6,"label":"order","currency":"USD","trace":"sample","total":42});
            if outcomes.len() != 1 || Value::Object(outcomes[0].result.clone()) != expected {
                return Err(AppError::Integration(
                    "resolved inputs or action result changed".into(),
                ));
            }
            let terminal = execution.finish().await?;
            if !matches!(terminal.status, RunStatus::Succeeded)
                || terminal.state.values()
                    != json!({"unit_price":7,"total":42}).as_object().unwrap()
            {
                return Err(AppError::Integration(
                    "only the explicit total write should persist".into(),
                ));
            }
            println!(
                "Partial: calculated; committed total: 42; transient inputs stayed out of state"
            );
            Ok::<(), AppError>(())
        })
        .await;
        let closed = tokio::time::timeout(
            Duration::from_secs(15),
            app.shutdown(Shutdown::CancelAndDrain {
                timeout: Duration::from_secs(10),
            }),
        )
        .await;
        drop(app);
        observed.map_err(|_| AppError::Timeout("invoice execution timed out".into()))??;
        closed.map_err(|_| AppError::Timeout("invoice cleanup timed out".into()))??;
        Ok(())
    }

examples/performance.rs (lines 1107-1113)

    async fn run(mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
        let manifest = if matches!(self.command, Command::ReopenHistory { .. }) {
            self.root = self.root.canonicalize()?;
            Some(self.read_manifest()?)
        } else {
            fs::create_dir(&self.root)?;
            self.root = self.root.canonicalize()?;
            fs::create_dir(self.root.join("workspace"))?;
            let mut marker = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(self.root.join(".performance-root"))?;
            marker.write_all(MANIFEST_SCHEMA.as_bytes())?;
            marker.sync_all()?;
            None
        };
        let actions = match self.command {
            Command::Linear => 1_000,
            Command::SeedHistory { rows } => (rows - 4) / 3,
            Command::Ready => 32 * 32,
            Command::Map => 4_096,
            Command::Repeated => 100,
            Command::NestedCancel => 16,
            Command::DelayedObserver => 1,
            Command::ReopenHistory { .. } => {
                manifest.as_ref().unwrap()["actions"].as_u64().unwrap()
            }
        };
        let linear = matches!(self.command, Command::Linear);
        let (definitions, definition_bytes) = self.definitions(actions)?;
        let probe = match self.command {
            Command::Ready => Some(WorkloadProbe::new(SourceMode::Permuted, 1_024)),
            Command::Map | Command::Repeated => {
                Some(WorkloadProbe::new(SourceMode::Parallel, actions as usize))
            }
            Command::NestedCancel => Some(WorkloadProbe::new(SourceMode::Held, 16)),
            Command::DelayedObserver => Some(WorkloadProbe::new(SourceMode::Partials, 1)),
            Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => None,
        };
        if let Some(manifest) = &manifest {
            assert_eq!(manifest["definition_bytes"], json!(definition_bytes));
        }
        let mut app = None;
        let started = Instant::now();
        let observed = AssertUnwindSafe(tokio::time::timeout(WORKLOAD_BUDGET, async {
            let connecting = Instant::now();
            let config = LanceStateStoreConfig::directory(self.root.join("state"))
                .with_operation_timeout(Duration::from_secs(30))?;
            let store = StateStore::connect(StateStoreConfig::Lance(config)).await?;
            let connect = connecting.elapsed();
            let opening = Instant::now();
            app = Some(
                App::open(
                    AppConfig::<_, Agent>::new(store, Vec::new(), definitions).with_actions(
                        probe
                            .as_ref()
                            .map(WorkloadAction::new)
                            .into_iter()
                            .collect(),
                    ),
                )
                .await?,
            );
            let open = opening.elapsed();
            let mut result = match self.command {
                Command::Linear | Command::SeedHistory { .. } => {
                    self.seed(app.as_ref().unwrap(), actions, linear).await?
                }
                Command::Ready => {
                    self.ready(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::Map => {
                    self.mapped(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::Repeated => {
                    self.repeated(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::NestedCancel => {
                    self.nested_cancel(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::DelayedObserver => {
                    self.delayed_observer(app.as_ref().unwrap(), probe.as_ref().unwrap())
                        .await?
                }
                Command::ReopenHistory { open_only } => {
                    self.reopen(
                        app.as_ref().unwrap(),
                        manifest.as_ref().unwrap(),
                        actions,
                        open_only,
                    )
                    .await?
                }
            };
            result["store_connect_ns"] = json!(connect.as_nanos());
            result["app_open_ns"] = json!(open.as_nanos());
            result["workload_elapsed_ns"] = json!(started.elapsed().as_nanos());
            Ok::<_, Box<dyn Error + Send + Sync>>(result)
        }))
        .catch_unwind()
        .await;
        let success = matches!(&observed, Ok(Ok(Ok(_))));
        let closing = Instant::now();
        let cleanup = if let Some(mut app) = app.take() {
            let closed = AssertUnwindSafe(Self::close(&mut app, success))
                .catch_unwind()
                .await;
            drop(app);
            Some(closed)
        } else {
            None
        };
        if !success {
            eprintln!(
                "performance workload failed; cleanup completed={}",
                matches!(&cleanup, Some(Ok(Ok(()))))
            );
        }
        let mut result = match observed {
            Ok(Ok(result)) => result?,
            Ok(Err(_)) => {
                return Err(
                    AppError::Timeout("performance workload budget exhausted".into()).into(),
                );
            }
            Err(panic) => resume_unwind(panic),
        };
        match cleanup {
            Some(Ok(result)) => result?,
            Some(Err(panic)) => resume_unwind(panic),
            None => return Err("successful workload did not retain its App cleanup owner".into()),
        }
        // All run/workflow/App owners and the caller-created store have dropped.
        // A cleanup timeout remains a failure even if reattachment later joined.
        result["schema"] = json!(MANIFEST_SCHEMA);
        result["settings"] = Self::settings(actions);
        if let Some(probe) = &probe {
            probe.assert_idle();
            result["caller_sources"] = probe.report();
            result["settings"]
                .as_object_mut()
                .unwrap()
                .remove("history_row_ceiling");
            result["settings"]["private_task_counts"] =
                json!("not exposed; App shutdown proves joined ownership");
            result["settings"]["lower_agents_and_tool_handlers"] = json!("none constructed");
        }
        result["definition_bytes"] = json!(definition_bytes);
        result["process_id"] = json!(std::process::id());
        result["os"] = json!(std::env::consts::OS);
        result["architecture"] = json!(std::env::consts::ARCH);
        result["package_version"] = json!(env!("CARGO_PKG_VERSION"));
        result["cleanup_joined"] = json!(true);
        result["cleanup_ns"] = json!(closing.elapsed().as_nanos());
        let encoded = serde_json::to_vec(&result)?;
        assert!(
            encoded.len() <= 16_384,
            "result stays within the manifest bound"
        );
        if matches!(self.command, Command::SeedHistory { .. }) {
            let mut file = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(self.root.join("manifest.json"))?;
            file.write_all(&encoded)?;
            file.sync_all()?;
        }
        println!("{}", std::str::from_utf8(&encoded)?);
        Ok(())
    }

pub fn with_conditions<N: ConditionBackend>( self, conditions: Vec<N>, ) -> AppConfig<CS, A, X, N, T, H, E>

Replaces the complete caller-owned condition inventory before opening App.

App validates stable logical keys and revisions without evaluating the predicates. Replacing an inventory drops its previous values.

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::{ConditionBackend, ConditionDeclaration}};
struct Ready { declaration: ConditionDeclaration }
impl ConditionBackend for Ready {
    fn declaration(&self) -> &ConditionDeclaration { &self.declaration }
    fn evaluate(&self, state: &WorkflowState) -> Result<bool, AppError> {
        let count = state.values().get("count").and_then(serde_json::Value::as_i64)
            .ok_or_else(|| AppError::InvalidInput("count must be an integer".into()))?;
        Ok(count >= 3)
    }
}
let condition = Ready { declaration: ConditionDeclaration {
    key: "ready".into(), revision: "r1".into(), reads: vec!["count".into()],
} };
let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
let definition = WorkflowDefinition::parse(r#"{
    "schema":"app-kit.workflow/v2", "id":"counter", "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":{"increment":{"mode":"first","branches":[
        {"when":{"kind":"custom","condition":"ready","revision":"r1"},"target":{"kind":"end"}},
        {"target":{"kind":"node","id":"increment"}}
    ]}}
}"#)?;
let config = AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition]);
let config = config.with_conditions(vec![condition]);
let mut app = App::open(config).await?;
let workspace = app.create_workspace(WorkspaceConfig { name: None, root: std::env::current_dir()? }).await?;
let workflow = workspace.workflow("counter".parse()?).await?;
let terminal = workflow.start(ExecutionOptions::default()).await?.wait().await?;
assert!(matches!(terminal.status, RunStatus::Succeeded));
assert_eq!(terminal.state.values()["count"], 3);
app.shutdown(Shutdown::Drain { timeout: std::time::Duration::from_secs(5) }).await?;

pub fn with_state_typing<N: StateTyping>( self, typing: N, ) -> AppConfig<CS, A, X, C, N, H, E>

Replaces the caller-owned workflow typing policy before opening App.

The policy remains concrete and need not implement Clone. Its data values are local to validation or execution and need not be Send.

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::{StateTyping, StateChange}};
use serde_json::{Map, Value};
struct Counter;
impl StateTyping for Counter {
    type Data = i64;
    fn construct_data(&self, state: &WorkflowState) -> Result<i64, AppError> {
        state.values().get("count").and_then(Value::as_i64)
            .ok_or_else(|| AppError::InvalidInput("count must be an integer".into()))
    }
    fn construct_state(&self, count: &i64) -> Result<WorkflowState, AppError> {
        WorkflowState::new(Map::from_iter([("count".into(), Value::from(*count))]))
    }
    fn validate_pre(&self, _state: &WorkflowState, change: &StateChange) -> Result<(), AppError> {
        if let StateChange::Increment { values } = change {
            if values.iter().any(|(_, delta)| delta.as_i64().is_some_and(|value| value < 0)) {
                return Err(AppError::InvalidInput("count cannot decrease".into()));
            }
        }
        Ok(())
    }
    fn validate_post(&self, before: &WorkflowState, after: &WorkflowState, _change: &StateChange) -> Result<(), AppError> {
        if self.construct_data(after)? < self.construct_data(before)? {
            return Err(AppError::InvalidInput("count cannot decrease".into()));
        }
        Ok(())
    }
    fn validate_state(&self, state: &WorkflowState) -> Result<(), AppError> {
        if self.construct_data(state)? < 0 {
            return Err(AppError::InvalidInput("count must be nonnegative".into()));
        }
        Ok(())
    }
}
let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
let definition = WorkflowDefinition::parse(r#"{
    "schema":"app-kit.workflow/v2", "id":"counter", "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]);
let config = config.with_state_typing(Counter);
let mut app = App::open(config).await?;
let workspace = app.create_workspace(WorkspaceConfig { name: None, root: std::env::current_dir()? }).await?;
let workflow = workspace.workflow("counter".parse()?).await?;
let invalid = WorkflowState::new(serde_json::from_str(r#"{"count":-1}"#)?)?;
assert!(workflow.prepare(invalid).await.is_err());
app.shutdown(Shutdown::RejectActive).await?;

pub fn with_hooks<N: LifecycleHooks>( self, hooks: N, ) -> AppConfig<CS, A, X, C, T, N, E>

Replaces the caller-owned lifecycle observer before opening App.

Replacement drops the prior observer without invoking it. The new observer retains its concrete type and need not implement Clone.

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::{LifecycleEvent, LifecycleHooks}};
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
struct CountEvents(Arc<AtomicUsize>);
impl LifecycleHooks for CountEvents {
    fn observe(&self, event: &LifecycleEvent<'_>) -> Result<(), AppError> {
        if matches!(event, LifecycleEvent::ExecuteStarted { .. }) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }
        Ok(())
    }
    fn observe_async<'a>(&'a self, event: &'a LifecycleEvent<'a>)
        -> impl std::future::Future<Output = Result<(), AppError>> + Send + 'a {
        async move {
            if matches!(event, LifecycleEvent::ExecuteStarted { .. }) {
                self.0.fetch_add(10, Ordering::SeqCst);
            }
            Ok(())
        }
    }
}
let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
let definition = WorkflowDefinition::parse(r#"{
    "schema":"app-kit.workflow/v2", "id":"counter", "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]);
let count = Arc::new(AtomicUsize::new(0));
let config = config.with_hooks(CountEvents(Arc::clone(&count)));
let mut app = App::open(config).await?;
let workspace = app.create_workspace(WorkspaceConfig { name: None, root: std::env::current_dir()? }).await?;
let workflow = workspace.workflow("counter".parse()?).await?;
let mut run = workflow.prepare(WorkflowState::default()).await?;
run.execution(ExecutionOptions::default()).await?.finish().await?;
assert_eq!(count.load(Ordering::SeqCst), 11);
app.shutdown(Shutdown::Drain { timeout: std::time::Duration::from_secs(5) }).await?;

pub fn with_child_executor<N: ChildExecutor>( self, executor: N, ) -> AppConfig<CS, A, X, C, T, H, N>

Replaces the caller-owned admitted child executor before opening App.

The executor remains concrete. Its associated execution futures are created only for admitted child work and need not implement Send.

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};
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(app_kit::workflow::InlineExecutor);
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?;