Skip to main content

Module workflow

Module workflow 

Source
Expand description

Opaque workflow definitions and runtime handles. Workflow authoring, execution, immutable state, and retained history.

Parse strict versioned JSON with WorkflowDefinition::parse, inspect its GraphDescription, and transfer definitions through crate::AppConfig. Workflow::start drives accepted work in the background; Workflow::prepare admits checkpoint zero for explicit WorkflowRun::execution or SyncExecution. Synchronous calls require an outside thread and a runtime that stays driven elsewhere. Both lanes share the same commit authority.

ExecutionEvent::Partial never commits state. ExecutionEvent::Committed identifies a durable boundary; action ordinals and checkpoint sequences are different coordinates. WorkflowRun::history reads a fixed cutoff, and Workflow::restore copies clean retained state into a new paused run. History remains until explicit WorkflowRun::delete.

Concrete behavior belongs to ActionBackend, ConditionBackend, StateTyping, LifecycleHooks and ChildExecutor. The child executor’s associated future can remain local/non-Send without future erasure. StateSerialization separately registers JSON field/tag codecs only. Supporting DTOs describe inputs, results and controls; runtime ownership stays private. Run observers do not cancel work when dropped; execution owners revoke uncommitted advancement while preserving admitted native commits.

§Execute, observe, restore and delete

This local workflow admits a paused run, explicitly drives its lazy execution, and then reads immutable history. Merely obtaining a run snapshot does not advance a prepared run. Restore copies a retained checkpoint into another paused run; deletion removes only the selected terminal history, not its independent copy. The example needs no credentials, Agent value or subprocess.

use agents_kit::Agent;
use app_kit::{App, AppConfig, AppError, Shutdown, WorkspaceConfig};
use app_kit::workflow::{CheckpointSelector, ExecutionEvent, ExecutionOptions,
    RestoreOptions, ResumePosition, RunSelector, RunStatus, WorkflowDefinition, WorkflowState};
use storage_kit::StateStore;
use storage_kit::state::config::{LanceStateStoreConfig, StateStoreConfig};
use std::{num::NonZeroUsize, time::Duration};
let store = StateStore::connect(StateStoreConfig::lance(LanceStateStoreConfig::memory())).await?;
let definition = WorkflowDefinition::parse(r#"{
    "schema":"app-kit.workflow/v2","id":"counter","revision":"r1",
    "state":{"count":0},"required":["count"],"reducers":{"count":"reject_concurrent"},
    "entry":"increment","nodes":[{"id":"increment","step":{"kind":"state",
    "mutations":[{"kind":"increment","target":"count","value":{"kind":"literal","value":1}}]}}],
    "routes":{}
}"#)?;
let mut app = App::open(AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition])).await?;
let outcome = async {
    let workspace = app.create_workspace(WorkspaceConfig {
        name: Some("history example".into()), root: std::env::current_dir()?,
    }).await?;
    let workflow = workspace.workflow("counter".parse()?).await?;
    let mut run = workflow.prepare(WorkflowState::default()).await?;
    assert!(matches!(run.snapshot().await?.status, RunStatus::Paused));
    assert_eq!(run.checkpoint(CheckpointSelector::Sequence(0)).await?.state.values()["count"], 0);
    let mut execution = run.execution(ExecutionOptions::default()).await?;
    let mut committed_actions = 0;
    while let Some(event) = execution.next().await? {
        if let ExecutionEvent::Committed { outcomes, .. } = event {
            committed_actions += outcomes.len();
        }
    }
    let terminal = execution.finish().await?;
    assert!(matches!(terminal.status, RunStatus::Succeeded));
    assert_eq!(terminal.state.values()["count"], 1);
    assert_eq!(committed_actions, 1);

    let first = run.history(None, NonZeroUsize::new(1).unwrap()).await?;
    assert_eq!(first.checkpoints[0].sequence, 0);
    let cursor = first.next.expect("the action has a later checkpoint");
    let later = run.history(Some(cursor), NonZeroUsize::new(10).unwrap()).await?;
    assert!(later.checkpoints.iter().any(|checkpoint| checkpoint.action_ordinal == Some(0)));
    let mut copy = workflow.restore(RestoreOptions {
        source: RunSelector::Latest(run.id().clone()), checkpoint: CheckpointSelector::Action(0),
        overrides: WorkflowState::default(), position: ResumePosition::AfterCompleted, fallback: None,
    }).await?;
    assert_ne!(copy.id(), run.id());
    assert!(matches!(copy.snapshot().await?.status, RunStatus::Paused));
    let copied = copy.checkpoint(CheckpointSelector::Sequence(0)).await?;
    assert_eq!(copied.state.values()["count"], 1);
    assert_eq!(&copied.lineage.as_ref().unwrap().source, run.id());
    let copied_terminal = copy.execution(ExecutionOptions::default()).await?.finish().await?;
    assert!(matches!(copied_terminal.status, RunStatus::Succeeded));
    copy.delete(copied_terminal.checkpoint).await?;
    let original_id = run.id().clone();
    run.delete(terminal.checkpoint).await?;
    assert!(matches!(workflow.open_run(RunSelector::Latest(original_id)).await, Err(AppError::NotFound(_))));
    Ok::<(), Box<dyn std::error::Error>>(())
}.await;
let cleanup = app.shutdown(Shutdown::CancelAndDrain { timeout: Duration::from_secs(5) }).await;
outcome?; // An execution error remains primary if shutdown also reports an error.
cleanup?;

A real Agent-backed workflow can stop at an actionable interaction rather than success: inspect crate::workflow::RunSnapshot::interactions and crate::workflow::RunStatus before choosing an explicit resolution. Caller actions own their external effects; a discarded uncommitted workflow state change does not undo external I/O.

§Configure caller behavior

crate::AppConfig’s configuration map links the complete implementations of all five behavior traits and their consuming attachment methods. Actions and conditions use exact declared keys/revisions; typing validates initial and resulting state; hooks observe lifecycle boundaries; child executors drive the already-admitted concrete child futures. None changes the lower Agent or StateStore associated child types. See crate::App::open for durable reopen and the cold migration boundary.

Structs§

ActionContext
Passive application context; it contains no runtime or lower backend handle.
ActionDeclaration
Caller-authored metadata and transient input rules for an action binding.
ActionOutcome
The result of one successfully committed action, distinct from user state.
ActionRequest
One action attempt’s immutable state, resolved transient inputs and identity.
Checkpoint
A committed workflow boundary and its original run identity.
ChildBatch
A non-Clone owner of concrete, pinned, admitted child futures.
ChildCompletion
Exact terminal child evidence produced only by an admitted child future.
ConditionDeclaration
Caller-authored binding identity and declared reads for a predicate.
Execution
Exclusive lazy advancement whose caller source stays on the polling task.
ExecutionOptions
Initial state, transient inputs and boundaries selected for one execution.
GraphComposition
Definitions and explicit connections for one ordinary compiler invocation.
GraphConnection
A default transition appended to an assembled graph.
GraphDescription
Passive graph metadata in authored node and transition order.
GraphEdge
One ordered transition; a missing destination denotes path exhaustion.
GraphImport
An owned graph piece and its graph-local node-name prefix.
GraphNode
One passive node declaration in a graph projection.
HistoryCursor
An opaque continuation of a captured run-history view.
HistoryPage
One bounded page of immutable checkpoints in ascending sequence order.
InlineExecutor
Default executor that applies the batch’s actual execution-mode policy.
JsonState
Raw JSON typing with no additional conversion or validation policy.
NoHooks
Default lifecycle implementation with no observation effects.
RestoreOptions
Selects and copies clean retained state into a new, parked run.
RunId
A validated root UUID or stable child SHA-256 identity.
RunLineage
The source boundary from which a fresh run was restored.
RunSnapshot
The current caller-visible run truth.
StateSerialization
Process-global field encoders, field decoders and tagged JSON decoders.
SyncExecution
Blocking advancement of the same caller-owned execution source.
Workflow
A workflow selected in one application workspace.
WorkflowDefinition
One opaque, parsed app-kit.workflow/v2 definition.
WorkflowId
A validated caller-authored workflow identity.
WorkflowRun
A durable workflow-run handle.
WorkflowState
Immutable, ordered workflow values validated at each operation boundary.

Enums§

ActionEvent
Output from the single lazy source owned by an action attempt.
ActionMode
Execution modes supported by a caller action.
CheckpointKind
The boundary represented by one retained checkpoint.
CheckpointSelector
One scoped retained-history selection, independent of storage revisions.
ExecutionEvent
One transient result or committed execution boundary.
GraphFormat
A deterministic read-only graph export format.
Interaction
A caller-actionable workflow interaction.
LifecycleEvent
An exhaustive borrowed view of one application lifecycle event.
NoActions
Uninhabited default action type for an empty App action inventory.
NoConditions
Uninhabited default predicate type for an empty App condition inventory.
Resolution
Settles one current interaction or unknown effect.
ResumePosition
The saved execution boundary used when restoring a fresh run.
RunSelector
Selects an exact occurrence or the latest native observation of a logical run.
RunStatus
The current stable run status.
StateChange
A resolved immutable state operation; expressions remain private syntax.
WorkflowFailure
A stable terminal workflow-failure category.

Traits§

ActionBackend
A caller-owned action whose lazy source produces results and explicit writes.
ChildExecutor
Caller-selected execution of a complete, already-admitted child batch.
ConditionBackend
A caller-owned predicate evaluated against the full immutable workflow state.
LifecycleHooks
Caller-owned lifecycle observation without a transport or exporter owner.
StateTyping
Caller-owned typed construction and state validation.