a3s_flow/engine/
mod.rs

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

a3s_flow/engine/
mod.rs
