Struct Workflow 

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

A workflow selected in one application workspace.

See execution and retained history.

Implementations

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

pub fn id(&self) -> &WorkflowId

Returns this workflow’s stable identity.

See execution and retained history.

pub async fn start( &self, options: ExecutionOptions, ) -> Result<WorkflowRun<CS, A, X, C, T, H, E>, AppError>

Atomically admits one run at checkpoint zero.

Schedules background advancement after admission. Dropping the returned run observer does not stop it. WorkflowRun::wait returns at an actionable or terminal boundary, so inspect status and interactions before assuming success.

Errors

Returns InvalidInput for invalid initial state/options or missing bindings, Busy for exhausted admission/capacity, Closed for sealed ownership, or Conflict for stale workspace/revision evidence. Lower/storage failures use Integration; accepted execution failures appear in run status.

See execution and retained history.

Examples found in repository

examples/performance.rs (line 884)

    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"}),
        )
    }

pub async fn prepare( &self, state: WorkflowState, ) -> Result<WorkflowRun<CS, A, X, C, T, H, E>, AppError>

Admits checkpoint zero without executing the entry node.

The initial patch is validated and atomically stored using the same workspace and App admission fences as background execution. Observers can read the parked run, and cancellation uses its checkpoint fence.

Errors

Returns InvalidInput for invalid initial state or missing bindings, Busy for exhausted admission/capacity, Closed for sealed ownership, or Conflict for stale workspace/revision evidence or exhausted bounded transaction-contention retries. Concurrent preparations are not guaranteed to all succeed. Lower/storage failures use Integration; accepted execution failures appear in run status.

See execution and retained history.

Examples found in repository

examples/performance.rs (line 403)

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

examples/history_and_recovery.rs (line 74)

    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 (line 52)

    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 restore( &self, options: RestoreOptions, ) -> Result<WorkflowRun<CS, A, X, C, T, H, E>, AppError>

Copies a clean retained boundary into a fresh, paused run.

Source selection and new admission commit atomically. The source remains unchanged; actions run only after explicit execution of the returned run.

Errors

Returns NotFound for absent source/checkpoint without fallback, InvalidInput for invalid overrides or selectors, Busy for exhausted capacity, and Conflict when the selected boundary is unsafe to copy. Sealing/revocation returns Closed; inconsistent history or storage failures return Integration.

See execution and retained history.

Examples found in repository

examples/history_and_recovery.rs (lines 104-110)

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

pub async fn open_run( &self, selector: RunSelector, ) -> Result<WorkflowRun<CS, A, X, C, T, H, E>, AppError>

Reopens one run by typed identity.

Errors

Returns InvalidInput for a malformed exact occurrence, NotFound when the selected run/occurrence is absent, or Closed for sealed scope. Exact selection never falls back to Latest; inconsistent retained evidence is Integration.

See execution and retained history.

Examples found in repository

examples/history_and_recovery.rs (line 164)

    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 (lines 492-495)

    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"}),
        )
    }

pub async fn list_runs( &self, after: Option<RunSelector>, limit: NonZeroUsize, ) -> Result<Vec<RunSnapshot>, AppError>

Lists runs by logical identity, then exact occurrence token.

An Exact cursor continues after that occurrence; a Latest cursor skips every occurrence of its logical identity. This inventory order does not represent the native save order used when opening Latest.

Errors

Returns InvalidInput for a limit above 1,000 or malformed cursor, NotFound for an unavailable cursor, and Closed or Integration for revoked or unreadable scope.

See execution and retained history.

Examples found in repository

examples/performance.rs (line 747)

    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"}),
        )
    }