a3s_flow/engine/
mod.rs

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

a3s_flow/engine/
mod.rs
