Struct Workspace 

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

An application workspace handle.

A typed projection into the owning App generation. Dropping it does not delete or close lower workspaces. Replacement uses a complete configuration and a revision fence; successful deletion revokes this projection. Retained terminal history can still be opened through its workflow after workspace deletion.

Example

let snapshot = workspace.snapshot().await?;
assert_eq!(&snapshot.id, workspace.id());
assert_eq!(snapshot.revision, 1);

Implementations

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

pub fn id(&self) -> &WorkspaceId

Returns this workspace’s stable identity.

Example

let id = workspace.id().clone();
assert_eq!(workspace.snapshot().await?.id, id);

Examples found in repository

examples/performance.rs (line 462)

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

examples/history_and_recovery.rs (line 146)

    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 snapshot(&self) -> Result<WorkspaceSnapshot, AppError>

Loads the current workspace snapshot.

Errors

Returns Closed after revocation or sealing, NotFound for absent coordination, Conflict during unresolved lifecycle progress, or Integration for unreadable committed evidence. A history-only workspace has no configuration snapshot.

Example

let snapshot = workspace.snapshot().await?;
assert_eq!(snapshot.revision, 1);
assert_eq!(snapshot.config.name.as_deref(), Some("local counter"));

pub async fn replace( &self, expected_revision: u64, config: WorkspaceConfig, ) -> Result<WorkspaceSnapshot, AppError>

Atomically replaces the complete workspace configuration.

This is full replacement, including the name and root. Preflight checks all lower workspaces before fan-out. A partial lower failure retains the old App revision and durable reverse compensation; preflight never reserves a revision.

Errors

Returns Conflict for a stale revision, a nonterminal run pinned to the current revision, or partial lower fan-out (even if compensation succeeds). Invalid configuration, capacity, revocation, storage/lower failures and bounded waits report InvalidInput, Busy, Closed, Integration, or Timeout.

Example

let before = workspace.snapshot().await?;
let after = workspace.replace(before.revision, WorkspaceConfig {
    name: Some("reviewed".into()), root: before.config.root,
}).await?;
assert_eq!(after.revision, before.revision + 1);
assert_eq!(after.config.name.as_deref(), Some("reviewed"));

pub async fn workflow( &self, id: WorkflowId, ) -> Result<Workflow<CS, A, X, C, T, H, E>, AppError>

Opens a configured workflow in this workspace.

Selects the current configured definition for new work. A deleted workspace may retain a history-only workflow when exact terminal evidence still exists.

Errors

Returns NotFound for an unavailable definition or retained scope, Closed for revoked admission, or Integration for inconsistent durable evidence.

Example

let workflow = workspace.workflow("counter".parse()?).await?;
assert_eq!(workflow.id().to_string(), "counter");

Examples found in repository

examples/performance.rs (line 402)

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

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

    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/actions_and_inputs.rs (line 51)

    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 delete(&mut self, expected_revision: u64) -> Result<(), AppError>

Deletes this workspace after proving its revision fence.

Seals new admission, requests cancellation, and deletes lower workspaces only after every run is terminal with no unresolved effect. Progress is durable and resumable; terminal run history is retained until explicit run deletion.

Errors

Returns Conflict for a stale revision or incompatible deletion intent, Busy while unresolved work prevents deletion, and Closed for revoked ownership. Invalid fences, storage/lower failures, or expired observation return InvalidInput, Integration, or Timeout.

Example

let revision = workspace.snapshot().await?.revision;
workspace.delete(revision).await?;
assert!(app.list_workspaces(None, std::num::NonZeroUsize::new(10).unwrap()).await?.is_empty());