app_kit/workflow/action/mod.rs
1use futures_util::Stream;
2use futures_util::stream::Empty;
3use serde_json::{Map, Value};
4
5use crate::workflow::definition::{JsonBoundary, MAX_COLLECTION_ITEMS, MAX_TEXT_BYTES, SchemaText};
6use crate::workflow::execution::MAX_RUNTIME_BYTES;
7use crate::workflow::{RunId, StateChange, WorkflowId, WorkflowState};
8use crate::{AppError, WorkspaceId};
9
10/// A caller-owned action whose lazy source produces results and explicit writes.
11///
12/// The declaration remains stable for its logical key and revision. App retains
13/// the exact implementation without requiring `Clone` or normalizing it through
14/// an erased handle. The returned source begins external work only when polled.
15///
16/// An explicit source error is an action failure, including
17/// [`AppError::Integration`]. A captured construction or polling panic is an
18/// execution failure; an App activation deadline is a timeout failure. EOF
19/// without a completed result succeeds with no output. Before storage commit
20/// admission, these outcomes, owning drop and abort discard staged workflow
21/// state and preserve the prior action position. None automatically retries.
22///
23/// The caller owns the source's internal I/O, cancellation, independently spawned
24/// tasks and external reconciliation. Workflow-state rollback does not undo
25/// those effects; a later explicit execution can repeat them. App does not infer
26/// an Unknown effect merely from polling the source. Separately recorded
27/// App-owned lower-operation or admitted-storage uncertainty retains its own
28/// reconciliation authority after source observation ends.
29///
30/// # Example
31///
32/// ```
33/// use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
34/// ActionMode, ActionRequest, StateChange}};
35/// use serde_json::{Map, Value};
36/// struct Increment { declaration: ActionDeclaration }
37/// impl ActionBackend for Increment {
38/// type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
39/// fn declaration(&self) -> &ActionDeclaration { &self.declaration }
40/// fn start(&self, request: ActionRequest) -> Self::Stream {
41/// let count = request.state.values()["count"].as_i64().unwrap() + 1;
42/// futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
43/// result: Map::from_iter([("new_count".into(), Value::from(count))]),
44/// writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
45/// })])
46/// }
47/// }
48/// let action = Increment { declaration: ActionDeclaration {
49/// key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
50/// writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
51/// bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
52/// source: Some("increment count by one".into()), mode: ActionMode::Both,
53/// } };
54/// use app_kit::workflow::{ActionContext, WorkflowState};
55/// let request = ActionRequest {
56/// state: WorkflowState::new(Map::from_iter([("count".into(), Value::from(2))]))?,
57/// inputs: Map::new(), context: ActionContext {
58/// workspace: "appws_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
59/// workflow: "counter".parse()?,
60/// run: "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
61/// occurrence: "00000000000040008000000000000001".into(),
62/// checkpoint: 0, action_ordinal: 0, node: "increment".into(), attempt: 0,
63/// mode: ActionMode::Asynchronous, parent: None, parent_occurrence: None,
64/// },
65/// };
66/// use futures_util::StreamExt;
67/// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
68/// # runtime.block_on(async {
69/// let mut source = action.start(request);
70/// let ActionEvent::Completed { result, writes } = source.next().await.unwrap()? else {
71/// panic!("increment emits a completed result");
72/// };
73/// assert_eq!(result["new_count"], 3);
74/// assert_eq!(writes.len(), 1);
75/// # Ok::<(), app_kit::AppError>(())
76/// # })?;
77/// # Ok::<(), Box<dyn std::error::Error>>(())
78/// ```
79pub trait ActionBackend: Send + Sync + 'static {
80 /// The exact caller source, which may be local to one task and need not be Send.
81 ///
82 /// App stops polling at the first Completed event or error. A source that
83 /// ends without either leaves the attempted action uncommitted.
84 type Stream: Stream<Item = Result<ActionEvent, AppError>> + 'static;
85
86 /// Returns the stable declaration validated when App binds the inventory.
87 ///
88 /// # Example
89 ///
90 /// ```
91 /// use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
92 /// ActionMode, ActionRequest, StateChange}};
93 /// use serde_json::{Map, Value};
94 /// struct Increment { declaration: ActionDeclaration }
95 /// impl ActionBackend for Increment {
96 /// type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
97 /// fn declaration(&self) -> &ActionDeclaration { &self.declaration }
98 /// fn start(&self, request: ActionRequest) -> Self::Stream {
99 /// let count = request.state.values()["count"].as_i64().unwrap() + 1;
100 /// futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
101 /// result: Map::from_iter([("new_count".into(), Value::from(count))]),
102 /// writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
103 /// })])
104 /// }
105 /// }
106 /// let action = Increment { declaration: ActionDeclaration {
107 /// key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
108 /// writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
109 /// bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
110 /// source: Some("increment count by one".into()), mode: ActionMode::Both,
111 /// } };
112 /// assert_eq!(action.declaration().reads, ["count"]);
113 /// assert_eq!(action.declaration().revision, "r1");
114 /// ```
115 fn declaration(&self) -> &ActionDeclaration;
116
117 /// Constructs the source for one attempt without performing external work.
118 ///
119 /// The source may be local and non-Send. Construct it without starting external
120 /// I/O: the first poll accepts that work. Emit partial results for observation and
121 /// one Completed result with explicit state writes; EOF before Completed commits
122 /// no action. The action owns cancellation and reconciliation of its own I/O and
123 /// spawned tasks. State rollback cannot undo caller external effects.
124 ///
125 /// # Example
126 ///
127 /// ```
128 /// use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
129 /// ActionMode, ActionRequest, StateChange}};
130 /// use serde_json::{Map, Value};
131 /// struct Increment { declaration: ActionDeclaration }
132 /// impl ActionBackend for Increment {
133 /// type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
134 /// fn declaration(&self) -> &ActionDeclaration { &self.declaration }
135 /// fn start(&self, request: ActionRequest) -> Self::Stream {
136 /// let count = request.state.values()["count"].as_i64().unwrap() + 1;
137 /// futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
138 /// result: Map::from_iter([("new_count".into(), Value::from(count))]),
139 /// writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
140 /// })])
141 /// }
142 /// }
143 /// let action = Increment { declaration: ActionDeclaration {
144 /// key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
145 /// writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
146 /// bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
147 /// source: Some("increment count by one".into()), mode: ActionMode::Both,
148 /// } };
149 /// use app_kit::workflow::{ActionContext, WorkflowState};
150 /// let request = ActionRequest {
151 /// state: WorkflowState::new(Map::from_iter([("count".into(), Value::from(2))]))?,
152 /// inputs: Map::new(), context: ActionContext {
153 /// workspace: "appws_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
154 /// workflow: "counter".parse()?,
155 /// run: "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
156 /// occurrence: "00000000000040008000000000000001".into(),
157 /// checkpoint: 0, action_ordinal: 0, node: "increment".into(), attempt: 0,
158 /// mode: ActionMode::Asynchronous, parent: None, parent_occurrence: None,
159 /// },
160 /// };
161 /// use futures_util::StreamExt;
162 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
163 /// # runtime.block_on(async {
164 /// let mut source = action.start(request);
165 /// let ActionEvent::Completed { result, writes } = source.next().await.unwrap()? else {
166 /// panic!("increment emits a completed result");
167 /// };
168 /// assert_eq!(result["new_count"], 3);
169 /// assert_eq!(writes.len(), 1);
170 /// # Ok::<(), app_kit::AppError>(())
171 /// # })?;
172 /// # Ok::<(), Box<dyn std::error::Error>>(())
173 /// ```
174 fn start(&self, request: ActionRequest) -> Self::Stream;
175}
176
177/// Execution modes supported by a caller action.
178#[derive(Clone, Copy, Debug, PartialEq, Eq)]
179pub enum ActionMode {
180 /// Supports execution driven by the synchronous session.
181 Synchronous,
182 /// Supports execution driven by an asynchronous session.
183 Asynchronous,
184 /// Supports either mode; an invocation context selects one concrete mode.
185 Both,
186}
187
188/// Caller-authored metadata and transient input rules for an action binding.
189///
190/// Read and write names describe state keys; they do not hide other values or
191/// restrict explicit writes. Input names are data, except the reserved `__*`
192/// namespace. Repeated declarations are accepted. Defaults apply only to
193/// optional, non-required inputs; bound values may satisfy required inputs and
194/// cannot be overridden by runtime input, even with an equal value.
195///
196/// # Example
197///
198/// ```
199/// use app_kit::workflow::{ActionDeclaration, ActionMode};
200/// use serde_json::Map;
201/// let declaration = ActionDeclaration {
202/// key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
203/// writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
204/// bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
205/// source: Some("increment count by one".into()), mode: ActionMode::Both,
206/// };
207/// assert!(declaration.required_inputs.is_empty());
208/// assert_eq!(declaration.writes, ["count"]);
209/// ```
210pub struct ActionDeclaration {
211 /// Canonical logical binding key, using the workflow identity grammar.
212 pub key: String,
213 /// Nonempty caller revision identifying this stable behavior.
214 pub revision: String,
215 /// Declared state reads, retaining caller order.
216 pub reads: Vec<String>,
217 /// Declared state writes, retaining caller order.
218 pub writes: Vec<String>,
219 /// Input names that must be present after defaults, bindings and runtime input.
220 pub required_inputs: Vec<String>,
221 /// Input names eligible for defaults.
222 pub optional_inputs: Vec<String>,
223 /// Fixed input values; undeclared input names are permitted.
224 pub bound_inputs: Map<String, Value>,
225 /// Defaults for optional inputs, merged before fixed bindings.
226 pub default_inputs: Map<String, Value>,
227 /// Caller tags preserved for graph inspection and selectors.
228 pub tags: Vec<String>,
229 /// Optional caller source text for inspection.
230 pub source: Option<String>,
231 /// Invocation modes this action supports.
232 pub mode: ActionMode,
233}
234
235/// One action attempt's immutable state, resolved transient inputs and identity.
236///
237/// # Example
238///
239/// ```
240/// use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
241/// ActionMode, ActionRequest, StateChange}};
242/// use serde_json::{Map, Value};
243/// struct Increment { declaration: ActionDeclaration }
244/// impl ActionBackend for Increment {
245/// type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
246/// fn declaration(&self) -> &ActionDeclaration { &self.declaration }
247/// fn start(&self, request: ActionRequest) -> Self::Stream {
248/// let count = request.state.values()["count"].as_i64().unwrap() + 1;
249/// futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
250/// result: Map::from_iter([("new_count".into(), Value::from(count))]),
251/// writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
252/// })])
253/// }
254/// }
255/// let action = Increment { declaration: ActionDeclaration {
256/// key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
257/// writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
258/// bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
259/// source: Some("increment count by one".into()), mode: ActionMode::Both,
260/// } };
261/// use app_kit::workflow::{ActionContext, WorkflowState};
262/// let request = ActionRequest {
263/// state: WorkflowState::new(Map::from_iter([("count".into(), Value::from(2))]))?,
264/// inputs: Map::new(), context: ActionContext {
265/// workspace: "appws_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
266/// workflow: "counter".parse()?,
267/// run: "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
268/// occurrence: "00000000000040008000000000000001".into(),
269/// checkpoint: 0, action_ordinal: 0, node: "increment".into(), attempt: 0,
270/// mode: ActionMode::Asynchronous, parent: None, parent_occurrence: None,
271/// },
272/// };
273/// use futures_util::StreamExt;
274/// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
275/// # runtime.block_on(async {
276/// let mut source = action.start(request);
277/// let ActionEvent::Completed { result, writes } = source.next().await.unwrap()? else {
278/// panic!("increment emits a completed result");
279/// };
280/// assert_eq!(result["new_count"], 3);
281/// assert_eq!(writes.len(), 1);
282/// # Ok::<(), app_kit::AppError>(())
283/// # })?;
284/// # Ok::<(), Box<dyn std::error::Error>>(())
285/// ```
286pub struct ActionRequest {
287 /// The full immutable base state, independent of transient inputs and results.
288 pub state: WorkflowState,
289 /// Resolved defaults, fixed bindings and all caller runtime inputs.
290 pub inputs: Map<String, Value>,
291 /// Application identity and execution position for this attempt.
292 pub context: ActionContext,
293}
294
295/// Passive application context; it contains no runtime or lower backend handle.
296///
297/// # Example
298///
299/// ```
300/// use app_kit::workflow::{ActionContext, ActionMode};
301/// let context = ActionContext {
302/// workspace: "appws_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
303/// workflow: "counter".parse()?,
304/// run: "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
305/// occurrence: "00000000000040008000000000000001".into(),
306/// checkpoint: 0, action_ordinal: 0, node: "increment".into(), attempt: 0,
307/// mode: ActionMode::Asynchronous, parent: None, parent_occurrence: None,
308/// };
309/// assert_eq!(context.action_ordinal, 0);
310/// assert!(context.parent.is_none());
311/// # Ok::<(), app_kit::AppError>(())
312/// ```
313#[derive(Clone)]
314pub struct ActionContext {
315 /// Workspace containing the workflow run.
316 pub workspace: WorkspaceId,
317 /// Exact logical workflow identity.
318 pub workflow: WorkflowId,
319 /// Run receiving this action's staged result.
320 pub run: RunId,
321 /// Exact admitted execution occurrence.
322 pub occurrence: String,
323 /// Committed base checkpoint sequence.
324 pub checkpoint: u64,
325 /// Attempted action ordinal; failed attempts do not advance committed state.
326 pub action_ordinal: u64,
327 /// Node being invoked.
328 pub node: String,
329 /// Attempt identity within this activation.
330 pub attempt: u64,
331 /// Actual invocation mode, always Synchronous or Asynchronous, never Both.
332 pub mode: ActionMode,
333 /// Parent run when invoked as a child workflow.
334 pub parent: Option<RunId>,
335 /// Exact occurrence of the parent, when this action belongs to a child.
336 pub parent_occurrence: Option<String>,
337}
338
339/// Output from the single lazy source owned by an action attempt.
340pub enum ActionEvent {
341 /// Transient partial output that does not commit workflow state.
342 Partial {
343 /// Caller result fields for this partial observation.
344 result: Map<String, Value>,
345 },
346 /// The first authoritative terminal result and its staged state operations.
347 Completed {
348 /// Transient final result, distinct from durable workflow state.
349 result: Map<String, Value>,
350 /// Ordered state operations to validate and reduce at the commit barrier.
351 writes: Vec<StateChange>,
352 },
353 /// Begins an explicitly addressed tracing span within this action attempt.
354 SpanStarted {
355 /// Caller-local span identity.
356 id: String,
357 /// Display name of the span.
358 name: String,
359 /// Optional parent span identity within the attempt.
360 parent: Option<String>,
361 /// Preceding spans on which this span depends.
362 dependencies: Vec<String>,
363 },
364 /// Attaches an attribute to an addressed span or the action's root span.
365 SpanAttribute {
366 /// Target span, or None for the action's root span.
367 span: Option<String>,
368 /// Attribute name.
369 key: String,
370 /// Attribute value.
371 value: Value,
372 /// Caller tags attached to this attribute.
373 tags: Map<String, Value>,
374 },
375 /// Finishes an explicitly addressed span.
376 SpanFinished {
377 /// The previously started span identity.
378 id: String,
379 },
380}
381
382impl ActionEvent {
383 pub(crate) fn validate_span(&self) -> Result<(), AppError> {
384 let bytes = match self {
385 Self::SpanStarted {
386 id,
387 name,
388 parent,
389 dependencies,
390 } => {
391 if dependencies.len() > MAX_COLLECTION_ITEMS
392 || [id.as_str(), name.as_str()]
393 .into_iter()
394 .chain(parent.as_deref())
395 .chain(dependencies.iter().map(String::as_str))
396 .any(|text| text.len() > MAX_TEXT_BYTES)
397 {
398 return Err(AppError::InvalidInput(
399 "caller span metadata exceeds its boundary".into(),
400 ));
401 }
402 dependencies.iter().fold(
403 id.len()
404 .saturating_add(name.len())
405 .saturating_add(parent.as_ref().map_or(0, String::len)),
406 |bytes, dependency| bytes.saturating_add(dependency.len()),
407 )
408 }
409 Self::SpanAttribute {
410 span,
411 key,
412 value,
413 tags,
414 } => {
415 if key.len() > MAX_TEXT_BYTES
416 || span
417 .as_ref()
418 .is_some_and(|span| span.len() > MAX_TEXT_BYTES)
419 || tags.len() > MAX_COLLECTION_ITEMS
420 || tags.values().any(|value| {
421 value
422 .as_str()
423 .is_none_or(|value| value.len() > MAX_TEXT_BYTES)
424 })
425 {
426 return Err(AppError::InvalidInput(
427 "caller span attribute exceeds its boundary".into(),
428 ));
429 }
430 JsonBoundary::validate(value)?;
431 JsonBoundary::validate_map(tags)?;
432 let value_bytes = serde_json::to_vec(value)
433 .map_err(|_| AppError::InvalidInput("caller span attribute is invalid".into()))?
434 .len();
435 let tag_bytes = serde_json::to_vec(tags)
436 .map_err(|_| AppError::InvalidInput("caller span tags are invalid".into()))?
437 .len();
438 key.len()
439 .saturating_add(span.as_ref().map_or(0, String::len))
440 .saturating_add(value_bytes)
441 .saturating_add(tag_bytes)
442 }
443 Self::SpanFinished { id } => {
444 if id.len() > MAX_TEXT_BYTES {
445 return Err(AppError::InvalidInput(
446 "caller span identity exceeds its boundary".into(),
447 ));
448 }
449 id.len()
450 }
451 Self::Partial { .. } | Self::Completed { .. } => {
452 return Err(AppError::Integration(
453 "action event is not span metadata".into(),
454 ));
455 }
456 };
457 if bytes > MAX_RUNTIME_BYTES {
458 return Err(AppError::InvalidInput(
459 "caller span event exceeds its byte boundary".into(),
460 ));
461 }
462 Ok(())
463 }
464}
465
466/// A caller-owned predicate evaluated against the full immutable workflow state.
467///
468/// App validates declared reads before evaluating the predicate, including
469/// reads within composed conditions that would otherwise short-circuit.
470///
471/// # Example
472///
473/// ```
474/// use app_kit::{AppError, workflow::{ConditionBackend, ConditionDeclaration, WorkflowState}};
475/// struct Ready { declaration: ConditionDeclaration }
476/// impl ConditionBackend for Ready {
477/// fn declaration(&self) -> &ConditionDeclaration { &self.declaration }
478/// fn evaluate(&self, state: &WorkflowState) -> Result<bool, AppError> {
479/// let count = state.values().get("count").and_then(serde_json::Value::as_i64)
480/// .ok_or_else(|| AppError::InvalidInput("count must be an integer".into()))?;
481/// Ok(count >= 3)
482/// }
483/// }
484/// let condition = Ready { declaration: ConditionDeclaration {
485/// key: "ready".into(), revision: "r1".into(), reads: vec!["count".into()],
486/// } };
487/// let state = WorkflowState::new(serde_json::from_str(r#"{"count":3}"#)?)?;
488/// assert!(condition.evaluate(&state)?);
489/// # Ok::<(), Box<dyn std::error::Error>>(())
490/// ```
491pub trait ConditionBackend: Send + Sync + 'static {
492 /// Returns the stable declaration for one logical predicate revision.
493 ///
494 /// # Example
495 ///
496 /// ```
497 /// use app_kit::{AppError, workflow::{ConditionBackend, ConditionDeclaration, WorkflowState}};
498 /// struct Ready { declaration: ConditionDeclaration }
499 /// impl ConditionBackend for Ready {
500 /// fn declaration(&self) -> &ConditionDeclaration { &self.declaration }
501 /// fn evaluate(&self, state: &WorkflowState) -> Result<bool, AppError> {
502 /// let count = state.values().get("count").and_then(serde_json::Value::as_i64)
503 /// .ok_or_else(|| AppError::InvalidInput("count must be an integer".into()))?;
504 /// Ok(count >= 3)
505 /// }
506 /// }
507 /// let condition = Ready { declaration: ConditionDeclaration {
508 /// key: "ready".into(), revision: "r1".into(), reads: vec!["count".into()],
509 /// } };
510 /// assert_eq!(condition.declaration().key, "ready");
511 /// assert_eq!(condition.declaration().reads, ["count"]);
512 /// ```
513 fn declaration(&self) -> &ConditionDeclaration;
514
515 /// Evaluates the predicate without advancing workflow state or action ordinals.
516 ///
517 /// # Errors
518 ///
519 /// Return an error when evaluation cannot produce a valid boolean. App validates
520 /// declared reads before calling this method; caller failures stop that routing
521 /// decision rather than being interpreted as false.
522 ///
523 /// # Example
524 ///
525 /// ```
526 /// use app_kit::{AppError, workflow::{ConditionBackend, ConditionDeclaration, WorkflowState}};
527 /// struct Ready { declaration: ConditionDeclaration }
528 /// impl ConditionBackend for Ready {
529 /// fn declaration(&self) -> &ConditionDeclaration { &self.declaration }
530 /// fn evaluate(&self, state: &WorkflowState) -> Result<bool, AppError> {
531 /// let count = state.values().get("count").and_then(serde_json::Value::as_i64)
532 /// .ok_or_else(|| AppError::InvalidInput("count must be an integer".into()))?;
533 /// Ok(count >= 3)
534 /// }
535 /// }
536 /// let condition = Ready { declaration: ConditionDeclaration {
537 /// key: "ready".into(), revision: "r1".into(), reads: vec!["count".into()],
538 /// } };
539 /// let state = WorkflowState::new(serde_json::from_str(r#"{"count":3}"#)?)?;
540 /// assert!(condition.evaluate(&state)?);
541 /// # Ok::<(), Box<dyn std::error::Error>>(())
542 /// ```
543 fn evaluate(&self, state: &WorkflowState) -> Result<bool, AppError>;
544}
545
546/// Caller-authored binding identity and declared reads for a predicate.
547///
548/// # Example
549///
550/// ```
551/// use app_kit::workflow::ConditionDeclaration;
552/// let declaration = ConditionDeclaration {
553/// key: "ready".into(), revision: "r1".into(), reads: vec!["count".into()],
554/// };
555/// assert_eq!(declaration.reads, ["count"]);
556/// ```
557pub struct ConditionDeclaration {
558 /// Canonical logical binding key, using the workflow identity grammar.
559 pub key: String,
560 /// Nonempty caller revision identifying this stable behavior.
561 pub revision: String,
562 /// Top-level state keys required before predicate evaluation.
563 pub reads: Vec<String>,
564}
565
566/// Uninhabited default action type for an empty App action inventory.
567///
568/// An empty inventory needs no placeholder action or declaration:
569/// ```
570/// use app_kit::workflow::NoActions;
571/// let actions: Vec<NoActions> = Vec::new();
572/// assert!(actions.is_empty());
573/// ```
574pub enum NoActions {}
575
576/// Uninhabited default predicate type for an empty App condition inventory.
577///
578/// ```
579/// use app_kit::workflow::NoConditions;
580/// let conditions: Vec<NoConditions> = Vec::new();
581/// assert!(conditions.is_empty());
582/// ```
583pub enum NoConditions {}
584
585impl ActionBackend for NoActions {
586 type Stream = Empty<Result<ActionEvent, AppError>>;
587
588 fn declaration(&self) -> &ActionDeclaration {
589 match *self {}
590 }
591
592 fn start(&self, _request: ActionRequest) -> Self::Stream {
593 match *self {}
594 }
595}
596
597impl ConditionBackend for NoConditions {
598 fn declaration(&self) -> &ConditionDeclaration {
599 match *self {}
600 }
601
602 fn evaluate(&self, _state: &WorkflowState) -> Result<bool, AppError> {
603 match *self {}
604 }
605}
606
607pub(crate) enum ActionInputs {
608 Ready(Map<String, Value>),
609 Required(Vec<String>),
610}
611
612impl ActionInputs {
613 pub(crate) fn into_values(self) -> Result<Map<String, Value>, AppError> {
614 match self {
615 Self::Ready(values) => Ok(values),
616 Self::Required(_) => Err(AppError::InvalidInput(
617 "required action input is missing".into(),
618 )),
619 }
620 }
621}
622
623impl ActionDeclaration {
624 pub(crate) fn result_projection(&self, result: &Map<String, Value>) -> StateChange {
625 let mut values = Map::new();
626 for key in &self.writes {
627 if let Some(value) = result.get(key) {
628 values.insert(key.clone(), value.clone());
629 }
630 }
631 StateChange::Assign { values }
632 }
633
634 pub(crate) fn resolve_inputs(
635 &self,
636 runtime: &Map<String, Value>,
637 mode: ActionMode,
638 ) -> Result<ActionInputs, AppError> {
639 if !matches!(
640 (mode, self.mode),
641 (
642 ActionMode::Synchronous,
643 ActionMode::Synchronous | ActionMode::Both
644 ) | (
645 ActionMode::Asynchronous,
646 ActionMode::Asynchronous | ActionMode::Both
647 )
648 ) {
649 return Err(AppError::InvalidInput(
650 "action does not support the selected execution mode".into(),
651 ));
652 }
653 JsonBoundary::validate_inputs(runtime)?;
654 let mut resolved = self.default_inputs.clone();
655 for (key, value) in &self.bound_inputs {
656 resolved.insert(key.clone(), value.clone());
657 }
658 for (key, value) in runtime {
659 if self.bound_inputs.contains_key(key) {
660 return Err(AppError::InvalidInput(
661 "runtime input cannot override a bound action input".into(),
662 ));
663 }
664 resolved.insert(key.clone(), value.clone());
665 }
666 JsonBoundary::validate_map(&resolved)?;
667 let mut required = self
668 .required_inputs
669 .iter()
670 .filter(|key| !resolved.contains_key(*key))
671 .cloned()
672 .collect::<Vec<_>>();
673 required.sort_unstable();
674 required.dedup();
675 if required.is_empty() {
676 Ok(ActionInputs::Ready(resolved))
677 } else {
678 Ok(ActionInputs::Required(required))
679 }
680 }
681
682 pub(crate) fn validate_reads(&self, state: &WorkflowState) -> Result<(), AppError> {
683 if self
684 .reads
685 .iter()
686 .any(|key| !state.values().contains_key(key))
687 {
688 return Err(AppError::InvalidInput(
689 "declared action state read is missing".into(),
690 ));
691 }
692 Ok(())
693 }
694
695 pub(crate) fn validate(&self) -> Result<(), AppError> {
696 SchemaText::node_or_state("action binding key", &self.key)?;
697 SchemaText::revision(&self.revision)?;
698 if [
699 self.reads.len(),
700 self.writes.len(),
701 self.required_inputs.len(),
702 self.optional_inputs.len(),
703 self.bound_inputs.len(),
704 self.default_inputs.len(),
705 self.tags.len(),
706 ]
707 .into_iter()
708 .any(|length| length > MAX_COLLECTION_ITEMS)
709 {
710 return Err(AppError::InvalidInput(
711 "action declaration exceeds the private collection bound".into(),
712 ));
713 }
714 for key in self.reads.iter().chain(&self.writes) {
715 JsonBoundary::validate_key(key)?;
716 }
717 for key in self
718 .required_inputs
719 .iter()
720 .chain(&self.optional_inputs)
721 .chain(self.bound_inputs.keys())
722 .chain(self.default_inputs.keys())
723 {
724 JsonBoundary::validate_key(key)?;
725 if key.starts_with("__") {
726 return Err(AppError::InvalidInput(
727 "action input uses the reserved runtime namespace".into(),
728 ));
729 }
730 }
731 if self
732 .default_inputs
733 .keys()
734 .any(|key| self.required_inputs.contains(key) || !self.optional_inputs.contains(key))
735 {
736 return Err(AppError::InvalidInput(
737 "action defaults require optional, non-required inputs".into(),
738 ));
739 }
740 if self.tags.iter().any(|tag| tag.len() > MAX_TEXT_BYTES)
741 || self
742 .source
743 .as_ref()
744 .is_some_and(|source| source.len() > MAX_TEXT_BYTES)
745 {
746 return Err(AppError::InvalidInput(
747 "action declaration text exceeds the private byte bound".into(),
748 ));
749 }
750 JsonBoundary::validate_map(&self.bound_inputs)?;
751 JsonBoundary::validate_map(&self.default_inputs)?;
752 JsonBoundary::validate_size(&(
753 &self.key,
754 &self.revision,
755 &self.reads,
756 &self.writes,
757 &self.required_inputs,
758 &self.optional_inputs,
759 &self.bound_inputs,
760 &self.default_inputs,
761 &self.tags,
762 &self.source,
763 ))
764 }
765}
766
767impl ConditionDeclaration {
768 pub(crate) fn validate(&self) -> Result<(), AppError> {
769 SchemaText::node_or_state("condition binding key", &self.key)?;
770 SchemaText::revision(&self.revision)?;
771 if self.reads.len() > MAX_COLLECTION_ITEMS {
772 return Err(AppError::InvalidInput(
773 "condition declaration exceeds the private collection bound".into(),
774 ));
775 }
776 for key in &self.reads {
777 JsonBoundary::validate_key(key)?;
778 }
779 JsonBoundary::validate_size(&(&self.key, &self.revision, &self.reads))
780 }
781}