Skip to main content

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}