a3s_flow/engine/
mod.rs

1use chrono::{DateTime, Duration as ChronoDuration, Utc};
2use std::sync::Arc;
3use tokio::task::JoinSet;
4use uuid::Uuid;
5
6use crate::error::{FlowError, Result};
7use crate::model::{
8    project_run, ActiveHookSnapshot, FlowEvent, FlowEventEnvelope, HookStatus, RetryPolicy,
9    RuntimeCommand, StepCommand, StepFailureAction, StepStatus, WaitStatus, WorkflowRunSnapshot,
10    WorkflowRunSummary, WorkflowRunSuspension, WorkflowSpec,
11};
12use crate::observe::{FlowEventObserver, NoopFlowEventObserver};
13use crate::runtime::{FlowRuntime, StepInvocation, WorkflowInvocation};
14use crate::store::{FlowEventStore, InMemoryEventStore};
15
16mod validation;
17use validation::{
18    ensure_hook_command_matches, ensure_same_start, ensure_step_batch_valid,
19    ensure_step_command_matches, ensure_wait_command_matches, is_event_conflict, validate_run_id,
20};
21
22/// Builder for a [`FlowEngine`].
23pub struct FlowEngineBuilder {
24    store: Arc<dyn FlowEventStore>,
25    runtime: Arc<dyn FlowRuntime>,
26    observer: Arc<dyn FlowEventObserver>,
27    max_replay_iterations: usize,
28}
29
30impl FlowEngineBuilder {
31    pub fn new(runtime: Arc<dyn FlowRuntime>) -> Self {
32        Self {
33            store: Arc::new(InMemoryEventStore::new()),
34            runtime,
35            observer: Arc::new(NoopFlowEventObserver),
36            max_replay_iterations: 1024,
37        }
38    }
39
40    pub fn with_store(mut self, store: Arc<dyn FlowEventStore>) -> Self {
41        self.store = store;
42        self
43    }
44
45    pub fn with_observer(mut self, observer: Arc<dyn FlowEventObserver>) -> Self {
46        self.observer = observer;
47        self
48    }
49
50    pub fn with_max_replay_iterations(mut self, max_replay_iterations: usize) -> Self {
51        self.max_replay_iterations = max_replay_iterations.max(1);
52        self
53    }
54
55    pub fn build(self) -> FlowEngine {
56        FlowEngine {
57            store: self.store,
58            runtime: self.runtime,
59            observer: self.observer,
60            max_replay_iterations: self.max_replay_iterations,
61        }
62    }
63}
64
65/// Event-sourced workflow engine.
66#[derive(Clone)]
67pub struct FlowEngine {
68    store: Arc<dyn FlowEventStore>,
69    runtime: Arc<dyn FlowRuntime>,
70    observer: Arc<dyn FlowEventObserver>,
71    max_replay_iterations: usize,
72}
73
74struct StepExecutionContext {
75    step_id: String,
76    step_name: String,
77    input: serde_json::Value,
78    retry: RetryPolicy,
79    now: DateTime<Utc>,
80}
81
82impl FlowEngine {
83    pub fn builder(runtime: Arc<dyn FlowRuntime>) -> FlowEngineBuilder {
84        FlowEngineBuilder::new(runtime)
85    }
86
87    pub fn new(store: Arc<dyn FlowEventStore>, runtime: Arc<dyn FlowRuntime>) -> Self {
88        Self {
89            store,
90            runtime,
91            observer: Arc::new(NoopFlowEventObserver),
92            max_replay_iterations: 1024,
93        }
94    }
95
96    pub fn in_memory(runtime: Arc<dyn FlowRuntime>) -> Self {
97        Self::new(Arc::new(InMemoryEventStore::new()), runtime)
98    }
99
100    pub fn store(&self) -> Arc<dyn FlowEventStore> {
101        Arc::clone(&self.store)
102    }
103
104    pub fn observer(&self) -> Arc<dyn FlowEventObserver> {
105        Arc::clone(&self.observer)
106    }
107
108    /// Start a workflow run and drive it until completion or suspension.
109    pub async fn start(&self, spec: WorkflowSpec, input: serde_json::Value) -> Result<String> {
110        let run_id = Uuid::new_v4().to_string();
111        self.start_with_id(run_id, spec, input).await
112    }
113
114    /// Start a workflow run using a caller-provided durable run id.
115    ///
116    /// Reusing the same `run_id` with the same workflow spec and input is
117    /// idempotent. Reusing it with different spec or input returns a conflict.
118    pub async fn start_with_id(
119        &self,
120        run_id: impl Into<String>,
121        spec: WorkflowSpec,
122        input: serde_json::Value,
123    ) -> Result<String> {
124        spec.validate()?;
125        let run_id = run_id.into();
126        validate_run_id(&run_id)?;
127
128        for _ in 0..self.max_replay_iterations {
129            match self.store.list(&run_id).await {
130                Ok(history) => {
131                    let snapshot = project_run(&run_id, &history)?;
132                    ensure_same_start(&run_id, &snapshot, &spec, &input)?;
133                    if !history
134                        .iter()
135                        .any(|event| matches!(event.event, FlowEvent::RunStarted))
136                    {
137                        match self
138                            .record_event_at(&run_id, snapshot.last_sequence, FlowEvent::RunStarted)
139                            .await
140                        {
141                            Ok(_) => {}
142                            Err(err) if is_event_conflict(&err) => continue,
143                            Err(err) => return Err(err),
144                        }
145                    }
146                    match self.drive(&run_id).await {
147                        Ok(_) => return Ok(run_id),
148                        Err(err) if is_event_conflict(&err) => continue,
149                        Err(err) => return Err(err),
150                    }
151                }
152                Err(FlowError::RunNotFound(_)) => {
153                    let created = match self
154                        .record_event_at(
155                            &run_id,
156                            0,
157                            FlowEvent::RunCreated {
158                                spec: spec.clone(),
159                                input: input.clone(),
160                            },
161                        )
162                        .await
163                    {
164                        Ok(created) => created,
165                        Err(err) if is_event_conflict(&err) => continue,
166                        Err(err) => return Err(err),
167                    };
168                    match self
169                        .record_event_at(&run_id, created.sequence, FlowEvent::RunStarted)
170                        .await
171                    {
172                        Ok(_) => {}
173                        Err(err) if is_event_conflict(&err) => continue,
174                        Err(err) => return Err(err),
175                    }
176                    match self.drive(&run_id).await {
177                        Ok(_) => return Ok(run_id),
178                        Err(err) if is_event_conflict(&err) => continue,
179                        Err(err) => return Err(err),
180                    }
181                }
182                Err(err) => return Err(err),
183            }
184        }
185
186        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
187    }
188
189    /// Resume a wait once its timer has fired.
190    pub async fn resume_wait(&self, run_id: &str, wait_id: &str) -> Result<()> {
191        for _ in 0..self.max_replay_iterations {
192            let snapshot = self.snapshot(run_id).await?;
193            if snapshot.status.is_terminal() {
194                return Err(FlowError::RunTerminal(run_id.to_string()));
195            }
196            match snapshot.waits.get(wait_id) {
197                Some(wait) if wait.status == WaitStatus::Waiting => {
198                    match self
199                        .record_event_at(
200                            run_id,
201                            snapshot.last_sequence,
202                            FlowEvent::WaitCompleted {
203                                wait_id: wait_id.to_string(),
204                            },
205                        )
206                        .await
207                    {
208                        Ok(_) => {}
209                        Err(err) if is_event_conflict(&err) => continue,
210                        Err(err) => return Err(err),
211                    }
212                    match self.drive(run_id).await {
213                        Ok(_) => return Ok(()),
214                        Err(err) if is_event_conflict(&err) => continue,
215                        Err(err) => return Err(err),
216                    }
217                }
218                Some(_) => match self.drive(run_id).await {
219                    Ok(_) => return Ok(()),
220                    Err(err) if is_event_conflict(&err) => continue,
221                    Err(err) => return Err(err),
222                },
223                None => {
224                    return Err(FlowError::InvalidTransition(format!(
225                        "wait {wait_id} does not exist for run {run_id}"
226                    )))
227                }
228            }
229        }
230
231        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
232    }
233
234    /// Resume an active hook with external payload.
235    pub async fn resume_hook(
236        &self,
237        run_id: &str,
238        hook_id: &str,
239        payload: serde_json::Value,
240    ) -> Result<()> {
241        for _ in 0..self.max_replay_iterations {
242            let snapshot = self.snapshot(run_id).await?;
243            if snapshot.status.is_terminal() {
244                return Err(FlowError::RunTerminal(run_id.to_string()));
245            }
246            match snapshot.hooks.get(hook_id) {
247                Some(hook) if hook.status == HookStatus::Active => {
248                    match self
249                        .record_event_at(
250                            run_id,
251                            snapshot.last_sequence,
252                            FlowEvent::HookReceived {
253                                hook_id: hook_id.to_string(),
254                                payload: payload.clone(),
255                            },
256                        )
257                        .await
258                    {
259                        Ok(_) => {}
260                        Err(err) if is_event_conflict(&err) => continue,
261                        Err(err) => return Err(err),
262                    }
263                    match self.drive(run_id).await {
264                        Ok(_) => return Ok(()),
265                        Err(err) if is_event_conflict(&err) => continue,
266                        Err(err) => return Err(err),
267                    }
268                }
269                Some(_) => match self.drive(run_id).await {
270                    Ok(_) => return Ok(()),
271                    Err(err) if is_event_conflict(&err) => continue,
272                    Err(err) => return Err(err),
273                },
274                None => {
275                    return Err(FlowError::InvalidTransition(format!(
276                        "hook {hook_id} does not exist for run {run_id}"
277                    )))
278                }
279            }
280        }
281
282        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
283    }
284
285    /// Dispose an active hook without accepting a callback payload.
286    ///
287    /// This is useful when a host withdraws an approval request, expires a
288    /// webhook token, or closes an external callback route. The workflow is
289    /// driven after the disposal event so replay code can observe
290    /// [`WorkflowContext::hook_disposed`](crate::context::WorkflowContext::hook_disposed)
291    /// and complete, fail, or schedule an alternate path.
292    pub async fn dispose_hook(&self, run_id: &str, hook_id: &str) -> Result<()> {
293        for _ in 0..self.max_replay_iterations {
294            let snapshot = self.snapshot(run_id).await?;
295            if snapshot.status.is_terminal() {
296                return Err(FlowError::RunTerminal(run_id.to_string()));
297            }
298            match snapshot.hooks.get(hook_id) {
299                Some(hook) if hook.status == HookStatus::Active => {
300                    match self
301                        .record_event_at(
302                            run_id,
303                            snapshot.last_sequence,
304                            FlowEvent::HookDisposed {
305                                hook_id: hook_id.to_string(),
306                            },
307                        )
308                        .await
309                    {
310                        Ok(_) => {}
311                        Err(err) if is_event_conflict(&err) => continue,
312                        Err(err) => return Err(err),
313                    }
314                    match self.drive(run_id).await {
315                        Ok(_) => return Ok(()),
316                        Err(err) if is_event_conflict(&err) => continue,
317                        Err(err) => return Err(err),
318                    }
319                }
320                Some(_) => match self.drive(run_id).await {
321                    Ok(_) => return Ok(()),
322                    Err(err) if is_event_conflict(&err) => continue,
323                    Err(err) => return Err(err),
324                },
325                None => {
326                    return Err(FlowError::InvalidTransition(format!(
327                        "hook {hook_id} does not exist for run {run_id}"
328                    )))
329                }
330            }
331        }
332
333        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
334    }
335
336    /// Resume an active hook by its external token.
337    ///
338    /// This is the API webhook handlers normally want: the callback receives a
339    /// token, while `run_id` and `hook_id` remain engine internals.
340    pub async fn resume_hook_by_token(
341        &self,
342        token: &str,
343        payload: serde_json::Value,
344    ) -> Result<(String, String)> {
345        let mut matches = Vec::new();
346        for run_id in self.store.list_run_ids().await? {
347            let snapshot = self.snapshot(&run_id).await?;
348            if snapshot.status.is_terminal() {
349                continue;
350            }
351            for hook in snapshot.hooks.values() {
352                if hook.status == HookStatus::Active && hook.token == token {
353                    matches.push((run_id.clone(), hook.hook_id.clone()));
354                }
355            }
356        }
357
358        match matches.len() {
359            0 => Err(FlowError::HookTokenNotFound(token.to_string())),
360            1 => {
361                let (run_id, hook_id) = matches.remove(0);
362                self.resume_hook(&run_id, &hook_id, payload).await?;
363                Ok((run_id, hook_id))
364            }
365            _ => Err(FlowError::InvalidTransition(format!(
366                "hook token {token:?} is active in multiple runs"
367            ))),
368        }
369    }
370
371    /// Dispose an active hook by its external token.
372    ///
373    /// This mirrors [`resume_hook_by_token`](Self::resume_hook_by_token) for
374    /// callback routers that only know the public token.
375    pub async fn dispose_hook_by_token(&self, token: &str) -> Result<(String, String)> {
376        let mut matches = Vec::new();
377        for run_id in self.store.list_run_ids().await? {
378            let snapshot = self.snapshot(&run_id).await?;
379            if snapshot.status.is_terminal() {
380                continue;
381            }
382            for hook in snapshot.hooks.values() {
383                if hook.status == HookStatus::Active && hook.token == token {
384                    matches.push((run_id.clone(), hook.hook_id.clone()));
385                }
386            }
387        }
388
389        match matches.len() {
390            0 => Err(FlowError::HookTokenNotFound(token.to_string())),
391            1 => {
392                let (run_id, hook_id) = matches.remove(0);
393                self.dispose_hook(&run_id, &hook_id).await?;
394                Ok((run_id, hook_id))
395            }
396            _ => Err(FlowError::InvalidTransition(format!(
397                "hook token {token:?} is active in multiple runs"
398            ))),
399        }
400    }
401
402    /// List active waits whose `resume_at` is at or before `now`.
403    ///
404    /// Scheduler integrations can use this to inspect due timers before
405    /// deciding how aggressively to drive them.
406    pub async fn list_due_waits(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
407        let mut due = Vec::new();
408        for run_id in self.store.list_run_ids().await? {
409            let snapshot = self.snapshot(&run_id).await?;
410            if snapshot.status.is_terminal() {
411                continue;
412            }
413            for wait in snapshot.waits.values() {
414                if wait.status == WaitStatus::Waiting && wait.resume_at <= now {
415                    due.push((run_id.clone(), wait.wait_id.clone()));
416                }
417            }
418        }
419        due.sort();
420        Ok(due)
421    }
422
423    /// Complete every due wait and drive the affected workflows.
424    ///
425    /// Returns the `(run_id, wait_id)` pairs that were resumed. A wait already
426    /// completed by another caller is skipped by [`Self::resume_wait`].
427    pub async fn resume_due_waits(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
428        let due = self.list_due_waits(now).await?;
429        let mut resumed = Vec::with_capacity(due.len());
430        for (run_id, wait_id) in due {
431            self.resume_wait(&run_id, &wait_id).await?;
432            resumed.push((run_id, wait_id));
433        }
434        Ok(resumed)
435    }
436
437    /// List pending step retries whose `retry_after` is at or before `now`.
438    pub async fn list_due_retries(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
439        let mut due = Vec::new();
440        for run_id in self.store.list_run_ids().await? {
441            let snapshot = self.snapshot(&run_id).await?;
442            if snapshot.status.is_terminal() {
443                continue;
444            }
445            for (step_id, _) in snapshot.due_retries(now) {
446                due.push((run_id.clone(), step_id));
447            }
448        }
449        due.sort();
450        Ok(due)
451    }
452
453    /// Drive every run with a due step retry.
454    pub async fn resume_due_retries(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
455        let due = self.list_due_retries(now).await?;
456        let mut run_ids = Vec::new();
457        for (run_id, _) in &due {
458            if !run_ids.contains(run_id) {
459                run_ids.push(run_id.clone());
460            }
461        }
462        for run_id in run_ids {
463            self.drive_at(&run_id, now).await?;
464        }
465        Ok(due)
466    }
467
468    pub async fn cancel(&self, run_id: &str, reason: Option<String>) -> Result<()> {
469        for _ in 0..self.max_replay_iterations {
470            let snapshot = self.snapshot(run_id).await?;
471            if snapshot.status.is_terminal() {
472                return Ok(());
473            }
474            match self
475                .record_event_at(
476                    run_id,
477                    snapshot.last_sequence,
478                    FlowEvent::RunCancelled {
479                        reason: reason.clone(),
480                    },
481                )
482                .await
483            {
484                Ok(_) => return Ok(()),
485                Err(err) if is_event_conflict(&err) => continue,
486                Err(err) => return Err(err),
487            }
488        }
489
490        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
491    }
492
493    pub async fn snapshot(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
494        let history = self.store.list(run_id).await?;
495        project_run(run_id, &history)
496    }
497
498    pub async fn history(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>> {
499        self.store.list(run_id).await
500    }
501
502    pub async fn list_run_ids(&self) -> Result<Vec<String>> {
503        self.store.list_run_ids().await
504    }
505
506    pub async fn list_snapshots(&self) -> Result<Vec<WorkflowRunSnapshot>> {
507        let mut snapshots = Vec::new();
508        for run_id in self.store.list_run_ids().await? {
509            snapshots.push(self.snapshot(&run_id).await?);
510        }
511        Ok(snapshots)
512    }
513
514    /// Summarize run state across the active store.
515    ///
516    /// Suspension counters include only non-terminal runs, so a cancelled run
517    /// that still has an old wait or hook in history is not reported as
518    /// actionable work.
519    pub async fn run_summary(&self) -> Result<WorkflowRunSummary> {
520        let snapshots = self.list_snapshots().await?;
521        Ok(WorkflowRunSummary::from_snapshots(&snapshots))
522    }
523
524    /// List open waits, active hooks, and pending delayed retries.
525    ///
526    /// The `due` flag on wait and retry suspensions is computed against `now`.
527    /// Terminal runs are skipped so cancelled histories do not produce
528    /// actionable operator work.
529    pub async fn list_open_suspensions(
530        &self,
531        now: DateTime<Utc>,
532    ) -> Result<Vec<WorkflowRunSuspension>> {
533        let mut suspensions = Vec::new();
534        for run_id in self.store.list_run_ids().await? {
535            let snapshot = self.snapshot(&run_id).await?;
536            if snapshot.status.is_terminal() {
537                continue;
538            }
539            for wait in snapshot.waits.values() {
540                if wait.status == WaitStatus::Waiting {
541                    suspensions.push(WorkflowRunSuspension::Wait {
542                        run_id: run_id.clone(),
543                        wait: wait.clone(),
544                        due: wait.resume_at <= now,
545                    });
546                }
547            }
548            for hook in snapshot.hooks.values() {
549                if hook.status == HookStatus::Active {
550                    suspensions.push(WorkflowRunSuspension::Hook {
551                        run_id: run_id.clone(),
552                        hook: hook.clone(),
553                    });
554                }
555            }
556            for step in snapshot.steps.values() {
557                if step.status == StepStatus::Pending {
558                    if let Some(retry_after) = step.retry_after {
559                        suspensions.push(WorkflowRunSuspension::Retry {
560                            run_id: run_id.clone(),
561                            step: step.clone(),
562                            due: retry_after <= now,
563                        });
564                    }
565                }
566            }
567        }
568        suspensions.sort_by(|left, right| {
569            (left.run_id(), left.kind_order(), left.subject_id()).cmp(&(
570                right.run_id(),
571                right.kind_order(),
572                right.subject_id(),
573            ))
574        });
575        Ok(suspensions)
576    }
577
578    /// Return the earliest open wait or delayed retry across non-terminal runs.
579    ///
580    /// This is useful for hosts that want to sleep until the next scheduler tick
581    /// instead of polling at a fixed interval. Active hooks are intentionally
582    /// ignored because they do not have a scheduled wake-up time.
583    pub async fn next_wakeup(&self, now: DateTime<Utc>) -> Result<Option<WorkflowRunSuspension>> {
584        let mut wakeups = self.list_open_suspensions(now).await?;
585        wakeups.retain(|suspension| suspension.scheduled_at().is_some());
586        wakeups.sort_by(|left, right| {
587            (
588                left.scheduled_at(),
589                left.run_id(),
590                left.kind_order(),
591                left.subject_id(),
592            )
593                .cmp(&(
594                    right.scheduled_at(),
595                    right.run_id(),
596                    right.kind_order(),
597                    right.subject_id(),
598                ))
599        });
600        Ok(wakeups.into_iter().next())
601    }
602
603    /// List active external callback hooks across non-terminal runs.
604    ///
605    /// Callback routers and dashboards can use this to discover public hook
606    /// tokens and their audit metadata without projecting every run manually.
607    /// The result is sorted by run ID and hook ID for stable polling output.
608    pub async fn list_active_hooks(&self) -> Result<Vec<ActiveHookSnapshot>> {
609        let mut hooks = Vec::new();
610        for run_id in self.store.list_run_ids().await? {
611            let snapshot = self.snapshot(&run_id).await?;
612            if snapshot.status.is_terminal() {
613                continue;
614            }
615            for hook in snapshot.hooks.values() {
616                if hook.status == HookStatus::Active {
617                    hooks.push(ActiveHookSnapshot {
618                        run_id: run_id.clone(),
619                        hook: hook.clone(),
620                    });
621                }
622            }
623        }
624        hooks.sort_by(|left, right| {
625            (left.run_id.as_str(), left.hook.hook_id.as_str())
626                .cmp(&(right.run_id.as_str(), right.hook.hook_id.as_str()))
627        });
628        Ok(hooks)
629    }
630
631    /// Replay and dispatch until the run reaches a terminal state or an open
632    /// wait/hook suspension.
633    pub async fn drive(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
634        self.drive_at(run_id, Utc::now()).await
635    }
636
637    async fn drive_at(&self, run_id: &str, now: DateTime<Utc>) -> Result<WorkflowRunSnapshot> {
638        'replay: for _ in 0..self.max_replay_iterations {
639            let history = self.store.list(run_id).await?;
640            let snapshot = project_run(run_id, &history)?;
641            if snapshot.status.is_terminal()
642                || snapshot
643                    .waits
644                    .values()
645                    .any(|wait| wait.status == WaitStatus::Waiting)
646                || snapshot
647                    .hooks
648                    .values()
649                    .any(|hook| hook.status == HookStatus::Active)
650                || (snapshot.has_future_retry(now) && snapshot.due_retries(now).is_empty())
651            {
652                return Ok(snapshot);
653            }
654
655            let command = self
656                .runtime
657                .run_workflow(WorkflowInvocation {
658                    run_id: run_id.to_string(),
659                    spec: snapshot.spec.clone(),
660                    input: snapshot.input.clone(),
661                    history,
662                })
663                .await?;
664
665            match command {
666                RuntimeCommand::Complete { output } => {
667                    match self
668                        .record_event_at(
669                            run_id,
670                            snapshot.last_sequence,
671                            FlowEvent::RunCompleted { output },
672                        )
673                        .await
674                    {
675                        Ok(_) => {}
676                        Err(err) if is_event_conflict(&err) => continue,
677                        Err(err) => return Err(err),
678                    }
679                    return self.snapshot(run_id).await;
680                }
681                RuntimeCommand::Fail { error } => {
682                    match self
683                        .record_event_at(
684                            run_id,
685                            snapshot.last_sequence,
686                            FlowEvent::RunFailed { error },
687                        )
688                        .await
689                    {
690                        Ok(_) => {}
691                        Err(err) if is_event_conflict(&err) => continue,
692                        Err(err) => return Err(err),
693                    }
694                    return self.snapshot(run_id).await;
695                }
696                RuntimeCommand::ScheduleStep {
697                    step_id,
698                    step_name,
699                    input,
700                    retry,
701                } => {
702                    if let Some(step) = snapshot.steps.get(&step_id) {
703                        ensure_step_command_matches(run_id, step, &step_name, &input, retry)?;
704                        if matches!(step.status, StepStatus::Completed | StepStatus::Failed) {
705                            return Err(FlowError::InvalidTransition(format!(
706                                "workflow rescheduled terminal step {step_id} without progress"
707                            )));
708                        }
709                    }
710                    match self
711                        .execute_step(
712                            run_id,
713                            &snapshot,
714                            StepExecutionContext {
715                                step_id,
716                                step_name,
717                                input,
718                                retry,
719                                now,
720                            },
721                        )
722                        .await
723                    {
724                        Ok(()) => {}
725                        Err(err) if is_event_conflict(&err) => continue,
726                        Err(err) => return Err(err),
727                    }
728                }
729                RuntimeCommand::ScheduleSteps { steps } => {
730                    ensure_step_batch_valid(&steps)?;
731                    for step in &steps {
732                        if let Some(existing) = snapshot.steps.get(&step.step_id) {
733                            ensure_step_command_matches(
734                                run_id,
735                                existing,
736                                &step.step_name,
737                                &step.input,
738                                step.retry,
739                            )?;
740                        }
741                    }
742                    if steps.iter().all(|step| {
743                        snapshot.steps.get(&step.step_id).is_some_and(|existing| {
744                            matches!(existing.status, StepStatus::Completed | StepStatus::Failed)
745                        })
746                    }) {
747                        let step_ids = steps
748                            .iter()
749                            .map(|step| step.step_id.as_str())
750                            .collect::<Vec<_>>()
751                            .join(", ");
752                        return Err(FlowError::InvalidTransition(format!(
753                            "workflow rescheduled only terminal steps without progress: {step_ids}"
754                        )));
755                    }
756                    match self.execute_step_batch(run_id, &snapshot, steps, now).await {
757                        Ok(()) => {}
758                        Err(err) if is_event_conflict(&err) => continue 'replay,
759                        Err(err) => return Err(err),
760                    }
761                }
762                RuntimeCommand::WaitUntil { wait_id, resume_at } => {
763                    match snapshot.waits.get(&wait_id) {
764                        Some(wait) => {
765                            ensure_wait_command_matches(run_id, wait, resume_at)?;
766                            if wait.status == WaitStatus::Completed {
767                                continue;
768                            }
769                            return self.snapshot(run_id).await;
770                        }
771                        None => {
772                            match self
773                                .record_event_at(
774                                    run_id,
775                                    snapshot.last_sequence,
776                                    FlowEvent::WaitCreated { wait_id, resume_at },
777                                )
778                                .await
779                            {
780                                Ok(_) => {}
781                                Err(err) if is_event_conflict(&err) => continue,
782                                Err(err) => return Err(err),
783                            }
784                            return self.snapshot(run_id).await;
785                        }
786                    }
787                }
788                RuntimeCommand::CreateHook {
789                    hook_id,
790                    token,
791                    metadata,
792                } => match snapshot.hooks.get(&hook_id) {
793                    Some(hook) => {
794                        ensure_hook_command_matches(run_id, hook, &token, &metadata)?;
795                        if matches!(hook.status, HookStatus::Received | HookStatus::Disposed) {
796                            continue;
797                        }
798                        return self.snapshot(run_id).await;
799                    }
800                    None => {
801                        self.ensure_hook_token_available(run_id, &hook_id, &token)
802                            .await?;
803                        match self
804                            .record_event_at(
805                                run_id,
806                                snapshot.last_sequence,
807                                FlowEvent::HookCreated {
808                                    hook_id,
809                                    token,
810                                    metadata,
811                                },
812                            )
813                            .await
814                        {
815                            Ok(_) => {}
816                            Err(err) if is_event_conflict(&err) => continue,
817                            Err(err) => return Err(err),
818                        }
819                        return self.snapshot(run_id).await;
820                    }
821                },
822            }
823        }
824
825        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
826    }
827
828    async fn record_event_at(
829        &self,
830        run_id: &str,
831        expected_sequence: u64,
832        event: FlowEvent,
833    ) -> Result<FlowEventEnvelope> {
834        let envelope = self
835            .store
836            .append_if_sequence(run_id, expected_sequence, event)
837            .await?;
838        self.observer.observe(envelope.clone()).await;
839        Ok(envelope)
840    }
841
842    async fn ensure_hook_token_available(
843        &self,
844        run_id: &str,
845        hook_id: &str,
846        token: &str,
847    ) -> Result<()> {
848        for existing_run_id in self.store.list_run_ids().await? {
849            let snapshot = self.snapshot(&existing_run_id).await?;
850            if snapshot.status.is_terminal() {
851                continue;
852            }
853            for hook in snapshot.hooks.values() {
854                if hook.status != HookStatus::Active || hook.token != token {
855                    continue;
856                }
857                if existing_run_id == run_id && hook.hook_id == hook_id {
858                    continue;
859                }
860                return Err(FlowError::HookTokenConflict {
861                    token: token.to_string(),
862                    existing_run_id,
863                    existing_hook_id: hook.hook_id.clone(),
864                });
865            }
866        }
867        Ok(())
868    }
869
870    async fn execute_step(
871        &self,
872        run_id: &str,
873        snapshot: &WorkflowRunSnapshot,
874        context: StepExecutionContext,
875    ) -> Result<()> {
876        let StepExecutionContext {
877            step_id,
878            step_name,
879            input,
880            retry,
881            now,
882        } = context;
883        let mut expected_sequence = snapshot.last_sequence;
884        if let Some(step) = snapshot.steps.get(&step_id) {
885            ensure_step_command_matches(run_id, step, &step_name, &input, retry)?;
886            if matches!(step.status, StepStatus::Completed | StepStatus::Failed) {
887                return Ok(());
888            }
889            if step.status == StepStatus::Pending
890                && step
891                    .retry_after
892                    .is_some_and(|retry_after| retry_after > now)
893            {
894                return Ok(());
895            }
896        } else {
897            let envelope = self
898                .record_event_at(
899                    run_id,
900                    expected_sequence,
901                    FlowEvent::StepCreated {
902                        step_id: step_id.clone(),
903                        step_name: step_name.clone(),
904                        input: input.clone(),
905                        retry,
906                    },
907                )
908                .await?;
909            expected_sequence = envelope.sequence;
910        }
911
912        let max_attempts = retry.max_attempts.max(1);
913        let mut attempt = snapshot
914            .steps
915            .get(&step_id)
916            .map(|step| step.attempt)
917            .unwrap_or(0);
918        let mut redelivering_running_step = snapshot
919            .steps
920            .get(&step_id)
921            .is_some_and(|step| step.status == StepStatus::Running);
922
923        loop {
924            if redelivering_running_step {
925                // A process can die after the step side effect succeeds but before
926                // StepCompleted is durable. Redeliver the same attempt so an
927                // idempotent step can recover that ambiguous boundary.
928                redelivering_running_step = false;
929            } else {
930                attempt = attempt.checked_add(1).ok_or_else(|| {
931                    FlowError::InvalidTransition(format!("step attempt overflowed for {step_id}"))
932                })?;
933                let started = self
934                    .record_event_at(
935                        run_id,
936                        expected_sequence,
937                        FlowEvent::StepStarted {
938                            step_id: step_id.clone(),
939                            attempt,
940                        },
941                    )
942                    .await?;
943                expected_sequence = started.sequence;
944            }
945
946            let history = self.store.list(run_id).await?;
947            let invocation = StepInvocation {
948                run_id: run_id.to_string(),
949                step_id: step_id.clone(),
950                step_name: step_name.clone(),
951                input: input.clone(),
952                history,
953            };
954
955            match self.runtime.run_step(invocation).await {
956                Ok(output) => {
957                    self.record_event_at(
958                        run_id,
959                        expected_sequence,
960                        FlowEvent::StepCompleted { step_id, output },
961                    )
962                    .await?;
963                    return Ok(());
964                }
965                Err(err) if attempt < max_attempts => {
966                    let retry_after = if retry.delay_ms > 0 {
967                        Some(Utc::now() + ChronoDuration::milliseconds(retry.delay_ms as i64))
968                    } else {
969                        None
970                    };
971                    let retrying = self
972                        .record_event_at(
973                            run_id,
974                            expected_sequence,
975                            FlowEvent::StepRetrying {
976                                step_id: step_id.clone(),
977                                attempt,
978                                error: err.to_string(),
979                                retry_after,
980                            },
981                        )
982                        .await?;
983                    expected_sequence = retrying.sequence;
984                    if retry_after.is_some() {
985                        return Ok(());
986                    }
987                }
988                Err(err) => {
989                    let error = err.to_string();
990                    let failed = self
991                        .record_event_at(
992                            run_id,
993                            expected_sequence,
994                            FlowEvent::StepFailed {
995                                step_id: step_id.clone(),
996                                attempt,
997                                error: error.clone(),
998                            },
999                        )
1000                        .await?;
1001                    if retry.on_exhausted == StepFailureAction::ContinueWorkflow {
1002                        return Ok(());
1003                    }
1004                    self.record_event_at(run_id, failed.sequence, FlowEvent::RunFailed { error })
1005                        .await?;
1006                    return Ok(());
1007                }
1008            }
1009        }
1010    }
1011
1012    async fn execute_step_batch(
1013        &self,
1014        run_id: &str,
1015        snapshot: &WorkflowRunSnapshot,
1016        steps: Vec<StepCommand>,
1017        now: DateTime<Utc>,
1018    ) -> Result<()> {
1019        let mut expected_sequence = snapshot.last_sequence;
1020
1021        // Make every sibling identity durable before any side effect starts.
1022        for step in &steps {
1023            if snapshot.steps.contains_key(&step.step_id) {
1024                continue;
1025            }
1026            let created = self
1027                .record_event_at(
1028                    run_id,
1029                    expected_sequence,
1030                    FlowEvent::StepCreated {
1031                        step_id: step.step_id.clone(),
1032                        step_name: step.step_name.clone(),
1033                        input: step.input.clone(),
1034                        retry: step.retry,
1035                    },
1036                )
1037                .await?;
1038            expected_sequence = created.sequence;
1039        }
1040
1041        let mut active = Vec::new();
1042        for step in steps {
1043            let existing = snapshot.steps.get(&step.step_id);
1044            let attempt = match existing.map(|existing| existing.status) {
1045                Some(StepStatus::Completed | StepStatus::Failed) => continue,
1046                Some(StepStatus::Running) => {
1047                    existing.map(|existing| existing.attempt).ok_or_else(|| {
1048                        FlowError::InvalidTransition(format!(
1049                            "running batch step {} has no projected attempt",
1050                            step.step_id
1051                        ))
1052                    })?
1053                }
1054                Some(StepStatus::Pending) => {
1055                    if existing
1056                        .and_then(|existing| existing.retry_after)
1057                        .is_some_and(|retry_after| retry_after > now)
1058                    {
1059                        continue;
1060                    }
1061                    let attempt = existing
1062                        .and_then(|existing| existing.attempt.checked_add(1))
1063                        .ok_or_else(|| {
1064                            FlowError::InvalidTransition(format!(
1065                                "step attempt overflowed for {}",
1066                                step.step_id
1067                            ))
1068                        })?;
1069                    let started = self
1070                        .record_event_at(
1071                            run_id,
1072                            expected_sequence,
1073                            FlowEvent::StepStarted {
1074                                step_id: step.step_id.clone(),
1075                                attempt,
1076                            },
1077                        )
1078                        .await?;
1079                    expected_sequence = started.sequence;
1080                    attempt
1081                }
1082                None => {
1083                    let attempt = 1;
1084                    let started = self
1085                        .record_event_at(
1086                            run_id,
1087                            expected_sequence,
1088                            FlowEvent::StepStarted {
1089                                step_id: step.step_id.clone(),
1090                                attempt,
1091                            },
1092                        )
1093                        .await?;
1094                    expected_sequence = started.sequence;
1095                    attempt
1096                }
1097            };
1098            active.push((step, attempt));
1099        }
1100
1101        while !active.is_empty() {
1102            let history = self.store.list(run_id).await?;
1103            let mut tasks = JoinSet::new();
1104            for (index, (step, _)) in active.iter().enumerate() {
1105                let runtime = Arc::clone(&self.runtime);
1106                let invocation = StepInvocation {
1107                    run_id: run_id.to_string(),
1108                    step_id: step.step_id.clone(),
1109                    step_name: step.step_name.clone(),
1110                    input: step.input.clone(),
1111                    history: history.clone(),
1112                };
1113                tasks.spawn(async move { (index, runtime.run_step(invocation).await) });
1114            }
1115            let mut observed_outcomes = vec![false; active.len()];
1116            let mut immediate_retries = Vec::new();
1117            while let Some(joined) = tasks.join_next().await {
1118                let (index, outcome) = joined.map_err(|error| {
1119                    FlowError::Runtime(format!(
1120                        "concurrent step task failed before returning an outcome: {error}"
1121                    ))
1122                })?;
1123                if index >= observed_outcomes.len() || observed_outcomes[index] {
1124                    return Err(FlowError::InvalidTransition(
1125                        "concurrent step batch returned an invalid outcome index".to_string(),
1126                    ));
1127                }
1128                observed_outcomes[index] = true;
1129                let (step, attempt) = &active[index];
1130                match outcome {
1131                    Ok(output) => {
1132                        let completed = self
1133                            .record_event_at(
1134                                run_id,
1135                                expected_sequence,
1136                                FlowEvent::StepCompleted {
1137                                    step_id: step.step_id.clone(),
1138                                    output,
1139                                },
1140                            )
1141                            .await?;
1142                        expected_sequence = completed.sequence;
1143                    }
1144                    Err(error) if *attempt < step.retry.max_attempts.max(1) => {
1145                        let error = error.to_string();
1146                        let retry_after = if step.retry.delay_ms > 0 {
1147                            Some(
1148                                Utc::now()
1149                                    + ChronoDuration::milliseconds(step.retry.delay_ms as i64),
1150                            )
1151                        } else {
1152                            None
1153                        };
1154                        let retrying = self
1155                            .record_event_at(
1156                                run_id,
1157                                expected_sequence,
1158                                FlowEvent::StepRetrying {
1159                                    step_id: step.step_id.clone(),
1160                                    attempt: *attempt,
1161                                    error,
1162                                    retry_after,
1163                                },
1164                            )
1165                            .await?;
1166                        expected_sequence = retrying.sequence;
1167                        if retry_after.is_none() {
1168                            immediate_retries.push((step.clone(), *attempt));
1169                        }
1170                    }
1171                    Err(error) => {
1172                        let error = error.to_string();
1173                        let failed = self
1174                            .record_event_at(
1175                                run_id,
1176                                expected_sequence,
1177                                FlowEvent::StepFailed {
1178                                    step_id: step.step_id.clone(),
1179                                    attempt: *attempt,
1180                                    error: error.clone(),
1181                                },
1182                            )
1183                            .await?;
1184                        expected_sequence = failed.sequence;
1185                        if step.retry.on_exhausted == StepFailureAction::FailRun {
1186                            self.record_event_at(
1187                                run_id,
1188                                expected_sequence,
1189                                FlowEvent::RunFailed { error },
1190                            )
1191                            .await?;
1192                            return Ok(());
1193                        }
1194                    }
1195                }
1196            }
1197            if let Some(index) = observed_outcomes.iter().position(|observed| !observed) {
1198                return Err(FlowError::Runtime(format!(
1199                    "concurrent step batch omitted outcome index {index}"
1200                )));
1201            }
1202            if immediate_retries.is_empty() {
1203                return Ok(());
1204            }
1205
1206            let mut next_active = Vec::with_capacity(immediate_retries.len());
1207            for (step, attempt) in immediate_retries {
1208                let attempt = attempt.checked_add(1).ok_or_else(|| {
1209                    FlowError::InvalidTransition(format!(
1210                        "step attempt overflowed for {}",
1211                        step.step_id
1212                    ))
1213                })?;
1214                let started = self
1215                    .record_event_at(
1216                        run_id,
1217                        expected_sequence,
1218                        FlowEvent::StepStarted {
1219                            step_id: step.step_id.clone(),
1220                            attempt,
1221                        },
1222                    )
1223                    .await?;
1224                expected_sequence = started.sequence;
1225                next_active.push((step, attempt));
1226            }
1227            active = next_active;
1228        }
1229
1230        Ok(())
1231    }
1232}
a3s_flow/engine/mod.rs

a3s_flow/engine/
mod.rs
