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

a3s_flow/engine/
mod.rs
