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

a3s_flow/engine/
mod.rs
