ledgence_orchestration_api/
workflow.rs

1//! Checkpoint workflow contracts. Workflow state never occupies CloudEvent data.
2//!
3//! A logical activation is an ordinary leased task plus an explicit, immutable
4//! continuation context. Its retries share one local-step journal. A terminal
5//! controller result is interpreted as a decision only for registered activations.
6use crate::*;
7use ledgence_worker_api::{ProgramDescriptor, ProgramRef, validate_wire_value};
8use serde_json::Value;
9use std::collections::{BTreeMap, BTreeSet};
10
11pub const WORKFLOW_RUNTIME_SCHEMA: &str = "ledgence.workflow.activation.v1";
12pub const WORKFLOW_VERSION: u8 = 1;
13pub const WORKFLOW_MAX_COMMANDS: usize = 64;
14pub const WORKFLOW_MAX_LOCAL_STEPS: usize = 128;
15pub const WORKFLOW_CHECKPOINT_MAX_BYTES: usize = 64 * 1024;
16pub const WORKFLOW_DECISION_MAX_BYTES: usize = 256 * 1024;
17pub const WORKFLOW_INPUTS_MAX_BYTES: usize = 256 * 1024;
18pub const WORKFLOW_LOCAL_RECORD_MAX_BYTES: usize = 128 * 1024;
19pub const WORKFLOW_LOCAL_LEDGER_MAX_BYTES: usize = 256 * 1024;
20pub const WORKFLOW_CONTEXT_MAX_BYTES: usize = 640 * 1024;
21pub const WORKFLOW_MAX_WORK_BATCH: u32 = 16;
22
23/// Controller results include a platform decision envelope around application
24/// values. Ordinary task output retains its depth-64 contract; registered
25/// activations allow metadata depth 96 before the coordinator validates each
26/// application value and the exact decision shape. The lifecycle core verifies
27/// the report's activation identity against the acquired task before acceptance.
28pub fn validate_task_output(output: &Value, workflow_activation: bool) -> Result<()> {
29    if workflow_activation {
30        ledgence_worker_api::validate_runtime_payload(output, SETTLEMENT_MAX_BYTES)?;
31    } else {
32        validate_wire_value(output)?;
33    }
34    Ok(())
35}
36
37#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
38#[serde(deny_unknown_fields)]
39pub struct LocalStepRecord {
40    pub key: String,
41    pub callable: String,
42    #[serde(deserialize_with = "crate::observation::required_value")]
43    pub input: Value,
44    #[serde(deserialize_with = "crate::observation::required_value")]
45    pub output: Value,
46}
47impl LocalStepRecord {
48    pub fn validate(&self) -> Result<()> {
49        validate_text(&self.key, 128)?;
50        validate_text(&self.callable, 512)?;
51        validate_wire_value(&self.input)?;
52        validate_wire_value(&self.output)?;
53        bounded(self, WORKFLOW_LOCAL_RECORD_MAX_BYTES, "local step")
54    }
55
56    /// Binding and value are immutable once acknowledged, including numeric
57    /// representation (1 and 1.0 are different). JSON object order is irrelevant.
58    pub fn matches(&self, other: &Self) -> Result<bool> {
59        self.validate()?;
60        other.validate()?;
61        Ok(
62            canonical_json_bytes(&serde_json::to_value(self).map_err(encoding)?)?
63                == canonical_json_bytes(&serde_json::to_value(other).map_err(encoding)?)?,
64        )
65    }
66}
67
68/// Frozen activation inputs and checkpoint, plus the current committed journal.
69/// New child completions do not mutate the frozen input batch or revision.
70#[derive(Debug, Clone, Serialize, Deserialize)]
71#[serde(deny_unknown_fields)]
72pub struct WorkflowActivationContext {
73    /// Present together only for a nested owned workflow; roots omit both.
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub parent_workflow_id: Option<String>,
76    #[serde(default, skip_serializing_if = "Option::is_none")]
77    pub root_workflow_id: Option<String>,
78    pub v: u8,
79    pub workflow_id: String,
80    pub activation_id: String,
81    pub revision: u64,
82    pub continuation: String,
83    #[serde(deserialize_with = "crate::observation::required_value")]
84    pub state: Value,
85    pub inputs: BTreeMap<String, WorkflowChildResult>,
86    pub local_steps: Vec<LocalStepRecord>,
87    /// Frozen result of one external event/timer wait. Absent on older contexts
88    /// and ordinary child continuations; never merged into user CloudEvent data.
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub wake: Option<WorkflowWake>,
91}
92impl WorkflowActivationContext {
93    pub fn validate(&self) -> Result<()> {
94        version(self.v)?;
95        validate_workflow_lineage(
96            Some(&self.workflow_id),
97            self.parent_workflow_id.as_deref(),
98            self.root_workflow_id.as_deref(),
99        )?;
100        validate_text(&self.activation_id, 128)?;
101        validate_text(&self.continuation, 128)?;
102        checkpoint(&self.state)?;
103        if self.inputs.len() > WORKFLOW_MAX_COMMANDS
104            || self.local_steps.len() > WORKFLOW_MAX_LOCAL_STEPS
105        {
106            return Err(invalid("workflow context has too many entries"));
107        }
108        for (key, child) in &self.inputs {
109            validate_text(key, 128)?;
110            child.validate()?;
111        }
112        if let Some(wake) = &self.wake {
113            wake.validate()?;
114            #[derive(Serialize)]
115            struct FrozenInputs<'a> {
116                inputs: &'a BTreeMap<String, WorkflowChildResult>,
117                wake: &'a WorkflowWake,
118            }
119            bounded(
120                &FrozenInputs {
121                    inputs: &self.inputs,
122                    wake,
123                },
124                WORKFLOW_INPUTS_MAX_BYTES,
125                "workflow inputs and wake",
126            )?;
127        } else {
128            bounded(&self.inputs, WORKFLOW_INPUTS_MAX_BYTES, "workflow inputs")?;
129        }
130        let mut keys = BTreeSet::new();
131        for step in &self.local_steps {
132            step.validate()?;
133            if !keys.insert(&step.key) {
134                return Err(invalid("duplicate local step key"));
135            }
136        }
137        bounded(
138            &self.local_steps,
139            WORKFLOW_LOCAL_LEDGER_MAX_BYTES,
140            "local step ledger",
141        )?;
142        bounded(self, WORKFLOW_CONTEXT_MAX_BYTES, "workflow context")
143    }
144}
145
146#[derive(Debug, Clone, Serialize, Deserialize)]
147#[serde(deny_unknown_fields)]
148pub struct WorkflowChildCommand {
149    #[serde(default, skip_serializing_if = "WorkflowChildKind::is_task")]
150    pub kind: WorkflowChildKind,
151    pub key: String,
152    pub program: ProgramRef,
153    pub queue: String,
154    #[serde(deserialize_with = "crate::observation::required_value")]
155    pub data: Value,
156    #[serde(default)]
157    pub retry_policy: RetryPolicy,
158    #[serde(default = "attempt_timeout")]
159    pub attempt_timeout_ms: u64,
160}
161const fn attempt_timeout() -> u64 {
162    300_000
163}
164/// Compatibility name for the original task-only command contract.
165pub type WorkflowTaskCommand = WorkflowChildCommand;
166
167impl WorkflowChildCommand {
168    pub fn submission(&self, scope: &Scope, correlation_key: Option<String>) -> SubmitTask {
169        SubmitTask {
170            tenant_id: scope.tenant_id.clone(),
171            namespace: scope.namespace.clone(),
172            queue: self.queue.clone(),
173            program: self.program.clone(),
174            correlation_key,
175            data: self.data.clone(),
176            retry_policy: self.retry_policy.clone(),
177            attempt_timeout_ms: self.attempt_timeout_ms,
178        }
179    }
180    pub fn validate(&self) -> Result<()> {
181        validate_text(&self.key, 128)?;
182        self.submission(
183            &Scope {
184                tenant_id: "validation".into(),
185                namespace: "validation".into(),
186            },
187            None,
188        )
189        .validate()?;
190        Ok(())
191    }
192}
193
194#[derive(Debug, Clone, Serialize, Deserialize)]
195pub struct WorkflowDecision {
196    pub v: u8,
197    pub activation_id: String,
198    pub revision: u64,
199    #[serde(flatten)]
200    pub action: WorkflowAction,
201}
202
203#[derive(Debug, Clone, Serialize, Deserialize)]
204#[serde(tag = "kind", rename_all = "snake_case")]
205pub enum WorkflowAction {
206    Wait {
207        state: Value,
208        continuation: String,
209        commands: Vec<WorkflowChildCommand>,
210        wait: WorkflowWait,
211    },
212    Suspend {
213        state: Value,
214        continuation: String,
215        commands: Vec<WorkflowChildCommand>,
216        /// Sealed all-terminal membership. Empty membership is immediately ready.
217        until: Vec<String>,
218    },
219    Continue {
220        state: Value,
221        continuation: String,
222        commands: Vec<WorkflowChildCommand>,
223    },
224    Complete {
225        output: Value,
226    },
227    Fail {
228        error: ApplicationError,
229    },
230}
231impl WorkflowDecision {
232    /// Strictly decode a controller outcome; serde flatten alone cannot reject
233    /// unknown fields reliably, so the exact top-level shape is checked first.
234    pub fn decode(value: &Value) -> Result<Self> {
235        // Bound a borrowed value before deserialization clones its payload.
236        bounded(value, WORKFLOW_DECISION_MAX_BYTES, "workflow decision")?;
237        let object = value
238            .as_object()
239            .ok_or_else(|| invalid("workflow decision must be an object"))?;
240        let action_fields: &[&str] = match object.get("kind").and_then(Value::as_str) {
241            Some("wait") => &["state", "continuation", "commands", "wait"],
242            Some("suspend") => &["state", "continuation", "commands", "until"],
243            Some("continue") => &["state", "continuation", "commands"],
244            Some("complete") => &["output"],
245            Some("fail") => &["error"],
246            _ => return Err(invalid("unknown workflow decision kind")),
247        };
248        let common = ["v", "activation_id", "revision", "kind"];
249        if object.len() != common.len() + action_fields.len()
250            || object.keys().any(|key| {
251                !common.contains(&key.as_str()) && !action_fields.contains(&key.as_str())
252            })
253        {
254            return Err(invalid("unknown or missing workflow decision fields"));
255        }
256        let decision: Self = serde_json::from_value(value.clone()).map_err(encoding)?;
257        decision.validate()?;
258        Ok(decision)
259    }
260    pub fn validate(&self) -> Result<()> {
261        version(self.v)?;
262        validate_text(&self.activation_id, 128)?;
263        match &self.action {
264            WorkflowAction::Wait {
265                state,
266                continuation,
267                commands,
268                wait,
269            } => {
270                validate_continuation(state, continuation, commands)?;
271                wait.validate()?;
272            }
273            WorkflowAction::Suspend {
274                state,
275                continuation,
276                commands,
277                until,
278            } => {
279                validate_continuation(state, continuation, commands)?;
280                if until.len() > WORKFLOW_MAX_COMMANDS {
281                    return Err(invalid("wait membership exceeds supported limit"));
282                }
283                let mut keys = BTreeSet::new();
284                for key in until {
285                    validate_text(key, 128)?;
286                    if !keys.insert(key) {
287                        return Err(invalid("duplicate wait member"));
288                    }
289                }
290            }
291            WorkflowAction::Continue {
292                state,
293                continuation,
294                commands,
295            } => validate_continuation(state, continuation, commands)?,
296            WorkflowAction::Complete { output } => validate_wire_value(output)?,
297            WorkflowAction::Fail { error } => validate_error(error)?,
298        }
299        bounded(self, WORKFLOW_DECISION_MAX_BYTES, "workflow decision")
300    }
301    pub fn commands(&self) -> &[WorkflowChildCommand] {
302        match &self.action {
303            WorkflowAction::Wait { commands, .. }
304            | WorkflowAction::Suspend { commands, .. }
305            | WorkflowAction::Continue { commands, .. } => commands,
306            WorkflowAction::Complete { .. } | WorkflowAction::Fail { .. } => &[],
307        }
308    }
309}
310fn validate_continuation(
311    state: &Value,
312    continuation: &str,
313    commands: &[WorkflowChildCommand],
314) -> Result<()> {
315    checkpoint(state)?;
316    validate_text(continuation, 128)?;
317    if commands.len() > WORKFLOW_MAX_COMMANDS {
318        return Err(invalid("workflow command batch exceeds supported limit"));
319    }
320    let mut keys = BTreeSet::new();
321    for command in commands {
322        command.validate()?;
323        if !keys.insert(&command.key) {
324            return Err(invalid("duplicate workflow command key"));
325        }
326    }
327    Ok(())
328}
329fn checkpoint(state: &Value) -> Result<()> {
330    validate_wire_value(state)?;
331    bounded(state, WORKFLOW_CHECKPOINT_MAX_BYTES, "workflow checkpoint")
332}
333fn version(v: u8) -> Result<()> {
334    if v != WORKFLOW_VERSION {
335        return Err(invalid("unsupported workflow version"));
336    }
337    Ok(())
338}
339pub fn validate_workflow_error(error: &ApplicationError) -> Result<()> {
340    validate_error(error)
341}
342fn validate_error(error: &ApplicationError) -> Result<()> {
343    validate_text(&error.kind, 128)?;
344    if error.message.len() > 4096 {
345        return Err(invalid("workflow error message exceeds 4096 bytes"));
346    }
347    Ok(())
348}
349fn invalid(message: &str) -> ContractError {
350    ContractError::InvalidInput(message.into())
351}
352fn encoding(error: serde_json::Error) -> ContractError {
353    ContractError::InvalidInput(format!("invalid workflow JSON: {error}"))
354}
355fn bounded(value: &impl Serialize, bytes: usize, label: &str) -> Result<()> {
356    crate::submission::check_encoded_size(value, bytes, label).map_err(Into::into)
357}
358
359#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
360#[serde(rename_all = "snake_case")]
361pub enum WorkflowState {
362    Running,
363    Waiting,
364    Failing,
365    Cancelling,
366    Succeeded,
367    Failed,
368    Cancelled,
369}
370impl WorkflowState {
371    pub fn is_terminal(self) -> bool {
372        matches!(self, Self::Succeeded | Self::Failed | Self::Cancelled)
373    }
374}
375#[derive(Debug, Clone, Serialize, Deserialize)]
376#[serde(deny_unknown_fields)]
377pub struct WorkflowSnapshot {
378    /// Present together only for a nested owned workflow; roots omit both.
379    #[serde(default, skip_serializing_if = "Option::is_none")]
380    pub parent_workflow_id: Option<String>,
381    #[serde(default, skip_serializing_if = "Option::is_none")]
382    pub root_workflow_id: Option<String>,
383    pub workflow_id: String,
384    pub scope: Scope,
385    pub state: WorkflowState,
386    pub revision: u64,
387    #[serde(deserialize_with = "crate::observation::required_option")]
388    pub activation_id: Option<String>,
389    pub submitted_at: Timestamp,
390    #[serde(deserialize_with = "crate::observation::required_option")]
391    pub terminal_at: Option<Timestamp>,
392    #[serde(deserialize_with = "crate::observation::required_option")]
393    pub correlation_key: Option<String>,
394}
395#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
396#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
397pub enum WorkflowOutcome {
398    Succeeded {
399        #[serde(deserialize_with = "crate::observation::required_value")]
400        output: Value,
401    },
402    Failed {
403        error: ApplicationError,
404    },
405    Cancelled {},
406}
407#[derive(Debug, Clone, Serialize, Deserialize)]
408#[serde(deny_unknown_fields)]
409pub struct WorkflowResult {
410    pub workflow: WorkflowSnapshot,
411    #[serde(deserialize_with = "crate::observation::required_option")]
412    pub outcome: Option<WorkflowOutcome>,
413}
414
415#[derive(Debug, Clone, Serialize, Deserialize)]
416#[serde(deny_unknown_fields)]
417pub struct LocalResultCommand {
418    pub owner: LeaseOwner,
419    pub record: LocalStepRecord,
420}
421#[derive(Debug, Clone, Serialize, Deserialize)]
422#[serde(deny_unknown_fields)]
423pub struct LocalResultReceipt {
424    pub key: String,
425    pub already_accepted: bool,
426}
427#[derive(Debug, Clone)]
428pub struct WorkflowWork {
429    pub id: String,
430    pub token: String,
431    pub workflow_id: String,
432    pub source: WorkflowWorkSource,
433    /// Only controller outcomes are loaded here. Child completion notifications
434    /// stay compact; their payloads are read once a continuation needs them.
435    pub outcome: Option<TaskOutcome>,
436    /// Previously registered child bindings; their pinned descriptors win on replay.
437    pub resolved_children: Vec<ResolvedWorkflowChild>,
438}
439#[derive(Debug, Clone)]
440pub struct ResolvedWorkflowChild {
441    pub kind: WorkflowChildKind,
442    pub key: String,
443    pub descriptor: ProgramDescriptor,
444}
445#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
446pub struct WorkflowProgress {
447    pub processed: u32,
448    pub activations_scheduled: u32,
449    /// Newly registered ordinary tasks and owned workflow runs, combined.
450    pub children_scheduled: u32,
451}
452
453/// Optional workflow persistence over the same transactional authority as the
454/// application's task store. Implementations must atomically create tasks and
455/// dispatch obligations, append terminal completion work with task finalization,
456/// and apply checkpoints/child bindings/waits with their scheduling obligations.
457/// Separate, non-atomic task and workflow backends do not satisfy this port.
458///
459/// Child keys are workflow-scoped and retain their first normalized submission
460/// and pinned descriptor. Local keys are activation-scoped: exact records may be
461/// acknowledged again to their original accepting owner after expiry; a different
462/// attempt must hold current live authority. New records always require that
463/// authority and must be fenced against cancellation/continuation changes.
464/// Frozen activation inputs and revision never change as child events arrive.
465/// Work claims are bounded, leased, recoverable, and released before external
466/// program resolution; application rechecks ownership and persisted source state.
467/// Successful replies follow commit of every required write. Transport loss may
468/// leave a committed operation whose immutable identity must be reconciled.
469pub trait WorkflowStore: Send + Sync {
470    /// Accept a directly addressed, one-shot event after committing its receipt.
471    /// Exact source/ID/key/payload replays must reconcile before terminal checks.
472    /// Implementations serialize acceptance and wait resolution under workflow
473    /// authority; caller timestamps never decide event/deadline eligibility.
474    fn send_workflow_event<'a>(
475        &'a self,
476        command: &'a WorkflowEventCommand,
477    ) -> ContractFuture<'a, WorkflowEventReceipt>;
478    fn lookup_workflow_submission<'a>(
479        &'a self,
480        scope: &'a Scope,
481        key: &'a str,
482    ) -> ContractFuture<'a, Option<WorkflowSnapshot>>;
483    /// Replays the original normalized submission before resolving packages.
484    fn replay_workflow_submission<'a>(
485        &'a self,
486        command: &'a SubmitCommand,
487    ) -> ContractFuture<'a, Option<WorkflowSnapshot>>;
488    fn accept_resolved_workflow<'a>(
489        &'a self,
490        command: &'a SubmitCommand,
491        controller: &'a ProgramDescriptor,
492    ) -> ContractFuture<'a, WorkflowSnapshot>;
493    fn workflow_status<'a>(
494        &'a self,
495        scope: &'a Scope,
496        id: &'a str,
497    ) -> ContractFuture<'a, WorkflowSnapshot>;
498    fn workflow_result<'a>(
499        &'a self,
500        scope: &'a Scope,
501        id: &'a str,
502    ) -> ContractFuture<'a, WorkflowResult>;
503    fn activation_context<'a>(
504        &'a self,
505        owner: &'a LeaseOwner,
506    ) -> ContractFuture<'a, WorkflowActivationContext>;
507    fn record_local_result<'a>(
508        &'a self,
509        command: &'a LocalResultCommand,
510    ) -> ContractFuture<'a, LocalResultReceipt>;
511    /// Work leases own coordinator application, never worker execution. Claiming
512    /// releases database locks before any external descriptor resolution.
513    fn claim_work(&self, limit: u32) -> ContractFuture<'_, Vec<WorkflowWork>>;
514    fn apply_work<'a>(
515        &'a self,
516        work: &'a WorkflowWork,
517        resolved: &'a [ResolvedWorkflowChild],
518    ) -> ContractFuture<'a, WorkflowProgress>;
519    fn retry_work<'a>(&'a self, work: &'a WorkflowWork, reason: &'a str) -> ContractFuture<'a, ()>;
520    /// Persist permanent decision/application failure and drain owned children.
521    fn reject_work<'a>(
522        &'a self,
523        work: &'a WorkflowWork,
524        error: &'a ApplicationError,
525    ) -> ContractFuture<'a, ()>;
526    fn cancel_workflow<'a>(
527        &'a self,
528        scope: &'a Scope,
529        id: &'a str,
530    ) -> ContractFuture<'a, WorkflowSnapshot>;
531}
532
533/// Client and interactive-worker operations. Unsupported implementations must
534/// reject explicitly instead of silently submitting an ordinary task.
535pub trait WorkflowService: Send + Sync {
536    /// Accept a directly addressed, one-shot event after committing its receipt.
537    /// Exact source/ID/key/payload replays must reconcile before terminal checks.
538    /// Implementations serialize acceptance and wait resolution under workflow
539    /// authority; caller timestamps never decide event/deadline eligibility.
540    fn send_workflow_event<'a>(
541        &'a self,
542        command: &'a WorkflowEventCommand,
543    ) -> ContractFuture<'a, WorkflowEventReceipt>;
544    fn submit_workflow<'a>(
545        &'a self,
546        command: &'a SubmitCommand,
547    ) -> ContractFuture<'a, WorkflowSnapshot>;
548    fn workflow_status<'a>(
549        &'a self,
550        scope: &'a Scope,
551        id: &'a str,
552    ) -> ContractFuture<'a, WorkflowSnapshot>;
553    fn workflow_result<'a>(
554        &'a self,
555        scope: &'a Scope,
556        id: &'a str,
557    ) -> ContractFuture<'a, WorkflowResult>;
558    fn activation_context<'a>(
559        &'a self,
560        owner: &'a LeaseOwner,
561    ) -> ContractFuture<'a, WorkflowActivationContext>;
562    fn record_local_result<'a>(
563        &'a self,
564        command: &'a LocalResultCommand,
565    ) -> ContractFuture<'a, LocalResultReceipt>;
566    fn cancel_workflow<'a>(
567        &'a self,
568        scope: &'a Scope,
569        id: &'a str,
570    ) -> ContractFuture<'a, WorkflowSnapshot>;
571}
572
573impl WorkflowSnapshot {
574    pub fn validate(&self) -> Result<()> {
575        self.scope.validate()?;
576        validate_workflow_lineage(
577            Some(&self.workflow_id),
578            self.parent_workflow_id.as_deref(),
579            self.root_workflow_id.as_deref(),
580        )?;
581        if let Some(id) = &self.activation_id {
582            validate_text(id, 128)?;
583        }
584        if self.state.is_terminal() != self.terminal_at.is_some()
585            || (self.state.is_terminal() && self.activation_id.is_some())
586            || (self.state == WorkflowState::Running && self.activation_id.is_none())
587            || self.terminal_at.is_some_and(|at| at < self.submitted_at)
588            || self.submitted_at > 253_402_300_799_999
589            || self.terminal_at.is_some_and(|at| at > 253_402_300_799_999)
590        {
591            return Err(invalid("inconsistent workflow status"));
592        }
593        if self
594            .correlation_key
595            .as_ref()
596            .is_some_and(|key| key.len() > 512 || key.chars().any(char::is_control))
597        {
598            return Err(invalid("invalid workflow correlation key"));
599        }
600        Ok(())
601    }
602}
603impl WorkflowResult {
604    pub fn validate(&self) -> Result<()> {
605        self.workflow.validate()?;
606        match (&self.outcome, self.workflow.state) {
607            (None, state) if !state.is_terminal() => Ok(()),
608            (Some(WorkflowOutcome::Succeeded { output }), WorkflowState::Succeeded) => {
609                validate_wire_value(output)?;
610                bounded(output, WORKFLOW_DECISION_MAX_BYTES, "workflow output")
611            }
612            (Some(WorkflowOutcome::Failed { error }), WorkflowState::Failed) => {
613                validate_error(error)
614            }
615            (Some(WorkflowOutcome::Cancelled {}), WorkflowState::Cancelled) => Ok(()),
616            _ => Err(invalid("inconsistent workflow result")),
617        }
618    }
619}
620
621#[cfg(test)]
622mod tests {
623    use super::*;
624    use serde_json::json;
625    fn decision() -> Value {
626        json!({"v":1,"activation_id":"task_1","revision":0,"kind":"suspend","continuation":"after","state":null,"commands":[],"until":[]})
627    }
628    #[test]
629    fn decisions_require_exact_versioned_shape_and_sealed_unique_membership() {
630        assert!(WorkflowDecision::decode(&decision()).is_ok());
631        for (key, value) in [
632            ("v", json!(2)),
633            ("commands", json!(null)),
634            ("until", json!(["a", "a"])),
635            ("extra", json!(true)),
636        ] {
637            let mut input = decision();
638            input[key] = value;
639            assert!(WorkflowDecision::decode(&input).is_err(), "{key}");
640        }
641        let mut input = decision();
642        input.as_object_mut().unwrap().remove("state");
643        assert!(WorkflowDecision::decode(&input).is_err());
644    }
645    #[test]
646    fn durable_local_replay_binds_numeric_representation_and_result() {
647        let record = LocalStepRecord {
648            key: "fetch".into(),
649            callable: "app:fetch".into(),
650            input: json!({"a":1,"b":2}),
651            output: Value::Null,
652        };
653        let mut replay = record.clone();
654        replay.input = json!({"b":2,"a":1});
655        assert!(record.matches(&replay).unwrap());
656        replay.input = json!({"a":1.0,"b":2});
657        assert!(!record.matches(&replay).unwrap());
658        replay = record.clone();
659        replay.output = json!(false);
660        assert!(!record.matches(&replay).unwrap());
661    }
662    #[test]
663    fn oversized_checkpoints_fail_before_persistence() {
664        let mut input = decision();
665        input["state"] = Value::String("x".repeat(WORKFLOW_CHECKPOINT_MAX_BYTES));
666        assert!(WorkflowDecision::decode(&input).is_err());
667    }
668    #[test]
669    fn activation_context_rejects_nested_child_values_beyond_the_application_bound() {
670        let mut output = Value::Null;
671        for _ in 0..64 {
672            output = json!([output]);
673        }
674        let child = WorkflowChildResult::Task(WorkflowTaskResult {
675            task_id: "child".into(),
676            state: TaskState::Succeeded,
677            outcome: TaskOutcome::Succeeded {
678                attempt_id: "attempt".into(),
679                quiescence: Quiescence::Confirmed,
680                execution_may_have_started: true,
681                output,
682            },
683        });
684        let mut context = WorkflowActivationContext {
685            parent_workflow_id: None,
686            root_workflow_id: None,
687            v: 1,
688            workflow_id: "workflow".into(),
689            activation_id: "activation".into(),
690            revision: 0,
691            continuation: "next".into(),
692            state: Value::Null,
693            inputs: BTreeMap::from([("child".into(), child)]),
694            local_steps: vec![],
695            wake: None,
696        };
697        context.validate().unwrap();
698        let WorkflowChildResult::Task(child) = context.inputs.get_mut("child").unwrap() else {
699            unreachable!()
700        };
701        let TaskOutcome::Succeeded { output, .. } = &mut child.outcome else {
702            unreachable!()
703        };
704        *output = json!([output.take()]);
705        assert!(context.validate().is_err());
706    }
707
708    #[test]
709    fn workflow_wire_distinguishes_explicit_null_from_missing_required_fields() {
710        let snapshot = json!({"workflow_id":"workflow","scope":{"tenant_id":"t","namespace":"n"},"state":"waiting","revision":1,"activation_id":null,"submitted_at":1,"terminal_at":null,"correlation_key":null});
711        serde_json::from_value::<WorkflowSnapshot>(snapshot.clone())
712            .unwrap()
713            .validate()
714            .unwrap();
715        for key in ["activation_id", "terminal_at", "correlation_key"] {
716            let mut missing = snapshot.clone();
717            missing.as_object_mut().unwrap().remove(key);
718            assert!(
719                serde_json::from_value::<WorkflowSnapshot>(missing).is_err(),
720                "{key}"
721            );
722        }
723        assert!(
724            serde_json::from_value::<WorkflowResult>(json!({"workflow":snapshot,"outcome":null}))
725                .is_ok()
726        );
727        assert!(serde_json::from_value::<WorkflowResult>(json!({"workflow":snapshot})).is_err());
728        assert!(
729            serde_json::from_value::<WorkflowOutcome>(json!({"kind":"succeeded","output":null}))
730                .is_ok()
731        );
732        assert!(serde_json::from_value::<WorkflowOutcome>(json!({"kind":"succeeded"})).is_err());
733        let local = json!({"key":"key","callable":"app:f","input":null,"output":null});
734        serde_json::from_value::<LocalStepRecord>(local.clone())
735            .unwrap()
736            .validate()
737            .unwrap();
738        for key in ["input", "output"] {
739            let mut missing = local.clone();
740            missing.as_object_mut().unwrap().remove(key);
741            assert!(serde_json::from_value::<LocalStepRecord>(missing).is_err());
742        }
743    }
744}
ledgence_orchestration_api/workflow.rs

ledgence_orchestration_api/
workflow.rs
