actions_and_inputs/
actions_and_inputs.rs

1//! A local caller action with reusable inputs, a partial and explicit state writes.
2//! Run with `cargo run -p app-kit --example actions_and_inputs`.
3
4mod invoice;
5
6use invoice::InvoiceAction;
7use std::error::Error;
8use std::path::PathBuf;
9use std::time::Duration;
10
11use agents_kit::Agent;
12use app_kit::workflow::{
13    ExecutionEvent, ExecutionOptions, RunStatus, WorkflowDefinition, WorkflowState,
14};
15use app_kit::{App, AppConfig, AppError, Shutdown, WorkspaceConfig};
16use serde_json::{Map, Value, json};
17use storage_kit::StateStore;
18use storage_kit::state::config::{LanceStateStoreConfig, StateStoreConfig};
19
20struct InvoiceExample;
21
22impl InvoiceExample {
23    async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
24        let definition = WorkflowDefinition::parse(
25            &json!({
26                "schema":"app-kit.workflow/v2", "id":"invoice", "revision":"r1",
27                "state":{"unit_price":7,"total":0}, "required":["unit_price","total"],
28                "reducers":{"unit_price":"reject_concurrent","total":"reject_concurrent"},
29                "entry":"calculate",
30                "nodes":[{"id":"calculate","step":{
31                    "kind":"action","action":"invoice","revision":"r1","project_result":false
32                }}], "routes":{}
33            })
34            .to_string(),
35        )?;
36        let store =
37            StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
38        // Native caller actions need no Agent value or provider service.
39        let mut app = App::open(
40            AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition])
41                .with_actions(vec![InvoiceAction::new()]),
42        )
43        .await?;
44        let observed = tokio::time::timeout(Duration::from_secs(15), async {
45            let workspace = app
46                .create_workspace(WorkspaceConfig {
47                    name: Some("invoice example".into()),
48                    root: PathBuf::from(env!("CARGO_MANIFEST_DIR")),
49                })
50                .await?;
51            let workflow = workspace.workflow("invoice".parse()?).await?;
52            let mut run = workflow.prepare(WorkflowState::default()).await?;
53            let mut execution = run
54                .execution(ExecutionOptions {
55                    inputs: Map::from_iter([
56                        ("quantity".into(), json!(6)),
57                        ("trace".into(), json!("sample")),
58                    ]),
59                    ..ExecutionOptions::default()
60                })
61                .await?;
62            let Some(ExecutionEvent::Partial { result, .. }) = execution.next().await? else {
63                return Err(AppError::Integration(
64                    "expected the real action partial".into(),
65                ));
66            };
67            if result.get("progress") != Some(&json!("calculated")) {
68                return Err(AppError::Integration("partial changed".into()));
69            }
70            let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
71                return Err(AppError::Integration(
72                    "expected one committed boundary".into(),
73                ));
74            };
75            let expected =
76                json!({"quantity":6,"label":"order","currency":"USD","trace":"sample","total":42});
77            if outcomes.len() != 1 || Value::Object(outcomes[0].result.clone()) != expected {
78                return Err(AppError::Integration(
79                    "resolved inputs or action result changed".into(),
80                ));
81            }
82            let terminal = execution.finish().await?;
83            if !matches!(terminal.status, RunStatus::Succeeded)
84                || terminal.state.values()
85                    != json!({"unit_price":7,"total":42}).as_object().unwrap()
86            {
87                return Err(AppError::Integration(
88                    "only the explicit total write should persist".into(),
89                ));
90            }
91            println!(
92                "Partial: calculated; committed total: 42; transient inputs stayed out of state"
93            );
94            Ok::<(), AppError>(())
95        })
96        .await;
97        let closed = tokio::time::timeout(
98            Duration::from_secs(15),
99            app.shutdown(Shutdown::CancelAndDrain {
100                timeout: Duration::from_secs(10),
101            }),
102        )
103        .await;
104        drop(app);
105        observed.map_err(|_| AppError::Timeout("invoice execution timed out".into()))??;
106        closed.map_err(|_| AppError::Timeout("invoice cleanup timed out".into()))??;
107        Ok(())
108    }
109}
110
111#[tokio::main(flavor = "current_thread")]
112async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
113    InvoiceExample::run().await
114}
actions_and_inputs/actions_and_inputs.rs

actions_and_inputs/
actions_and_inputs.rs
