Struct App 

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

An open application generation.

Owns one durable application generation, its transferred agents and accepted work.

Supply a concrete storage_kit::StateStoreBackend through AppConfig. Each AgentSpec preserves its concrete agents_kit::AgentBackend and its associated Workspace/Conversation types. An empty agent inventory can execute state-only workflows; the explicit Agent type below constructs no Agent value.

Keep the Tokio runtime and any caller-owned lower runtimes alive and driven through Self::shutdown. Root drop revokes admission and requests supervised best-effort cleanup; it cannot substitute for awaited shutdown. Workspace and run observers do not keep admission alive after the App is revoked.

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 runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
runtime.block_on(async {
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 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 run = workflow.start(ExecutionOptions::default()).await?;
let snapshot = run.wait().await?;
assert!(matches!(snapshot.status, RunStatus::Succeeded));
assert_eq!(snapshot.state.values()["count"], 1);
app.shutdown(Shutdown::Drain { timeout: std::time::Duration::from_secs(5) }).await?;
})?;

Implementations

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

pub async fn open( config: AppConfig<CS, A, X, C, T, H, E>, ) -> Result<Self, AppError>

Opens or reconciles one application generation.

Validates bindings and retained records, claims one durable generation, and reconstructs recovery ownership before returning. A successful newer opener fences handles from the previous generation. Configuration is consumed.

Durable reopen and deployment

Reopen with a fresh handle to the same caller-owned StateStore, the same compatible workflow definitions, and freshly constructed or reopened lower integrations. Never reuse a consumed Agent or infer a new logical identity from its display representation. The state-only example below needs no Agent value and demonstrates that shutdown retains history until explicit deletion.

Existing App v6 records require a cold, one-way v6-to-v7 migration. Stop older clients, settle all their owners and take a whole-store physical backup before opening with this version. Opening performs the supported migration before claiming the new generation. Mixed-version operation, in-place downgrade, reset and corruption repair are not supported. Restore the physical backup for a rollback deployment; do not open migrated data with an older client. Unsupported epochs and corrupt current records fail rather than falling back to another decoder.

After Agent ownership transfer, ordinary failure awaits consuming Agent cleanup and preserves the primary failure. Cancelling this open future during transfer does not promise graceful cleanup: reopen the durable lower products and App to reconcile their retained facts.

Errors

Returns AppError::InvalidInput for invalid or conflicting construction metadata, AppError::Conflict for incompatible retained revisions or lost ownership, and AppError::Integration for unsupported durable versions, invalid retained records, or lower failures. Recovery can also report capacity, closure, or deadline errors; no live App is returned on failure.

Example

use agents_kit::Agent;
use app_kit::{App, AppConfig, Shutdown, WorkspaceConfig};
use app_kit::workflow::{ExecutionOptions, RunSelector, RunStatus, WorkflowDefinition};
use storage_kit::StateStore;
use storage_kit::state::config::{LanceStateStoreConfig, StateStoreConfig};
use std::time::Duration;
let directory = std::env::temp_dir().join(format!("ai-kit-app-doc-{}", uuid::Uuid::new_v4()));
std::fs::create_dir(&directory)?;
let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
let result = runtime.block_on(async {
    let definition_source = 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 = StateStoreConfig::lance(LanceStateStoreConfig::directory(&directory));
    let store = StateStore::connect(config.clone()).await?;
    let mut app = App::open(AppConfig::<_, Agent>::new(store, Vec::new(), vec![WorkflowDefinition::parse(definition_source)?])).await?;
    let first = async {
        let workspace = app.create_workspace(WorkspaceConfig { name: None, root: directory.clone() }).await?;
        let workflow = workspace.workflow("counter".parse()?).await?;
        let run = workflow.start(ExecutionOptions::default()).await?;
        let terminal = run.wait().await?;
        assert!(matches!(terminal.status, RunStatus::Succeeded));
        Ok::<_, app_kit::AppError>((workspace.id().clone(), run.id().clone()))
    }.await;
    let shutdown = app.shutdown(Shutdown::CancelAndDrain { timeout: Duration::from_secs(5) }).await;
    drop(app);
    let (workspace_id, run_id) = first?;
    shutdown?;

    let fresh_store = StateStore::connect(config).await?;
    let mut reopened = App::open(AppConfig::<_, Agent>::new(fresh_store, Vec::new(), vec![WorkflowDefinition::parse(definition_source)?])).await?;
    let readback = async {
        let workspace = reopened.workspace(workspace_id).await?;
        let workflow = workspace.workflow("counter".parse()?).await?;
        let mut run = workflow.open_run(RunSelector::Latest(run_id.clone())).await?;
        let terminal = run.snapshot().await?;
        assert!(matches!(terminal.status, RunStatus::Succeeded));
        assert_eq!(terminal.state.values()["count"], 1);
        run.delete(terminal.checkpoint).await?;
        assert!(matches!(workflow.open_run(RunSelector::Latest(run_id)).await,
            Err(app_kit::AppError::NotFound(_))));
        Ok::<(), app_kit::AppError>(())
    }.await;
    let shutdown = reopened.shutdown(Shutdown::CancelAndDrain { timeout: Duration::from_secs(5) }).await;
    drop(reopened);
    readback?;
    shutdown?;
    Ok::<(), Box<dyn std::error::Error>>(())
});
// A shutdown timeout bounds the waiter, not necessarily the owned finalizer.
// Remove the directory only after both generations demonstrably settled.
if result.is_ok() {
    std::fs::remove_dir_all(&directory)?;
} else {
    eprintln!("Retain {} for cleanup reconciliation", directory.display());
}
result?;

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 (lines 39-42)

    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 1106-1114)

    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 async fn create_workspace( &self, config: WorkspaceConfig, ) -> Result<Workspace<CS, A, X, C, T, H, E>, AppError>

Creates and materializes one workspace.

Materializes the complete configured Agent key/revision inventory before publishing the App workspace. Durable intent precedes each lower operation; unresolved partial progress remains recoverable rather than a visible success.

Errors

Returns AppError::InvalidInput for invalid configuration, AppError::Busy when admission is full, or AppError::Closed after revocation. Conflicting materialization evidence, lower/storage failures, and bounded waits report Conflict, Integration, or Timeout respectively.

Example

let workspace = app.create_workspace(WorkspaceConfig {
    name: Some("review".into()), root: std::env::current_dir()?,
}).await?;
assert_eq!(workspace.snapshot().await?.revision, 1);

Examples found in repository

examples/performance.rs (lines 397-400)

    async fn seed(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        actions: u64,
        linear: bool,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: Some("performance consumer".into()),
                root: self.root.join("workspace"),
            })
            .await?;
        let workflow = workspace.workflow("performance".parse()?).await?;
        let mut run = workflow.prepare(WorkflowState::default()).await?;
        let mut execution = run
            .execution(ExecutionOptions {
                halt_after: if linear {
                    Vec::new()
                } else {
                    vec!["count".into()]
                },
                ..ExecutionOptions::default()
            })
            .await?;
        let mut samples = Vec::with_capacity(actions as usize);
        let started = Instant::now();
        for ordinal in 0..actions {
            let step_started = Instant::now();
            let event = execution.step().await?;
            samples.push(step_started.elapsed());
            let Some(ExecutionEvent::Committed { outcomes, .. }) = event else {
                panic!("each serial workload step must commit one action");
            };
            assert_eq!(outcomes.len(), 1);
            assert_eq!(outcomes[0].ordinal, ordinal);
            assert_eq!(
                outcomes[0].node,
                if linear {
                    format!("n{ordinal:04}")
                } else {
                    "count".into()
                }
            );
            assert!(outcomes[0].result.is_empty());
            if !linear && ordinal == 0 {
                // One public halt and explicit resume add two control rows to
                // Initial + three checkpoints per action + Terminal.
                let halted = execution.finish().await?;
                assert!(matches!(&halted.status, RunStatus::Halted { node, result }
                    if node == "count" && result == &Some(Map::new())));
                assert_eq!(halted.checkpoint, 4);
                assert_eq!(
                    halted.state.values(),
                    &Map::from_iter([("count".into(), json!(1))])
                );
                assert_eq!(halted.last_action_ordinal, Some(0));
                assert!(halted.outcomes == outcomes);
                assert!(halted.interactions.is_empty());
                execution = run.execution(ExecutionOptions::default()).await?;
            }
        }
        let snapshot = execution.finish().await?;
        Self::verify_snapshot(&snapshot, actions);
        let execution_elapsed = started.elapsed();
        let history = self
            .history(&run, actions, linear, snapshot.checkpoint)
            .await?;
        if let Command::SeedHistory { rows } = self.command {
            assert_eq!(history["rows"], rows);
        }
        Ok(
            json!({"dataset":if linear {"linear"} else {"history"},"actions":actions,
            "workspace":workspace.id().to_string(),"run":run.id().to_string(),
            "occurrence":snapshot.occurrence,
            "expected_state":snapshot.state.values(),"last_action_ordinal":snapshot.last_action_ordinal,
            "terminal_checkpoint":snapshot.checkpoint,"retained_rows":history["rows"],
            "execution_ns":execution_elapsed.as_nanos(),
            "actions_per_second":actions as f64 / execution_elapsed.as_secs_f64(),
            "step_latency":Self::samples(samples),"history":history}),
        )
    }

    async fn reopen(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        manifest: &Value,
        actions: u64,
        open_only: bool,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let started = Instant::now();
        let workspace: WorkspaceId = manifest["workspace"]
            .as_str()
            .ok_or("missing workspace ID")?
            .parse()?;
        let run_id: RunId = manifest["run"].as_str().ok_or("missing run ID")?.parse()?;
        let occurrence = manifest["occurrence"]
            .as_str()
            .ok_or("missing run occurrence")?
            .to_owned();
        let workspace = app.workspace(workspace).await?;
        let workflow = workspace.workflow("performance".parse()?).await?;
        let run = workflow
            .open_run(RunSelector::Exact {
                id: run_id,
                occurrence: occurrence.clone(),
            })
            .await?;
        let snapshot = run.snapshot().await?;
        Self::verify_snapshot(&snapshot, actions);
        assert_eq!(snapshot.occurrence, occurrence);
        assert_eq!(json!(snapshot.checkpoint), manifest["terminal_checkpoint"]);
        assert_eq!(json!(snapshot.state.values()), manifest["expected_state"]);
        assert_eq!(
            json!(snapshot.last_action_ordinal),
            manifest["last_action_ordinal"]
        );
        let lookup = started.elapsed();
        if open_only {
            return Ok(
                json!({"dataset":"history","actions":actions,"mode":"open-only",
                "lookup_and_snapshot_ns":lookup.as_nanos(),
                "seed_retained_rows":manifest["retained_rows"],"observed_history_rows":null}),
            );
        }
        let history = self
            .history(&run, actions, false, snapshot.checkpoint)
            .await?;
        assert_eq!(history["rows"], manifest["retained_rows"]);
        Ok(
            json!({"dataset":"history","actions":actions,"mode":"full-history",
            "lookup_and_snapshot_ns":lookup.as_nanos(),"history":history}),
        )
    }

    fn definitions(&self, actions: u64) -> Result<(Vec<WorkflowDefinition>, usize), AppError> {
        if matches!(
            self.command,
            Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. }
        ) {
            let (definition, bytes) =
                Self::definition(matches!(self.command, Command::Linear), actions)?;
            return Ok((vec![definition], bytes));
        }
        let leaf = json!({
            "schema":"app-kit.workflow/v2","id":"performance-leaf","revision":"r1",
            "state":{"value":0,"result":0},"required":["value","result"],
            "reducers":{"value":"reject_concurrent","result":"reject_concurrent"},
            "entry":"work","nodes":[{"id":"work","deadline_ms":NODE_DEADLINE_MS,
                "step":{"kind":"action","action":"performance-action","revision":"r1"}}],"routes":{}
        });
        let documents = match self.command {
            Command::Ready => {
                let mut nodes = vec![json!({"id":"fork","step":{"kind":"state","mutations":[]}})];
                let mut branches = Vec::new();
                for index in 0..32 {
                    let node = format!("n{index:02}");
                    nodes.push(json!({"id":node,"deadline_ms":NODE_DEADLINE_MS,
                        "step":{"kind":"action","action":"performance-action","revision":"r1"}}));
                    branches.push(json!({"target":{"kind":"node","id":node}}));
                }
                vec![json!({
                    "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
                    "state":{"value":0,"result":0},"required":["value","result"],
                    "reducers":{"value":"reject_concurrent","result":"collect"},
                    "entry":"fork","nodes":nodes,
                    "routes":{"fork":{"mode":"broadcast","branches":branches}}
                })]
            }
            Command::Map | Command::NestedCancel => {
                let nested = matches!(self.command, Command::NestedCancel);
                let items = if nested { 2 } else { 4_096 };
                let mut documents = vec![
                    json!({
                        "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
                        "state":{"value":0,"result":0},"required":["value","result"],
                        "reducers":{"value":"reject_concurrent","result":"sum_integer"},
                        "entry":"map","nodes":[{"id":"map","deadline_ms":if nested {NODE_DEADLINE_MS} else {WORKLOAD_BUDGET.as_millis() as u64},
                            "step":{"kind":"map","workflow":if nested {"performance-mid"} else {"performance-leaf"},
                            "items":{"kind":"literal","value":(0..items).collect::<Vec<_>>()},
                            "input_key":"value","max_parallel":if nested {2} else {8},
                            "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
                    }),
                    leaf,
                ];
                if nested {
                    documents.push(json!({
                        "schema":"app-kit.workflow/v2","id":"performance-mid","revision":"r1",
                        "state":{"value":0,"result":0},"required":["value","result"],
                        "reducers":{"value":"reject_concurrent","result":"sum_integer"},
                        "entry":"map","nodes":[{"id":"map","deadline_ms":NODE_DEADLINE_MS,
                            "step":{"kind":"map","workflow":"performance-leaf",
                            "items":{"kind":"literal","value":(0..8).collect::<Vec<_>>()},
                            "input_key":"value","max_parallel":8,
                            "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
                    }));
                }
                documents
            }
            Command::Repeated | Command::DelayedObserver => vec![leaf],
            Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => {
                unreachable!()
            }
        };
        let mut bytes = 0;
        let mut definitions = Vec::with_capacity(documents.len());
        for document in documents {
            let source = document.to_string();
            assert!(source.len() <= 1_048_576);
            bytes += source.len();
            definitions.push(WorkflowDefinition::parse(&source)?);
        }
        Ok((definitions, bytes))
    }

    async fn ready(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        probe: &WorkloadProbe,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: None,
                root: self.root.join("workspace"),
            })
            .await?;
        let workflow = workspace.workflow("performance".parse()?).await?;
        let mut waves = Vec::with_capacity(32);
        let mut releases = Vec::with_capacity(32 * 31);
        let mut barriers = Vec::with_capacity(32);
        let started = Instant::now();
        for permutation in 0..32 {
            probe.assert_idle();
            let mut run = workflow.prepare(WorkflowState::default()).await?;
            let observer = workflow
                .open_run(RunSelector::Latest(run.id().clone()))
                .await?;
            let mut execution = run.execution(ExecutionOptions::default()).await?;
            let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
                panic!("the native fork must commit before the ready wave");
            };
            assert_eq!(outcomes.len(), 1);
            assert_eq!(
                (outcomes[0].node.as_str(), outcomes[0].ordinal),
                ("fork", 0)
            );
            let wave = Instant::now();
            let mut finish = Box::pin(execution.finish());
            tokio::select! {
                _ = probe.wait_for(&probe.constructed, (permutation + 1) * 32) => {}
                result = &mut finish => panic!("gated ready wave finished before admission: {}", result.is_ok()),
            }
            assert_eq!(probe.active.load(Ordering::SeqCst), 32);
            let order = (0..32)
                .map(|offset| (permutation + 31 - offset) % 32)
                .collect::<Vec<_>>();
            for (released, index) in order.iter().take(31).enumerate() {
                let releasing = Instant::now();
                probe.release(*index);
                tokio::select! {
                    _ = probe.wait_for(&probe.completed, permutation * 32 + released + 1) => {}
                    result = &mut finish => panic!("a partial ready wave committed: {}", result.is_ok()),
                }
                releases.push(releasing.elapsed());
                let visible = observer.snapshot().await?;
                assert_eq!(
                    visible.state.values()["result"],
                    0,
                    "staged siblings cannot expose a partial barrier"
                );
                assert_eq!(visible.last_action_ordinal, Some(0));
            }
            let barrier = Instant::now();
            probe.release(order[31]);
            let snapshot = finish.await?;
            barriers.push(barrier.elapsed());
            waves.push(wave.elapsed());
            assert!(matches!(snapshot.status, RunStatus::Succeeded));
            assert!(snapshot.interactions.is_empty());
            assert_eq!(
                snapshot.state.values()["result"],
                json!((1..=32).collect::<Vec<_>>())
            );
            assert_eq!(snapshot.last_action_ordinal, Some(32));
            assert_eq!(snapshot.outcomes.len(), 32);
            for (index, outcome) in snapshot.outcomes.iter().enumerate() {
                assert_eq!(outcome.node, format!("n{index:02}"));
                assert_eq!(outcome.ordinal, index as u64 + 1);
                assert_eq!(
                    outcome.result,
                    Map::from_iter([("value".into(), json!(index + 1))])
                );
            }
            assert_eq!(std::mem::take(&mut *probe.order.lock().unwrap()), order);
            assert_eq!(
                probe.completed.load(Ordering::SeqCst),
                (permutation + 1) * 32
            );
            probe.assert_idle();
        }
        assert_eq!(probe.peak.load(Ordering::SeqCst), 32);
        let elapsed = started.elapsed();
        Ok(
            json!({"dataset":"ready32","permutations":32,"ready_actions_per_wave":32,
            "caller_action_commits":1024,"native_fork_commits":32,
            "execution_ns":elapsed.as_nanos(),"caller_actions_per_second":1024.0 / elapsed.as_secs_f64(),
            "wave_latency_including_snapshot_probes":Self::samples(waves),
            "release_to_source_completion_latency":Self::samples(releases),
            "last_release_to_committed_barrier_latency":Self::samples(barriers)}),
        )
    }

    async fn mapped(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        probe: &WorkloadProbe,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: None,
                root: self.root.join("workspace"),
            })
            .await?;
        let workflow = workspace.workflow("performance".parse()?).await?;
        let mut run = workflow.prepare(WorkflowState::default()).await?;
        let observer = workflow
            .open_run(RunSelector::Latest(run.id().clone()))
            .await?;
        let started = Instant::now();
        let mut finish = Box::pin(run.execution(ExecutionOptions::default()).await?.finish());
        tokio::select! {
            _ = probe.wait_for(&probe.constructed, 8) => {}
            result = &mut finish => panic!("map finished before its first eight blocked sources: {}", result.is_ok()),
        }
        assert_eq!(probe.active.load(Ordering::SeqCst), 8);
        assert_eq!(probe.constructed.load(Ordering::SeqCst), 8);
        let unchanged = observer.snapshot().await?;
        assert_eq!(unchanged.state.values()["result"], 0);
        assert!(unchanged.last_action_ordinal.is_none());
        probe.release_all();
        let snapshot = finish.await?;
        let elapsed = started.elapsed();
        assert!(matches!(snapshot.status, RunStatus::Succeeded));
        assert!(snapshot.interactions.is_empty());
        assert_eq!(snapshot.state.values()["result"], 4_096_u64 * 4_097 / 2);
        assert_eq!(snapshot.last_action_ordinal, Some(0));
        assert_eq!(snapshot.outcomes.len(), 1);
        assert_eq!(snapshot.outcomes[0].node, "map");
        assert_eq!(probe.constructed.load(Ordering::SeqCst), 4_096);
        assert_eq!(probe.completed.load(Ordering::SeqCst), 4_096);
        assert_eq!(probe.peak.load(Ordering::SeqCst), 8);
        probe.assert_idle();
        let child = workspace.workflow("performance-leaf".parse()?).await?;
        let mut cursor = None;
        let mut values = BTreeSet::new();
        let verifying = Instant::now();
        loop {
            let page = child
                .list_runs(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
                .await?;
            assert!(page.len() <= PAGE_SIZE);
            if page.is_empty() {
                break;
            }
            for row in &page {
                assert!(matches!(row.status, RunStatus::Succeeded));
                let value = row.state.values()["value"].as_u64().unwrap();
                assert!(
                    value < 4_096 && values.insert(value),
                    "every mapped input has exactly one retained child"
                );
                assert_eq!(row.state.values()["result"], value + 1);
                assert_eq!(row.last_action_ordinal, Some(0));
                assert_eq!(row.outcomes.len(), 1);
                assert_eq!(row.outcomes[0].node, "work");
            }
            let last = page.last().unwrap();
            cursor = Some(RunSelector::Exact {
                id: last.id.clone(),
                occurrence: last.occurrence.clone(),
            });
            assert!(values.len() <= 4_096);
        }
        assert_eq!(values.len(), 4_096);
        Ok(json!({"dataset":"map4096","items":4096,"max_parallel":8,
            "parent_node_deadline_ms":WORKLOAD_BUDGET.as_millis(),
            "execution_ns":elapsed.as_nanos(),"child_actions_per_second":4096.0 / elapsed.as_secs_f64(),
            "map_completion_latency":Self::samples(vec![elapsed]),
            "child_inventory_verification_ns":verifying.elapsed().as_nanos(),"verified_children":values.len()}))
    }

    async fn repeated(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        probe: &WorkloadProbe,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: None,
                root: self.root.join("workspace"),
            })
            .await?;
        let workflow = workspace.workflow("performance-leaf".parse()?).await?;
        let started = Instant::now();
        // Admit one run at a time, then directly poll at most 16 original
        // executions together. No consumer task is spawned or detached.
        let mut runs = std::pin::pin!(
            stream::iter(0..100_u64)
                .then(|value| {
                    let workflow = &workflow;
                    async move {
                        let started = Instant::now();
                        let run = workflow
                            .prepare(WorkflowState::new(Map::from_iter([(
                                "value".into(),
                                json!(value),
                            )]))?)
                            .await?;
                        Ok::<_, AppError>((value, run, started))
                    }
                })
                .map(|prepared| async move {
                    let (value, mut run, started) = prepared?;
                    let snapshot = run
                        .execution(ExecutionOptions::default())
                        .await?
                        .finish()
                        .await?;
                    assert!(matches!(snapshot.status, RunStatus::Succeeded));
                    assert!(snapshot.interactions.is_empty());
                    assert_eq!(snapshot.state.values()["value"], value);
                    assert_eq!(snapshot.state.values()["result"], value + 1);
                    assert_eq!(snapshot.last_action_ordinal, Some(0));
                    assert_eq!(snapshot.outcomes.len(), 1);
                    Ok::<_, AppError>((run.id().clone(), started.elapsed()))
                })
                .buffer_unordered(16)
        );
        tokio::select! {
            _ = probe.wait_for(&probe.constructed, 16) => {}
            result = runs.next() => panic!(
                "run finished before the first 16 blocked sources: {:?}",
                result.map(|result| result.map(|_| ())),
            ),
        }
        assert_eq!(probe.active.load(Ordering::SeqCst), 16);
        assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
        probe.release_all();
        let mut ids = BTreeSet::new();
        let mut samples = Vec::with_capacity(100);
        while let Some(result) = runs.next().await {
            let (id, elapsed) = result?;
            assert!(ids.insert(id));
            samples.push(elapsed);
        }
        assert_eq!(ids.len(), 100);
        assert_eq!(probe.constructed.load(Ordering::SeqCst), 100);
        assert_eq!(probe.completed.load(Ordering::SeqCst), 100);
        assert_eq!(probe.peak.load(Ordering::SeqCst), 16);
        probe.assert_idle();
        let rows = workflow
            .list_runs(None, NonZeroUsize::new(128).unwrap())
            .await?;
        assert_eq!(rows.len(), 100);
        assert_eq!(
            rows.iter()
                .map(|row| row.id.clone())
                .collect::<BTreeSet<_>>(),
            ids
        );
        assert!(
            rows.iter()
                .all(|row| matches!(row.status, RunStatus::Succeeded))
        );
        let elapsed = started.elapsed();
        Ok(
            json!({"dataset":"repeated100","runs":100,"caller_in_flight_limit":16,"admission":"sequential",
            "production_active_limit":"unchanged; this measures only the caller's 16-journey envelope",
            "execution_ns":elapsed.as_nanos(),"runs_per_second":100.0 / elapsed.as_secs_f64(),
            "run_latency_including_admission":Self::samples(samples),"indefinite_stability_proof":false}),
        )
    }

    async fn nested_cancel(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        probe: &WorkloadProbe,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: None,
                root: self.root.join("workspace"),
            })
            .await?;
        let workflow = workspace.workflow("performance".parse()?).await?;
        let run = workflow.start(ExecutionOptions::default()).await?;
        tokio::select! {
            _ = probe.wait_for(&probe.constructed, 16) => {}
            result = run.wait() => panic!("nested run ended before its held leaf sources: {}", result.is_ok()),
        }
        assert_eq!(probe.active.load(Ordering::SeqCst), 16);
        let cancelling = Instant::now();
        let snapshot = tokio::time::timeout(Duration::from_secs(30), async {
            loop {
                let before = run.snapshot().await?;
                assert_eq!(before.state.values()["result"], 0);
                assert!(before.outcomes.is_empty() && before.last_action_ordinal.is_none());
                match run.cancel(before.checkpoint).await {
                    Ok(_) => break,
                    Err(AppError::Conflict(_)) => tokio::task::yield_now().await,
                    Err(error) => return Err(error),
                }
            }
            run.wait().await
        })
        .await
        .map_err(|_| {
            AppError::Timeout("nested cancellation exceeded its declared 30-second grace".into())
        })??;
        let elapsed = cancelling.elapsed();
        assert!(matches!(snapshot.status, RunStatus::Cancelled));
        assert!(snapshot.interactions.is_empty() && snapshot.outcomes.is_empty());
        assert!(snapshot.last_action_ordinal.is_none());
        assert_eq!(snapshot.state.values()["result"], 0);
        assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
        assert_eq!(probe.completed.load(Ordering::SeqCst), 0);
        probe.assert_idle();
        for (id, count) in [
            ("performance", 1),
            ("performance-mid", 2),
            ("performance-leaf", 16),
        ] {
            let child = workspace.workflow(id.parse()?).await?;
            let rows = child
                .list_runs(None, NonZeroUsize::new(128).unwrap())
                .await?;
            assert_eq!(rows.len(), count);
            for row in rows {
                assert!(matches!(row.status, RunStatus::Cancelled));
                assert!(row.outcomes.is_empty() && row.last_action_ordinal.is_none());
                assert_eq!(row.state.values()["result"], 0);
            }
        }
        Ok(
            json!({"dataset":"nested-cancel","depth":2,"intermediate_children":2,"leaf_children":16,
            "cooperative_cancel_grace_ms":30_000,"cancellation_latency":Self::samples(vec![elapsed]),
            "source_replays":0,"caller_sources_dropped":16,"agent_unknown_replay":"not applicable: no Agent attempt"}),
        )
    }

    async fn delayed_observer(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        probe: &WorkloadProbe,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: None,
                root: self.root.join("workspace"),
            })
            .await?;
        let workflow = workspace.workflow("performance-leaf".parse()?).await?;
        let run = workflow.start(ExecutionOptions::default()).await?;
        let observer = workflow
            .open_run(RunSelector::Latest(run.id().clone()))
            .await?;
        // This public observation is deliberately never polled while the source
        // produces its 256 partials and commits. It owns no execution work.
        let delayed = observer.wait();
        let produced = run.wait().await?;
        assert!(matches!(produced.status, RunStatus::Succeeded));
        assert_eq!(probe.completed.load(Ordering::SeqCst), 1);
        probe.assert_idle();
        tokio::time::sleep(Duration::from_secs(2)).await;
        let observing = Instant::now();
        let snapshot = delayed.await?;
        let elapsed = observing.elapsed();
        assert!(matches!(snapshot.status, RunStatus::Succeeded));
        assert_eq!(snapshot.checkpoint, produced.checkpoint);
        assert_eq!(snapshot.state.values()["result"], 1);
        assert_eq!(snapshot.last_action_ordinal, Some(0));
        assert_eq!(snapshot.outcomes.len(), 1);
        assert_eq!(
            snapshot.outcomes[0].result,
            Map::from_iter([("value".into(), json!(1))])
        );
        assert_eq!(probe.constructed.load(Ordering::SeqCst), 1);
        assert_eq!(probe.partials.load(Ordering::SeqCst), 256);
        probe.assert_idle();
        let terminal = run.snapshot().await?;
        assert_eq!(terminal.checkpoint, snapshot.checkpoint);
        assert_eq!(terminal.state, snapshot.state);
        let mut cursor = None;
        let mut actions = 0;
        let mut rows = 0;
        loop {
            let page = run
                .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
                .await?;
            for row in &page.checkpoints {
                rows += 1;
                assert!(rows <= 32);
                if matches!(row.kind, CheckpointKind::Action) {
                    actions += 1;
                    assert_eq!(row.action_ordinal, Some(0));
                    assert_eq!(row.outcomes.len(), 1);
                }
            }
            cursor = page.next;
            if cursor.is_none() {
                break;
            }
        }
        assert_eq!(
            actions, 1,
            "partials cannot become extra durable action commits"
        );
        Ok(
            json!({"dataset":"delayed-observer","unobserved_partial_events":256,"delay_after_terminal_confirmation_ms":2_000,
            "observation_latency":Self::samples(vec![elapsed]),"retained_rows":rows,
            "memory_scope":"external process RSS; this consumer retains no partial payloads"}),
        )
    }

examples/history_and_recovery.rs (lines 68-71)

    async fn create_fork(
        &self,
        app: &App<StateStore, Agent>,
    ) -> Result<(WorkspaceId, RunSelector, RunSelector), Box<dyn Error + Send + Sync>> {
        let workspace = app
            .create_workspace(WorkspaceConfig {
                name: Some("history example".into()),
                root: std::env::current_dir()?,
            })
            .await?;
        let workflow = workspace.workflow("history-counter".parse()?).await?;
        let mut source = workflow.prepare(WorkflowState::default()).await?;
        let completed = source
            .execution(ExecutionOptions::default())
            .await?
            .finish()
            .await?;
        if !matches!(completed.status, RunStatus::Succeeded)
            || completed.state.values()["count"] != 2
        {
            return Err(
                AppError::Integration("source did not complete both increments".into()).into(),
            );
        }

        // An action ordinal selects the whole committed boundary containing it;
        // it is independent of the checkpoint sequence and storage revision.
        let first = source.checkpoint(CheckpointSelector::Action(0)).await?;
        if !matches!(first.kind, CheckpointKind::Action)
            || first.position.as_deref() != Some("first")
            || first.state.values()["count"] != 1
        {
            return Err(
                AppError::Integration("first action history is inconsistent".into()).into(),
            );
        }
        let source_selector = RunSelector::Exact {
            id: source.id().clone(),
            occurrence: first.occurrence.clone(),
        };
        let mut fork = workflow
            .restore(RestoreOptions {
                source: source_selector.clone(),
                checkpoint: CheckpointSelector::Action(0),
                overrides: WorkflowState::new(Map::from_iter([("count".into(), json!(40))]))?,
                position: ResumePosition::AfterCompleted,
                fallback: None,
            })
            .await?;
        let parked = fork.snapshot().await?;
        if !matches!(parked.status, RunStatus::Paused)
            || parked.checkpoint != 0
            || parked.state.values()["count"] != 40
            || !parked.outcomes.is_empty()
        {
            return Err(
                AppError::Integration("restore did not create a parked clean copy".into()).into(),
            );
        }

        source.delete(completed.checkpoint).await?;
        if !matches!(source.snapshot().await, Err(AppError::NotFound(_))) {
            return Err(
                AppError::Integration("explicit source deletion did not finish".into()).into(),
            );
        }
        // Deletion does not invalidate the copied state. AfterCompleted skips
        // the first increment, so only the second changes the override: 40 -> 41.
        let completed = fork
            .execution(ExecutionOptions::default())
            .await?
            .finish()
            .await?;
        if !matches!(completed.status, RunStatus::Succeeded)
            || completed.state.values()["count"] != 41
            || completed.last_action_ordinal != Some(1)
        {
            return Err(
                AppError::Integration("fork resumed from the wrong boundary".into()).into(),
            );
        }
        println!("Source count: 2; copied count: 40; completed fork count: 41");
        Ok((
            workspace.id().clone(),
            source_selector,
            RunSelector::Exact {
                id: fork.id().clone(),
                occurrence: completed.occurrence,
            },
        ))
    }

examples/actions_and_inputs.rs (lines 46-49)

    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(())
    }

pub async fn workspace( &self, id: WorkspaceId, ) -> Result<Workspace<CS, A, X, C, T, H, E>, AppError>

Reopens one workspace by typed identity.

A retained terminal head can establish a history-only scope after coordination is deleted. That handle permits retained-run lookup, not fresh admission.

Errors

Returns AppError::NotFound when neither active coordination nor readable retained history exists in this scope. Revocation, incompatible lifecycle progress, and unreadable durable evidence report Closed, Conflict, or Integration.

Example

let id = workspace.id().clone();
let reopened = app.workspace(id.clone()).await?;
assert_eq!(reopened.id(), &id);

Examples found in repository

examples/history_and_recovery.rs (line 162)

    async fn read_reopened(
        &self,
        app: &App<StateStore, Agent>,
        workspace: WorkspaceId,
        source: RunSelector,
        fork: RunSelector,
    ) -> Result<(), AppError> {
        let workspace = app.workspace(workspace).await?;
        let workflow = workspace.workflow("history-counter".parse()?).await?;
        let fork = workflow.open_run(fork).await?;
        let committed = fork.checkpoint(CheckpointSelector::Action(1)).await?;
        let lineage = committed
            .lineage
            .as_ref()
            .ok_or_else(|| AppError::Integration("reopened fork lost its source lineage".into()))?;
        if committed.state.values()["count"] != 41
            || !matches!(&source, RunSelector::Exact { id, occurrence }
                if &lineage.source == id && &lineage.source_occurrence == occurrence)
            || lineage.action_ordinal != Some(0)
        {
            return Err(AppError::Integration(
                "reopened fork history changed".into(),
            ));
        }
        println!(
            "Reopened fork {} at count 41; source lineage remains readable",
            fork.id()
        );
        Ok(())
    }

examples/performance.rs (line 489)

    async fn reopen(
        &self,
        app: &App<StateStore, Agent, WorkloadAction>,
        manifest: &Value,
        actions: u64,
        open_only: bool,
    ) -> Result<Value, Box<dyn Error + Send + Sync>> {
        let started = Instant::now();
        let workspace: WorkspaceId = manifest["workspace"]
            .as_str()
            .ok_or("missing workspace ID")?
            .parse()?;
        let run_id: RunId = manifest["run"].as_str().ok_or("missing run ID")?.parse()?;
        let occurrence = manifest["occurrence"]
            .as_str()
            .ok_or("missing run occurrence")?
            .to_owned();
        let workspace = app.workspace(workspace).await?;
        let workflow = workspace.workflow("performance".parse()?).await?;
        let run = workflow
            .open_run(RunSelector::Exact {
                id: run_id,
                occurrence: occurrence.clone(),
            })
            .await?;
        let snapshot = run.snapshot().await?;
        Self::verify_snapshot(&snapshot, actions);
        assert_eq!(snapshot.occurrence, occurrence);
        assert_eq!(json!(snapshot.checkpoint), manifest["terminal_checkpoint"]);
        assert_eq!(json!(snapshot.state.values()), manifest["expected_state"]);
        assert_eq!(
            json!(snapshot.last_action_ordinal),
            manifest["last_action_ordinal"]
        );
        let lookup = started.elapsed();
        if open_only {
            return Ok(
                json!({"dataset":"history","actions":actions,"mode":"open-only",
                "lookup_and_snapshot_ns":lookup.as_nanos(),
                "seed_retained_rows":manifest["retained_rows"],"observed_history_rows":null}),
            );
        }
        let history = self
            .history(&run, actions, false, snapshot.checkpoint)
            .await?;
        assert_eq!(history["rows"], manifest["retained_rows"]);
        Ok(
            json!({"dataset":"history","actions":actions,"mode":"full-history",
            "lookup_and_snapshot_ns":lookup.as_nanos(),"history":history}),
        )
    }

pub async fn list_workspaces( &self, after: Option<WorkspaceId>, limit: NonZeroUsize, ) -> Result<Vec<WorkspaceSnapshot>, AppError>

Lists workspaces in canonical identity order.

The cursor is exclusive and pages contain active workspaces only. Limits are 1–1,000; larger requests are rejected rather than silently clamped.

Errors

Returns AppError::InvalidInput for an excessive limit, AppError::NotFound for an unknown cursor, and Closed or Integration for revoked or unreadable application state.

Example

let page = app.list_workspaces(None, std::num::NonZeroUsize::new(1).unwrap()).await?;
assert_eq!(&page[0].id, workspace.id());
let next = app.list_workspaces(Some(page[0].id.clone()), std::num::NonZeroUsize::new(1).unwrap()).await?;
assert!(next.is_empty());

pub async fn shutdown(&mut self, policy: Shutdown) -> Result<(), AppError>

Finalizes this application according to policy.

RejectActive leaves admission open if work remains. Drain policies seal admission; CancelAndDrain also requests explicit cancellation. Successful completion joins App-owned work and consumes each transferred Agent once. The timeout bounds this observer, not the continuing finalizer. Cancelling the observer or timing out never reopens admission or detaches accepted cleanup.

Errors

Returns AppError::Busy when RejectActive encounters accepted work, AppError::InvalidInput for an invalid timeout, or AppError::Timeout when observation expires. Durable conflicts and lower cleanup failures remain Conflict or Integration; cleanup does not erase an earlier failure.

Example

app.shutdown(Shutdown::Drain { timeout: std::time::Duration::from_secs(5) }).await?;

Examples found in repository

examples/history_and_recovery.rs (lines 189-191)

    async fn close(&self, app: &mut App<StateStore, Agent>) -> Result<(), AppError> {
        tokio::time::timeout(
            Duration::from_secs(15),
            app.shutdown(Shutdown::CancelAndDrain {
                timeout: Duration::from_secs(10),
            }),
        )
        .await
        .map_err(|_| AppError::Timeout("example App cleanup timed out".into()))?
    }

examples/performance.rs (line 1036)

    async fn close(
        app: &mut App<StateStore, Agent, WorkloadAction>,
        success: bool,
    ) -> Result<(), AppError> {
        let deadline = Instant::now() + CLEANUP_BUDGET;
        let mut first_failure = None;
        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(first_failure.unwrap_or_else(|| {
                    AppError::Timeout("performance cleanup budget exhausted".into())
                }));
            }
            let policy = if success {
                Shutdown::Drain {
                    timeout: CLEANUP_OBSERVATION,
                }
            } else {
                Shutdown::CancelAndDrain {
                    timeout: CLEANUP_OBSERVATION,
                }
            };
            let result =
                tokio::time::timeout(remaining.min(CLEANUP_OBSERVATION), app.shutdown(policy))
                    .await;
            match result {
                Ok(Ok(())) => return first_failure.map_or(Ok(()), Err),
                Ok(Err(error @ AppError::Timeout(_))) => {
                    first_failure.get_or_insert(error);
                }
                Err(_) => {
                    first_failure.get_or_insert_with(|| {
                        AppError::Timeout("performance cleanup observation timed out".into())
                    });
                }
                Ok(Err(error)) => return Err(first_failure.unwrap_or(error)),
            }
            // A timed-out observer reattaches to its original finalizer.
        }
    }

examples/actions_and_inputs.rs (lines 99-101)

    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(())
    }

Trait Implementations

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

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.