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