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    StepStatus, WaitStatus, WorkflowRunSnapshot, WorkflowRunStatus, WorkflowRunSummary,
9    WorkflowRunSuspension, WorkflowSpec,
10};
11use crate::observe::{FlowEventObserver, NoopFlowEventObserver};
12use crate::runtime::{FlowRuntime, WorkflowInvocation};
13use crate::store::{FlowEventStore, InMemoryEventStore};
14
15mod operations;
16mod steps;
17mod validation;
18use steps::StepExecutionContext;
19use validation::{
20    ensure_child_operation_matches, ensure_hook_command_matches, ensure_progress_matches,
21    ensure_same_start, ensure_step_batch_valid, ensure_step_command_matches,
22    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 = Vec::new();
341        for run_id in self.store.list_run_ids().await? {
342            let snapshot = self.snapshot(&run_id).await?;
343            if snapshot.status.is_terminal() {
344                continue;
345            }
346            for hook in snapshot.hooks.values() {
347                if hook.status == HookStatus::Active && hook.token == token {
348                    matches.push((run_id.clone(), hook.hook_id.clone()));
349                }
350            }
351        }
352
353        match matches.len() {
354            0 => Err(FlowError::HookTokenNotFound(token.to_string())),
355            1 => {
356                let (run_id, hook_id) = matches.remove(0);
357                self.resume_hook(&run_id, &hook_id, payload).await?;
358                Ok((run_id, hook_id))
359            }
360            _ => Err(FlowError::InvalidTransition(
361                "hook token is active in multiple runs (value redacted)".to_string(),
362            )),
363        }
364    }
365
366    /// Dispose an active hook by its external token.
367    ///
368    /// This mirrors [`resume_hook_by_token`](Self::resume_hook_by_token) for
369    /// callback routers that only know the public token.
370    pub async fn dispose_hook_by_token(&self, token: &str) -> Result<(String, String)> {
371        let mut matches = Vec::new();
372        for run_id in self.store.list_run_ids().await? {
373            let snapshot = self.snapshot(&run_id).await?;
374            if snapshot.status.is_terminal() {
375                continue;
376            }
377            for hook in snapshot.hooks.values() {
378                if hook.status == HookStatus::Active && hook.token == token {
379                    matches.push((run_id.clone(), hook.hook_id.clone()));
380                }
381            }
382        }
383
384        match matches.len() {
385            0 => Err(FlowError::HookTokenNotFound(token.to_string())),
386            1 => {
387                let (run_id, hook_id) = matches.remove(0);
388                self.dispose_hook(&run_id, &hook_id).await?;
389                Ok((run_id, hook_id))
390            }
391            _ => Err(FlowError::InvalidTransition(
392                "hook token is active in multiple runs (value redacted)".to_string(),
393            )),
394        }
395    }
396
397    /// List active waits whose `resume_at` is at or before `now`.
398    ///
399    /// Scheduler integrations can use this to inspect due timers before
400    /// deciding how aggressively to drive them.
401    pub async fn list_due_waits(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
402        let mut due = Vec::new();
403        for run_id in self.store.list_run_ids().await? {
404            let snapshot = self.snapshot(&run_id).await?;
405            if snapshot.status.is_terminal() {
406                continue;
407            }
408            for wait in snapshot.waits.values() {
409                if wait.status == WaitStatus::Waiting && wait.resume_at <= now {
410                    due.push((run_id.clone(), wait.wait_id.clone()));
411                }
412            }
413        }
414        due.sort();
415        Ok(due)
416    }
417
418    /// Complete every due wait and drive the affected workflows.
419    ///
420    /// Returns the `(run_id, wait_id)` pairs that were resumed. A wait already
421    /// completed by another caller is skipped by [`Self::resume_wait`].
422    pub async fn resume_due_waits(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
423        let due = self.list_due_waits(now).await?;
424        let mut resumed = Vec::with_capacity(due.len());
425        for (run_id, wait_id) in due {
426            self.resume_wait(&run_id, &wait_id).await?;
427            resumed.push((run_id, wait_id));
428        }
429        Ok(resumed)
430    }
431
432    /// List pending step retries whose `retry_after` is at or before `now`.
433    pub async fn list_due_retries(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
434        let mut due = Vec::new();
435        for run_id in self.store.list_run_ids().await? {
436            let snapshot = self.snapshot(&run_id).await?;
437            if snapshot.status.is_terminal() {
438                continue;
439            }
440            for (step_id, _) in snapshot.due_retries(now) {
441                due.push((run_id.clone(), step_id));
442            }
443        }
444        due.sort();
445        Ok(due)
446    }
447
448    /// Drive every run with a due step retry.
449    pub async fn resume_due_retries(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
450        let due = self.list_due_retries(now).await?;
451        let mut run_ids = Vec::new();
452        for (run_id, _) in &due {
453            if !run_ids.contains(run_id) {
454                run_ids.push(run_id.clone());
455            }
456        }
457        for run_id in run_ids {
458            self.drive_at(&run_id, now).await?;
459        }
460        Ok(due)
461    }
462
463    pub async fn snapshot(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
464        let history = self.store.list(run_id).await?;
465        project_run(run_id, &history)
466    }
467
468    pub async fn history(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>> {
469        self.store.list(run_id).await
470    }
471
472    pub async fn list_run_ids(&self) -> Result<Vec<String>> {
473        self.store.list_run_ids().await
474    }
475
476    pub async fn list_snapshots(&self) -> Result<Vec<WorkflowRunSnapshot>> {
477        let mut snapshots = Vec::new();
478        for run_id in self.store.list_run_ids().await? {
479            snapshots.push(self.snapshot(&run_id).await?);
480        }
481        Ok(snapshots)
482    }
483
484    /// Summarize run state across the active store.
485    ///
486    /// Suspension counters include only non-terminal runs, so a cancelled run
487    /// that still has an old wait or hook in history is not reported as
488    /// actionable work.
489    pub async fn run_summary(&self) -> Result<WorkflowRunSummary> {
490        let snapshots = self.list_snapshots().await?;
491        Ok(WorkflowRunSummary::from_snapshots(&snapshots))
492    }
493
494    /// List open waits, active hooks, and pending delayed retries.
495    ///
496    /// The `due` flag on wait and retry suspensions is computed against `now`.
497    /// Terminal runs are skipped so cancelled histories do not produce
498    /// actionable operator work.
499    pub async fn list_open_suspensions(
500        &self,
501        now: DateTime<Utc>,
502    ) -> Result<Vec<WorkflowRunSuspension>> {
503        let mut suspensions = Vec::new();
504        for run_id in self.store.list_run_ids().await? {
505            let snapshot = self.snapshot(&run_id).await?;
506            if snapshot.status.is_terminal() {
507                continue;
508            }
509            for wait in snapshot.waits.values() {
510                if wait.status == WaitStatus::Waiting {
511                    suspensions.push(WorkflowRunSuspension::Wait {
512                        run_id: run_id.clone(),
513                        wait: wait.clone(),
514                        due: wait.resume_at <= now,
515                    });
516                }
517            }
518            for hook in snapshot.hooks.values() {
519                if hook.status == HookStatus::Active {
520                    suspensions.push(WorkflowRunSuspension::Hook {
521                        run_id: run_id.clone(),
522                        hook: hook.clone(),
523                    });
524                }
525            }
526            for step in snapshot.steps.values() {
527                if step.status == StepStatus::Pending {
528                    if let Some(retry_after) = step.retry_after {
529                        suspensions.push(WorkflowRunSuspension::Retry {
530                            run_id: run_id.clone(),
531                            step: step.clone(),
532                            due: retry_after <= now,
533                        });
534                    }
535                }
536            }
537        }
538        suspensions.sort_by(|left, right| {
539            (left.run_id(), left.kind_order(), left.subject_id()).cmp(&(
540                right.run_id(),
541                right.kind_order(),
542                right.subject_id(),
543            ))
544        });
545        Ok(suspensions)
546    }
547
548    /// Return the earliest open wait or delayed retry across non-terminal runs.
549    ///
550    /// This is useful for hosts that want to sleep until the next scheduler tick
551    /// instead of polling at a fixed interval. Active hooks are intentionally
552    /// ignored because they do not have a scheduled wake-up time.
553    pub async fn next_wakeup(&self, now: DateTime<Utc>) -> Result<Option<WorkflowRunSuspension>> {
554        let mut wakeups = self.list_open_suspensions(now).await?;
555        wakeups.retain(|suspension| suspension.scheduled_at().is_some());
556        wakeups.sort_by(|left, right| {
557            (
558                left.scheduled_at(),
559                left.run_id(),
560                left.kind_order(),
561                left.subject_id(),
562            )
563                .cmp(&(
564                    right.scheduled_at(),
565                    right.run_id(),
566                    right.kind_order(),
567                    right.subject_id(),
568                ))
569        });
570        Ok(wakeups.into_iter().next())
571    }
572
573    /// List active external callback hooks across non-terminal runs.
574    ///
575    /// Callback routers and dashboards can use this to discover public hook
576    /// tokens and their audit metadata without projecting every run manually.
577    /// The result is sorted by run ID and hook ID for stable polling output.
578    pub async fn list_active_hooks(&self) -> Result<Vec<ActiveHookSnapshot>> {
579        let mut hooks = Vec::new();
580        for run_id in self.store.list_run_ids().await? {
581            let snapshot = self.snapshot(&run_id).await?;
582            if snapshot.status.is_terminal() {
583                continue;
584            }
585            for hook in snapshot.hooks.values() {
586                if hook.status == HookStatus::Active {
587                    hooks.push(ActiveHookSnapshot {
588                        run_id: run_id.clone(),
589                        hook: hook.clone(),
590                    });
591                }
592            }
593        }
594        hooks.sort_by(|left, right| {
595            (left.run_id.as_str(), left.hook.hook_id.as_str())
596                .cmp(&(right.run_id.as_str(), right.hook.hook_id.as_str()))
597        });
598        Ok(hooks)
599    }
600
601    /// Replay and dispatch until the run reaches a terminal state or an open
602    /// wait/hook suspension.
603    pub async fn drive(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
604        self.drive_at(run_id, Utc::now()).await
605    }
606
607    async fn drive_at(&self, run_id: &str, now: DateTime<Utc>) -> Result<WorkflowRunSnapshot> {
608        'replay: for _ in 0..self.max_replay_iterations {
609            let history = self.store.list(run_id).await?;
610            let snapshot = project_run(run_id, &history)?;
611            if snapshot.status.is_terminal()
612                || snapshot
613                    .waits
614                    .values()
615                    .any(|wait| wait.status == WaitStatus::Waiting)
616                || snapshot
617                    .hooks
618                    .values()
619                    .any(|hook| hook.status == HookStatus::Active)
620                || (snapshot.has_future_retry(now) && snapshot.due_retries(now).is_empty())
621            {
622                return Ok(snapshot);
623            }
624
625            let command = self
626                .runtime
627                .run_workflow(WorkflowInvocation {
628                    run_id: run_id.to_string(),
629                    spec: snapshot.spec.clone(),
630                    input: snapshot.input.clone(),
631                    history,
632                })
633                .await?;
634
635            match command {
636                RuntimeCommand::Complete { output } => {
637                    if snapshot.status == WorkflowRunStatus::Cancelling {
638                        return Err(FlowError::InvalidTransition(format!(
639                            "workflow run {run_id} completed after cancellation was requested; cleanup-aware cancellation must return cancel or fail"
640                        )));
641                    }
642                    match self
643                        .record_event_at(
644                            run_id,
645                            snapshot.last_sequence,
646                            FlowEvent::RunCompleted { output },
647                        )
648                        .await
649                    {
650                        Ok(_) => {}
651                        Err(err) if is_event_conflict(&err) => continue,
652                        Err(err) => return Err(err),
653                    }
654                    return self.snapshot(run_id).await;
655                }
656                RuntimeCommand::Fail { error } => {
657                    match self
658                        .record_event_at(
659                            run_id,
660                            snapshot.last_sequence,
661                            FlowEvent::RunFailed { error },
662                        )
663                        .await
664                    {
665                        Ok(_) => {}
666                        Err(err) if is_event_conflict(&err) => continue,
667                        Err(err) => return Err(err),
668                    }
669                    return self.snapshot(run_id).await;
670                }
671                RuntimeCommand::Cancel => {
672                    let cancellation = snapshot.cancellation.as_ref().ok_or_else(|| {
673                        FlowError::InvalidTransition(format!(
674                            "workflow run {run_id} returned cancel without a durable cancellation request"
675                        ))
676                    })?;
677                    match self
678                        .record_event_at(
679                            run_id,
680                            snapshot.last_sequence,
681                            FlowEvent::RunCancelled {
682                                reason: cancellation.request.reason.clone(),
683                            },
684                        )
685                        .await
686                    {
687                        Ok(_) => {}
688                        Err(err) if is_event_conflict(&err) => continue,
689                        Err(err) => return Err(err),
690                    }
691                    return self.snapshot(run_id).await;
692                }
693                RuntimeCommand::Timeout { deadline, reason } => {
694                    match self
695                        .record_event_at(
696                            run_id,
697                            snapshot.last_sequence,
698                            FlowEvent::RunTimedOut { deadline, reason },
699                        )
700                        .await
701                    {
702                        Ok(_) => {}
703                        Err(err) if is_event_conflict(&err) => continue,
704                        Err(err) => return Err(err),
705                    }
706                    return self.snapshot(run_id).await;
707                }
708                RuntimeCommand::RecordProgress { progress } => {
709                    progress.validate()?;
710                    if let Some(existing) = snapshot.progress(&progress.progress_id) {
711                        ensure_progress_matches(run_id, existing, &progress)?;
712                        return Err(FlowError::InvalidTransition(format!(
713                            "workflow rescheduled progress {} without progress",
714                            progress.progress_id
715                        )));
716                    }
717                    match self
718                        .record_event_at(
719                            run_id,
720                            snapshot.last_sequence,
721                            FlowEvent::RunProgressRecorded { progress },
722                        )
723                        .await
724                    {
725                        Ok(_) => {}
726                        Err(err) if is_event_conflict(&err) => continue,
727                        Err(err) => return Err(err),
728                    }
729                }
730                RuntimeCommand::LinkChildOperation { child } => {
731                    child.validate()?;
732                    if let Some(existing) = snapshot.child_operation(&child.reference_id) {
733                        ensure_child_operation_matches(run_id, existing, &child)?;
734                        return Err(FlowError::InvalidTransition(format!(
735                            "workflow rescheduled child operation {} without progress",
736                            child.reference_id
737                        )));
738                    }
739                    match self
740                        .record_event_at(
741                            run_id,
742                            snapshot.last_sequence,
743                            FlowEvent::ChildOperationLinked { child },
744                        )
745                        .await
746                    {
747                        Ok(_) => {}
748                        Err(err) if is_event_conflict(&err) => continue,
749                        Err(err) => return Err(err),
750                    }
751                }
752                RuntimeCommand::ScheduleStep {
753                    step_id,
754                    step_name,
755                    input,
756                    retry,
757                } => {
758                    if let Some(step) = snapshot.steps.get(&step_id) {
759                        ensure_step_command_matches(run_id, step, &step_name, &input, retry)?;
760                        if matches!(
761                            step.status,
762                            StepStatus::Completed | StepStatus::Failed | StepStatus::Cancelled
763                        ) {
764                            return Err(FlowError::InvalidTransition(format!(
765                                "workflow rescheduled terminal step {step_id} without progress"
766                            )));
767                        }
768                    }
769                    match self
770                        .execute_step(
771                            run_id,
772                            &snapshot,
773                            StepExecutionContext {
774                                step_id,
775                                step_name,
776                                input,
777                                retry,
778                                now,
779                            },
780                        )
781                        .await
782                    {
783                        Ok(()) => {}
784                        Err(err) if is_event_conflict(&err) => continue,
785                        Err(err) => return Err(err),
786                    }
787                }
788                RuntimeCommand::ScheduleSteps { steps } => {
789                    ensure_step_batch_valid(&steps)?;
790                    for step in &steps {
791                        if let Some(existing) = snapshot.steps.get(&step.step_id) {
792                            ensure_step_command_matches(
793                                run_id,
794                                existing,
795                                &step.step_name,
796                                &step.input,
797                                step.retry,
798                            )?;
799                        }
800                    }
801                    if steps.iter().all(|step| {
802                        snapshot.steps.get(&step.step_id).is_some_and(|existing| {
803                            matches!(
804                                existing.status,
805                                StepStatus::Completed | StepStatus::Failed | StepStatus::Cancelled
806                            )
807                        })
808                    }) {
809                        let step_ids = steps
810                            .iter()
811                            .map(|step| step.step_id.as_str())
812                            .collect::<Vec<_>>()
813                            .join(", ");
814                        return Err(FlowError::InvalidTransition(format!(
815                            "workflow rescheduled only terminal steps without progress: {step_ids}"
816                        )));
817                    }
818                    match self.execute_step_batch(run_id, &snapshot, steps, now).await {
819                        Ok(()) => {}
820                        Err(err) if is_event_conflict(&err) => continue 'replay,
821                        Err(err) => return Err(err),
822                    }
823                }
824                RuntimeCommand::WaitUntil { wait_id, resume_at } => {
825                    match snapshot.waits.get(&wait_id) {
826                        Some(wait) => {
827                            ensure_wait_command_matches(run_id, wait, resume_at)?;
828                            match wait.status {
829                                WaitStatus::Completed => continue,
830                                WaitStatus::Waiting => return self.snapshot(run_id).await,
831                                WaitStatus::Cancelled => {
832                                    return Err(FlowError::InvalidTransition(format!(
833                                        "workflow rescheduled cancelled wait {wait_id}; cancellation cleanup must use a distinct stable identity"
834                                    )))
835                                }
836                            }
837                        }
838                        None => {
839                            match self
840                                .record_event_at(
841                                    run_id,
842                                    snapshot.last_sequence,
843                                    FlowEvent::WaitCreated { wait_id, resume_at },
844                                )
845                                .await
846                            {
847                                Ok(_) => {}
848                                Err(err) if is_event_conflict(&err) => continue,
849                                Err(err) => return Err(err),
850                            }
851                            return self.snapshot(run_id).await;
852                        }
853                    }
854                }
855                RuntimeCommand::CreateHook {
856                    hook_id,
857                    token,
858                    metadata,
859                } => match snapshot.hooks.get(&hook_id) {
860                    Some(hook) => {
861                        ensure_hook_command_matches(run_id, hook, &token, &metadata)?;
862                        match hook.status {
863                            HookStatus::Received | HookStatus::Disposed => continue,
864                            HookStatus::Active => return self.snapshot(run_id).await,
865                            HookStatus::Cancelled => {
866                                return Err(FlowError::InvalidTransition(format!(
867                                    "workflow rescheduled cancelled hook {hook_id}; cancellation cleanup must use a distinct stable identity"
868                                )))
869                            }
870                        }
871                    }
872                    None => {
873                        self.ensure_hook_token_available(run_id, &hook_id, &token)
874                            .await?;
875                        match self
876                            .record_event_at(
877                                run_id,
878                                snapshot.last_sequence,
879                                FlowEvent::HookCreated {
880                                    hook_id,
881                                    token,
882                                    metadata,
883                                },
884                            )
885                            .await
886                        {
887                            Ok(_) => {}
888                            Err(err) if is_event_conflict(&err) => continue,
889                            Err(err) => return Err(err),
890                        }
891                        return self.snapshot(run_id).await;
892                    }
893                },
894            }
895        }
896
897        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
898    }
899
900    async fn terminate_run(&self, run_id: &str, event: FlowEvent) -> Result<()> {
901        for _ in 0..self.max_replay_iterations {
902            let snapshot = self.snapshot(run_id).await?;
903            if snapshot.status.is_terminal() {
904                return Ok(());
905            }
906            match self
907                .record_event_at(run_id, snapshot.last_sequence, event.clone())
908                .await
909            {
910                Ok(_) => return Ok(()),
911                Err(err) if is_event_conflict(&err) => continue,
912                Err(err) => return Err(err),
913            }
914        }
915        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
916    }
917
918    async fn record_event_at(
919        &self,
920        run_id: &str,
921        expected_sequence: u64,
922        event: FlowEvent,
923    ) -> Result<FlowEventEnvelope> {
924        let envelope = self
925            .store
926            .append_if_sequence(run_id, expected_sequence, event)
927            .await?;
928        self.observer.observe(envelope.clone()).await;
929        Ok(envelope)
930    }
931
932    async fn ensure_hook_token_available(
933        &self,
934        run_id: &str,
935        hook_id: &str,
936        token: &str,
937    ) -> Result<()> {
938        for existing_run_id in self.store.list_run_ids().await? {
939            let snapshot = self.snapshot(&existing_run_id).await?;
940            if snapshot.status.is_terminal() {
941                continue;
942            }
943            for hook in snapshot.hooks.values() {
944                if hook.status != HookStatus::Active || hook.token != token {
945                    continue;
946                }
947                if existing_run_id == run_id && hook.hook_id == hook_id {
948                    continue;
949                }
950                return Err(FlowError::HookTokenConflict {
951                    token: token.to_string(),
952                    existing_run_id,
953                    existing_hook_id: hook.hook_id.clone(),
954                });
955            }
956        }
957        Ok(())
958    }
959}
a3s_flow/engine/mod.rs

a3s_flow/engine/
mod.rs
