Skip to main content

roas_arazzo_executor/
run.rs

1//! The engine: what to send, what it meant, and what to do next.
2//!
3//! [`Run`] performs no IO. It hands out a request, is handed a response,
4//! and decides where that leaves the workflow — which is what lets one
5//! engine serve a blocking caller, an async one, and a test with no
6//! network at all.
7
8use crate::criterion;
9use crate::expression::{self, Exchange, ExpressionError, Scope, StepState, WorkflowState};
10use crate::http::{HttpRequest, HttpResponse};
11use crate::operation::{self, Source};
12use crate::report::{
13    CriterionOutcome, ExecutionError, ExecutionReport, Outcome, Performed, StepRecord,
14};
15use crate::select;
16use crate::select::SelectError;
17use roas_arazzo::v1_1::{
18    Criterion, CriterionKind, CriterionType, Description, FailureActionType, Parameter,
19    ParameterLocation, ReusableOr, SourceType, Step, SuccessActionType, ValueOrSelector, Workflow,
20};
21use serde_json::{Map, Value};
22use std::collections::{BTreeMap, BTreeSet};
23use std::time::{Duration, Instant};
24
25/// How far a run can go before it is treated as looping.
26#[derive(Clone, Copy, Debug)]
27struct Limits {
28    steps: usize,
29    depth: usize,
30    retries: u32,
31}
32
33impl Default for Limits {
34    fn default() -> Self {
35        Self {
36            steps: 1_000,
37            depth: 8,
38            retries: 10,
39        }
40    }
41}
42
43/// Everything a run needs besides the description itself.
44///
45/// Built by chaining: `Options::new().workflow("buyPet").input("petId", 7)`.
46#[derive(Clone, Debug, Default)]
47pub struct Options {
48    workflow: Option<String>,
49    inputs: Map<String, Value>,
50    sources: BTreeMap<String, Source>,
51    base_urls: BTreeMap<String, String>,
52    headers: Vec<(String, String)>,
53    limits: Limits,
54}
55
56impl Options {
57    /// Options with nothing set: the first workflow, no inputs, no
58    /// source documents.
59    #[must_use]
60    pub fn new() -> Self {
61        Self::default()
62    }
63
64    /// Run this workflow rather than the description's first.
65    #[must_use]
66    pub fn workflow(mut self, workflow_id: impl Into<String>) -> Self {
67        self.workflow = Some(workflow_id.into());
68        self
69    }
70
71    /// Set one workflow input.
72    #[must_use]
73    pub fn input(mut self, name: impl Into<String>, value: impl Into<Value>) -> Self {
74        self.inputs.insert(name.into(), value.into());
75        self
76    }
77
78    /// Set every input at once, from a JSON object.
79    #[must_use]
80    pub fn inputs(mut self, inputs: Value) -> Self {
81        if let Value::Object(inputs) = inputs {
82            self.inputs = inputs;
83        }
84        self
85    }
86
87    /// Supply a source description: the `name` it was declared with, the
88    /// `url` it was declared with, and the parsed document.
89    ///
90    /// Fetching documents is IO, which this crate leaves to its caller —
91    /// `roas-file-fetcher` and `roas-http-fetcher` do it for the loader
92    /// and do it here just as well.
93    #[must_use]
94    pub fn source(
95        mut self,
96        name: impl Into<String>,
97        url: impl Into<String>,
98        document: Value,
99    ) -> Self {
100        self.sources.insert(
101            name.into(),
102            Source {
103                url: url.into(),
104                document,
105            },
106        );
107        self
108    }
109
110    /// Send this source description's requests somewhere else — a test
111    /// server, a staging host — whatever its document says.
112    #[must_use]
113    pub fn base_url(mut self, source_name: impl Into<String>, url: impl Into<String>) -> Self {
114        self.base_urls.insert(source_name.into(), url.into());
115        self
116    }
117
118    /// Add a header to every request a step does not set itself.
119    #[must_use]
120    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
121        self.headers.push((name.into(), value.into()));
122        self
123    }
124
125    /// The most steps a run may take before it is called a loop.
126    #[must_use]
127    pub fn max_steps(mut self, steps: usize) -> Self {
128        self.limits.steps = steps;
129        self
130    }
131
132    /// The deepest a workflow may call another.
133    #[must_use]
134    pub fn max_depth(mut self, depth: usize) -> Self {
135        self.limits.depth = depth;
136        self
137    }
138
139    /// The most times one step may be retried.
140    #[must_use]
141    pub fn max_retries(mut self, retries: u32) -> Self {
142        self.limits.retries = retries;
143        self
144    }
145}
146
147/// What the engine wants next.
148///
149/// Deliberately not `#[non_exhaustive]`: a driving loop must handle
150/// every variant, and a new one would have to break that loop to mean
151/// anything — so a catch-all arm would hide the very change it was
152/// there to absorb.
153#[derive(Debug)]
154pub enum Progress {
155    /// Perform this request, then hand the response to
156    /// [`Run::supply`].
157    Send(HttpRequest),
158    /// Wait this long — a retry asked for it — then carry on.
159    Wait(Duration),
160    /// The run is over.
161    Done(Box<ExecutionReport>),
162}
163
164/// Everything a runtime expression can name at this point in the run.
165fn scope<'s>(
166    frame: &'s Frame<'_>,
167    steps: &'s BTreeMap<String, StepState>,
168    here: Option<&'s Exchange>,
169    finished: &'s BTreeMap<String, WorkflowState>,
170    ambient: &'s Ambient,
171) -> Scope<'s> {
172    Scope {
173        inputs: &frame.inputs,
174        outputs: &frame.outputs,
175        steps,
176        workflows: finished,
177        here,
178        sources: &ambient.sources,
179        components: &ambient.components,
180        self_: ambient.self_.as_deref(),
181        declared_steps: &frame.declared,
182        declared_workflows: &ambient.workflows,
183    }
184}
185
186/// What the calling step does when the workflow it started finishes.
187#[derive(Clone, Copy, Debug, PartialEq, Eq)]
188enum Then {
189    /// Finish the calling step, as any other step finishes.
190    Advance,
191    /// Try the calling step again — a `retry` that named a workflow.
192    Retry,
193    /// End the workflow that left: a `goto` does not come back.
194    EndCaller,
195}
196
197/// One workflow in progress.
198struct Frame<'d> {
199    workflow: &'d Workflow,
200    inputs: Value,
201    /// Step indices in the order `dependsOn` puts them.
202    order: Vec<usize>,
203    at: usize,
204    steps: BTreeMap<String, StepState>,
205    outputs: BTreeMap<String, Value>,
206    /// How many times each step has been attempted, for the report and
207    /// for the caller's safety limit.
208    attempts: BTreeMap<String, u32>,
209    /// Every step id this workflow declares.
210    declared: BTreeSet<String>,
211    /// How many retries each *failure action* has spent on a step.
212    /// `retryLimit` belongs to the action that states it, so a step
213    /// that fails two different ways gets both actions' budgets.
214    retries: BTreeMap<(String, usize), u32>,
215    /// The step of the calling frame that is waiting for this one, and
216    /// what it does when this one is done.
217    caller: Option<(String, Then)>,
218    /// When this frame started, for a calling step's `timeout`.
219    started: Instant,
220    /// Where to come back to when a retry sent the run to another step
221    /// first: "context transfers back upon completion".
222    detour: Option<usize>,
223    outcome: Outcome,
224}
225
226/// What a step's completion needs beyond what the step itself says.
227struct Completion {
228    /// The exchange, for a step that sent a request.
229    exchange: Option<Exchange>,
230    /// Outputs the step already has — a called workflow's, which its
231    /// own `outputs` may then add to or override.
232    given: BTreeMap<String, Value>,
233    /// Whether it counts as passed when the step states no criteria.
234    default_pass: bool,
235    attempt: u32,
236    performed: Performed,
237    elapsed: Duration,
238}
239
240/// The step waiting for a response.
241struct Pending {
242    step: usize,
243    attempt: u32,
244    exchange: Exchange,
245    started: Instant,
246}
247
248/// The parts of the description every expression can see, whichever
249/// workflow is running.
250struct Ambient {
251    /// `sourceDescriptions` as JSON, for `$sourceDescriptions.…`.
252    sources: Value,
253    /// `components` as JSON, for `$components.…`.
254    components: Value,
255    /// The description's `$self`, for `$self`.
256    self_: Option<String>,
257    /// Every workflow id the description declares.
258    workflows: BTreeSet<String>,
259}
260
261/// A workflow run, one request at a time.
262pub struct Run<'d> {
263    description: &'d Description,
264    options: &'d Options,
265    ambient: Ambient,
266    frames: Vec<Frame<'d>>,
267    /// Workflows still to run — dependencies first, then the one asked
268    /// for.
269    queue: Vec<&'d Workflow>,
270    finished: BTreeMap<String, WorkflowState>,
271    pending: Option<Pending>,
272    wait: Option<Duration>,
273    records: Vec<StepRecord>,
274    /// Source descriptions the run was not given, and which could hold
275    /// an operation — a bare `operationId` cannot be shown to be unique
276    /// while one of these is missing.
277    unsupplied: Vec<String>,
278    taken: usize,
279    report: Option<Box<ExecutionReport>>,
280}
281
282impl<'d> Run<'d> {
283    /// Prepare a run: pick the workflow, order what it depends on, and
284    /// stop before anything is sent.
285    ///
286    /// # Errors
287    ///
288    /// [`ExecutionError::UnknownWorkflow`] or
289    /// [`ExecutionError::Circular`] when the description does not
290    /// describe a runnable order.
291    pub fn start(
292        description: &'d Description,
293        options: &'d Options,
294    ) -> Result<Self, ExecutionError> {
295        let wanted = match &options.workflow {
296            Some(id) => description
297                .workflows
298                .iter()
299                .find(|workflow| &workflow.workflow_id == id)
300                .ok_or_else(|| ExecutionError::UnknownWorkflow(id.clone()))?,
301            None => description
302                .workflows
303                .first()
304                .ok_or_else(|| ExecutionError::UnknownWorkflow(String::new()))?,
305        };
306
307        let sources = serde_json::to_value(
308            description
309                .source_descriptions
310                .iter()
311                .map(|source| (source.name.clone(), source))
312                .collect::<BTreeMap<_, _>>(),
313        )
314        .unwrap_or(Value::Null);
315        let components = description
316            .components
317            .as_ref()
318            .and_then(|components| serde_json::to_value(components).ok())
319            .unwrap_or(Value::Null);
320
321        let mut queue = ordered_workflows(description, wanted)?;
322        let first = queue.remove(0);
323        let mut run = Self {
324            description,
325            options,
326            ambient: Ambient {
327                sources,
328                components,
329                self_: description.self_.clone(),
330                workflows: description
331                    .workflows
332                    .iter()
333                    .map(|workflow| workflow.workflow_id.clone())
334                    .collect(),
335            },
336            frames: Vec::new(),
337            queue,
338            finished: BTreeMap::new(),
339            pending: None,
340            wait: None,
341            records: Vec::new(),
342            unsupplied: description
343                .source_descriptions
344                .iter()
345                // An Arazzo document holds workflows, not operations, so
346                // its absence cannot make an operation ambiguous.
347                .filter(|source| source.type_ != Some(SourceType::Arazzo))
348                .filter(|source| !options.sources.contains_key(&source.name))
349                .map(|source| source.name.clone())
350                .collect(),
351            taken: 0,
352            report: None,
353        };
354        run.enter(first, Value::Object(options.inputs.clone()), None)?;
355        Ok(run)
356    }
357
358    /// Advance until something is needed from the caller.
359    ///
360    /// Not called `next`: a run is not an iterator, because what comes
361    /// out of it has to be answered with [`Run::supply`] before there is
362    /// anything more to come.
363    ///
364    /// # Errors
365    ///
366    /// Whatever stopped the run — see [`ExecutionError`].
367    pub fn advance(&mut self) -> Result<Progress, ExecutionError> {
368        if let Some(wait) = self.wait.take() {
369            return Ok(Progress::Wait(wait));
370        }
371        // One request is outstanding at a time. Handing out another
372        // would send it twice and lose the exchange the first one is
373        // waiting to be judged by.
374        if let Some(pending) = &self.pending {
375            return Err(ExecutionError::Awaiting {
376                method: pending.exchange.request.method.clone(),
377                url: pending.exchange.request.url.clone(),
378            });
379        }
380        loop {
381            if let Some(report) = self.report.take() {
382                return Ok(Progress::Done(report));
383            }
384            let Some(frame) = self.frames.last() else {
385                return Ok(Progress::Done(Box::new(self.finish())));
386            };
387            // The frame is spent: name its outputs and hand them back.
388            if frame.at >= frame.order.len() {
389                self.leave()?;
390                continue;
391            }
392
393            self.taken += 1;
394            if self.taken > self.options.limits.steps {
395                return Err(ExecutionError::Limit {
396                    limit: "step",
397                    at: self.options.limits.steps,
398                });
399            }
400
401            let index = frame.order[frame.at];
402            let step = &frame.workflow.steps[index];
403            if step.workflow_id.is_some() {
404                self.call(index)?;
405                continue;
406            }
407
408            let (request, exchange) = self.build(index)?;
409            let attempt = self
410                .frames
411                .last()
412                .and_then(|frame| frame.attempts.get(&step.step_id).copied())
413                .unwrap_or(0)
414                + 1;
415            self.pending = Some(Pending {
416                step: index,
417                attempt,
418                exchange,
419                started: Instant::now(),
420            });
421            return Ok(Progress::Send(request));
422        }
423    }
424
425    /// Hand back the response to the request [`Run::advance`] asked for.
426    ///
427    /// # Errors
428    ///
429    /// Whatever the response made impossible — a criterion that cannot
430    /// be decided, an output that names nothing, a `goto` with no
431    /// target.
432    pub fn supply(&mut self, response: HttpResponse) -> Result<(), ExecutionError> {
433        let Some(mut pending) = self.pending.take() else {
434            return Err(ExecutionError::NotWaiting);
435        };
436        let elapsed = pending.started.elapsed();
437        let status = response.status;
438        pending.exchange.response_body = response.body_as_json();
439        pending.exchange.response = Some(response);
440
441        if self.frames.last().is_none() {
442            return Err(ExecutionError::NotWaiting);
443        }
444        let performed = Performed::Request {
445            method: pending.exchange.request.method.clone(),
446            url: pending.exchange.request.url.clone(),
447            status,
448        };
449        // No criteria means the status is the whole judgement.
450        self.complete(
451            pending.step,
452            Completion {
453                exchange: Some(pending.exchange),
454                given: BTreeMap::new(),
455                default_pass: (200..400).contains(&status),
456                attempt: pending.attempt,
457                performed,
458                elapsed,
459            },
460        )
461    }
462
463    // ---- the steps of a run -----------------------------------------
464
465    /// Everything a step's completion needs that the step itself does
466    /// not say.
467    ///
468    /// A step ends the same way whether it sent a request or called a
469    /// workflow: its criteria are judged, its outputs are named, and
470    /// its actions decide where the workflow goes next.
471    fn complete(&mut self, index: usize, done: Completion) -> Result<(), ExecutionError> {
472        let frame = self.frames.last().expect("a frame to complete in");
473        let step = &frame.workflow.steps[index];
474        let step_id = step.step_id.clone();
475        let workflow_id = frame.workflow.workflow_id.clone();
476
477        // What the step produced is in scope while its own outputs are
478        // named — that is how a workflow step reads what it called.
479        let mut state = StepState {
480            exchange: done.exchange.clone(),
481            outputs: done.given.clone(),
482            passed: true,
483        };
484        let (passed, criteria, outputs) = {
485            let mut ahead = frame.steps.clone();
486            ahead.insert(step_id.clone(), state.clone());
487            let scope = scope(
488                frame,
489                &ahead,
490                done.exchange.as_ref(),
491                &self.finished,
492                &self.ambient,
493            );
494
495            let mut criteria = Vec::with_capacity(step.success_criteria.len());
496            // Criteria, where a step states them, are the whole
497            // judgement: a step that says `$statusCode == 404` means it.
498            let mut passed = if step.success_criteria.is_empty() {
499                done.default_pass
500            } else {
501                true
502            };
503            for criterion in &step.success_criteria {
504                let holds = criterion::passes(criterion, &scope)?;
505                criteria.push(CriterionOutcome {
506                    condition: criterion.condition.clone(),
507                    passed: holds,
508                });
509                passed = passed && holds;
510            }
511            // Only a step that did what it said can name what it
512            // produced: a failed one is about to be retried or given up
513            // on, and its outputs would name what is not there.
514            let outputs = if passed {
515                let mut outputs = done.given.clone();
516                outputs.extend(evaluate_outputs(&step.outputs, &scope)?);
517                outputs
518            } else {
519                // What it was handed only seeded the scope above: a
520                // step that failed names nothing, a workflow step
521                // included, so no recovery step reads a token from a
522                // call that went wrong.
523                BTreeMap::new()
524            };
525            (passed, criteria, outputs)
526        };
527        state.outputs = outputs.clone();
528        state.passed = passed;
529
530        // The step is in scope before its actions are chosen: an
531        // `onSuccess` criterion reading `$steps.<this step>.outputs` is
532        // asking about the step that just finished.
533        let frame = self.frames.last_mut().expect("the frame is still there");
534        frame.steps.insert(step_id.clone(), state);
535
536        let action = self.decide(index, passed, done.exchange.as_ref())?;
537        let described = describe(&action);
538
539        self.records.push(StepRecord {
540            workflow_id,
541            step_id,
542            attempt: done.attempt,
543            performed: done.performed,
544            criteria,
545            passed,
546            outputs,
547            action: described,
548            elapsed: done.elapsed,
549        });
550        self.apply(action)
551    }
552
553    /// Push a frame for `workflow`.
554    fn enter(
555        &mut self,
556        workflow: &'d Workflow,
557        inputs: Value,
558        caller: Option<(String, Then)>,
559    ) -> Result<(), ExecutionError> {
560        if self.frames.len() >= self.options.limits.depth {
561            return Err(ExecutionError::Limit {
562                limit: "workflow depth",
563                at: self.options.limits.depth,
564            });
565        }
566        self.frames.push(Frame {
567            workflow,
568            inputs,
569            order: ordered_steps(workflow, self.description)?,
570            at: 0,
571            steps: BTreeMap::new(),
572            outputs: BTreeMap::new(),
573            declared: workflow
574                .steps
575                .iter()
576                .map(|step| step.step_id.clone())
577                .collect(),
578            attempts: BTreeMap::new(),
579            retries: BTreeMap::new(),
580            caller,
581            started: Instant::now(),
582            detour: None,
583            outcome: Outcome::Succeeded,
584        });
585        Ok(())
586    }
587
588    /// Finish the top frame: name its outputs and give them to whoever
589    /// is waiting.
590    fn leave(&mut self) -> Result<(), ExecutionError> {
591        let frame = self.frames.pop().expect("a frame to leave");
592        let outputs = {
593            let scope = scope(&frame, &frame.steps, None, &self.finished, &self.ambient);
594            if frame.outcome == Outcome::Succeeded {
595                evaluate_outputs(&frame.workflow.outputs, &scope)?
596            } else {
597                // A workflow that stopped early names outputs from steps
598                // that never ran. Those go with the steps; anything else
599                // wrong with an output is still worth saying.
600                evaluate_what_ran(&frame.workflow.outputs, &scope)?
601            }
602        };
603        self.finished.insert(
604            frame.workflow.workflow_id.clone(),
605            WorkflowState {
606                inputs: frame.inputs.clone(),
607                outputs: outputs.clone(),
608            },
609        );
610
611        let Some((step_id, then)) = frame.caller else {
612            // A root workflow: its outputs are the run's, unless it was
613            // only a dependency of the one that was asked for.
614            if self.queue.is_empty() {
615                self.report = Some(Box::new(ExecutionReport {
616                    workflow_id: frame.workflow.workflow_id.clone(),
617                    outcome: frame.outcome,
618                    outputs,
619                    steps: std::mem::take(&mut self.records),
620                }));
621            } else {
622                let next = self.queue.remove(0);
623                let inputs = Value::Object(self.options.inputs.clone());
624                self.enter(next, inputs, None)?;
625            }
626            return Ok(());
627        };
628        let Some(parent) = self.frames.last() else {
629            return Ok(());
630        };
631        let index = parent.order[parent.at];
632        debug_assert_eq!(parent.workflow.steps[index].step_id, step_id);
633
634        match then {
635            // A `goto` handed the workflow over: what it did is what the
636            // workflow that left it did, and there is nothing to come
637            // back to.
638            Then::EndCaller => {
639                let parent = self.frames.last_mut().expect("the parent is still there");
640                parent.steps.insert(
641                    step_id,
642                    StepState {
643                        exchange: None,
644                        outputs,
645                        passed: frame.outcome != Outcome::Failed,
646                    },
647                );
648                parent.at = parent.order.len();
649                parent.outcome = frame.outcome;
650                Ok(())
651            }
652            // A `retry` sent the run through another workflow first;
653            // now the step that failed is tried again.
654            Then::Retry => Ok(()),
655            // An ordinary workflow step: it ends like any other step,
656            // with its own criteria, outputs and actions.
657            Then::Advance => {
658                let elapsed = frame.started.elapsed();
659                let timed_out = parent.workflow.steps[index]
660                    .timeout
661                    .and_then(|timeout| u64::try_from(timeout).ok())
662                    .is_some_and(|timeout| elapsed > Duration::from_millis(timeout));
663                let attempt = self
664                    .frames
665                    .last()
666                    .and_then(|parent| parent.attempts.get(&step_id).copied())
667                    .unwrap_or(0)
668                    + 1;
669                self.complete(
670                    index,
671                    Completion {
672                        exchange: None,
673                        given: outputs,
674                        default_pass: frame.outcome != Outcome::Failed && !timed_out,
675                        attempt,
676                        performed: Performed::Workflow {
677                            workflow_id: frame.workflow.workflow_id.clone(),
678                            outcome: frame.outcome,
679                        },
680                        elapsed,
681                    },
682                )
683            }
684        }
685    }
686
687    /// A step that calls a workflow.
688    fn call(&mut self, index: usize) -> Result<(), ExecutionError> {
689        let frame = self.frames.last().expect("a frame to call from");
690        let step = &frame.workflow.steps[index];
691        let step_id = step.step_id.clone();
692        let wanted = step.workflow_id.clone().unwrap_or_default();
693        if wanted.starts_with('$') {
694            return Err(ExecutionError::Unsupported(format!(
695                "step `{step_id}` calls `{wanted}`, and this executor runs only workflows of the description it was given"
696            )));
697        }
698        let workflow = self
699            .description
700            .workflows
701            .iter()
702            .find(|workflow| workflow.workflow_id == wanted)
703            .ok_or_else(|| ExecutionError::UnknownWorkflow(wanted.clone()))?;
704
705        // A workflow step's parameters are the workflow's inputs.
706        // "When the step... specifies a `workflowId`, then all
707        // parameters map to workflow inputs", and a workflow's own
708        // parameters are "applicable for all steps described under this
709        // workflow... can be overridden at the step level but cannot be
710        // removed there" — so both lists go, the step's last.
711        let mut arguments = frame.workflow.parameters.clone();
712        arguments.extend(step.parameters.iter().cloned());
713        let inputs = self.arguments(&arguments)?;
714        self.enter(workflow, inputs, Some((step_id, Then::Advance)))
715    }
716
717    /// The inputs a called workflow starts with: the caller's own,
718    /// with whatever parameters were passed to it written over them.
719    fn arguments(&self, arguments: &[ReusableOr<Parameter>]) -> Result<Value, ExecutionError> {
720        let frame = self.frames.last().expect("a frame to pass arguments from");
721        let scope = scope(frame, &frame.steps, None, &self.finished, &self.ambient);
722        // A called workflow starts with what it was passed and nothing
723        // else: only the parameters are forwarded, not the caller's
724        // whole input context, so a child reading `$inputs.x` is asking
725        // for something the calling step gave it.
726        let mut inputs = Map::new();
727        for parameter in parameters(arguments, self.description, &scope)? {
728            inputs.insert(parameter.name, parameter.value);
729        }
730        Ok(Value::Object(inputs))
731    }
732
733    /// Assemble the request a step wants sent.
734    fn build(&self, index: usize) -> Result<(HttpRequest, Exchange), ExecutionError> {
735        let frame = self.frames.last().expect("a frame to build in");
736        let step = &frame.workflow.steps[index];
737        let endpoint = operation::resolve(
738            step,
739            &self.options.sources,
740            &self.options.base_urls,
741            &self.unsupplied,
742        )?;
743        let scope = scope(frame, &frame.steps, None, &self.finished, &self.ambient);
744
745        // The workflow's parameters first, so a step's own override them.
746        let mut resolved = parameters(&frame.workflow.parameters, self.description, &scope)?;
747        for parameter in parameters(&step.parameters, self.description, &scope)? {
748            resolved.retain(|existing| {
749                !(existing.name == parameter.name && existing.location == parameter.location)
750            });
751            resolved.push(parameter);
752        }
753
754        let mut path = BTreeMap::new();
755        let mut query = BTreeMap::new();
756        let mut headers: Vec<(String, String)> = Vec::new();
757        let mut cookies = Vec::new();
758        let mut querystring = None;
759        for parameter in resolved {
760            match parameter.location {
761                ParameterLocation::Path => {
762                    path.insert(parameter.name, parameter.value);
763                }
764                ParameterLocation::Query => {
765                    query.insert(parameter.name, parameter.value);
766                }
767                ParameterLocation::Querystring => {
768                    querystring = Some(text(&parameter.value));
769                }
770                ParameterLocation::Header => {
771                    headers.push((parameter.name, text(&parameter.value)));
772                }
773                ParameterLocation::Cookie => {
774                    cookies.push(format!("{}={}", parameter.name, text(&parameter.value)));
775                }
776                ParameterLocation::Channel => {
777                    return Err(ExecutionError::Unsupported(format!(
778                        "step `{}` has a `channel` parameter, which belongs to an AsyncAPI step",
779                        step.step_id
780                    )));
781                }
782            }
783        }
784        if !cookies.is_empty() {
785            headers.push(("Cookie".to_owned(), cookies.join("; ")));
786        }
787        for (name, value) in &self.options.headers {
788            if !headers
789                .iter()
790                .any(|(existing, _)| existing.eq_ignore_ascii_case(name))
791            {
792                headers.push((name.clone(), value.clone()));
793            }
794        }
795
796        let body = body(step, &scope)?;
797        if let Some(body) = &body
798            && !headers
799                .iter()
800                .any(|(name, _)| name.eq_ignore_ascii_case("content-type"))
801        {
802            headers.push(("Content-Type".to_owned(), body.content_type.clone()));
803        }
804
805        let url = url(&endpoint, &path, &query, querystring.as_deref()).map_err(|reason| {
806            ExecutionError::BadRequest {
807                step: step.step_id.clone(),
808                reason,
809            }
810        })?;
811        let request = HttpRequest {
812            method: endpoint.method,
813            url,
814            headers,
815            body: body.as_ref().map(|body| body.bytes.clone()),
816            timeout: step
817                .timeout
818                .and_then(|timeout| u64::try_from(timeout).ok())
819                .map(Duration::from_millis),
820        };
821        Ok((
822            request.clone(),
823            Exchange {
824                request,
825                path,
826                query,
827                body: body.map(|body| body.value),
828                response: None,
829                response_body: None,
830            },
831        ))
832    }
833
834    /// Which action a step's outcome calls for.
835    ///
836    /// The exchange is in scope: `$statusCode == 503` is how a failure
837    /// action says which failure it is about.
838    fn decide(
839        &mut self,
840        index: usize,
841        passed: bool,
842        exchange: Option<&Exchange>,
843    ) -> Result<Action, ExecutionError> {
844        let frame = self.frames.last().expect("a frame to decide in");
845        let step = &frame.workflow.steps[index];
846        let scope = scope(frame, &frame.steps, exchange, &self.finished, &self.ambient);
847
848        if passed {
849            // A step's own actions first, then the workflow's.
850            let actions = step
851                .on_success
852                .iter()
853                .chain(frame.workflow.success_actions.iter());
854            for action in actions {
855                let action = success_action(action, self.description)?;
856                if !holds(&action.criteria, &scope)? {
857                    continue;
858                }
859                return Ok(match action.type_ {
860                    SuccessActionType::End => Action::End(Outcome::Ended),
861                    SuccessActionType::Goto => Action::Goto {
862                        step: action.step_id.clone(),
863                        workflow: action.workflow_id.clone(),
864                        parameters: action.parameters.clone(),
865                    },
866                });
867            }
868            return Ok(Action::Advance);
869        }
870
871        let actions = step
872            .on_failure
873            .iter()
874            .chain(frame.workflow.failure_actions.iter());
875        for (at, action) in actions.enumerate() {
876            let action = failure_action(action, self.description)?;
877            if !holds(&action.criteria, &scope)? {
878                continue;
879            }
880            return Ok(match action.type_ {
881                FailureActionType::End => Action::End(Outcome::Failed),
882                FailureActionType::Goto => Action::Goto {
883                    step: action.step_id.clone(),
884                    workflow: action.workflow_id.clone(),
885                    parameters: action.parameters.clone(),
886                },
887                FailureActionType::Retry => {
888                    // "A non-negative integer indicating how many
889                    // attempts to retry the step MAY be attempted... If
890                    // not specified then a single retry SHALL be
891                    // attempted", and "The retryLimit MUST be exhausted
892                    // prior to executing subsequent failure actions" —
893                    // so an exhausted retry gives way to whatever the
894                    // description says next rather than ending here.
895                    let allowed = action
896                        .retry_limit
897                        .map_or(1, |limit| u32::try_from(limit).unwrap_or(u32::MAX));
898                    let spent = frame
899                        .retries
900                        .get(&(step.step_id.clone(), at))
901                        .copied()
902                        .unwrap_or(0);
903                    // The caller's cap is a rail against a description
904                    // that would retry forever, counted over the step;
905                    // the action's own limit is what the description
906                    // asked for.
907                    let taken = frame.attempts.get(&step.step_id).copied().unwrap_or(0);
908                    if spent >= allowed || taken >= self.options.limits.retries {
909                        continue;
910                    }
911                    Action::Retry {
912                        at,
913                        after: action.retry_after,
914                        step: action.step_id.clone(),
915                        workflow: action.workflow_id.clone(),
916                        parameters: action.parameters.clone(),
917                    }
918                }
919            });
920        }
921        // Nothing said what to do about a failure, so the workflow stops
922        // where it is.
923        Ok(Action::End(Outcome::Failed))
924    }
925
926    /// Carry an action out.
927    fn apply(&mut self, action: Action) -> Result<(), ExecutionError> {
928        let frame = self.frames.last_mut().expect("a frame to act in");
929        match action {
930            Action::Advance => {
931                // A step run as a retry's detour hands control back to
932                // the step that asked for it, which is tried again.
933                match frame.detour.take() {
934                    Some(back) => frame.at = back,
935                    None => frame.at += 1,
936                }
937                Ok(())
938            }
939            Action::End(outcome) => {
940                frame.outcome = outcome;
941                frame.at = frame.order.len();
942                // Nothing is forced on the caller here: a workflow that
943                // ends failed comes back through the step that called
944                // it, whose own `onFailure` may yet have something to
945                // say about it.
946                Ok(())
947            }
948            Action::Retry {
949                at,
950                after,
951                step: target,
952                workflow,
953                parameters: arguments,
954            } => {
955                // `decide` has already refused a retry whose limit is
956                // used up, so reaching here means another attempt is
957                // owed. Count it before anything else.
958                let index = frame.order[frame.at];
959                let step_id = frame.workflow.steps[index].step_id.clone();
960                *frame.attempts.entry(step_id.clone()).or_insert(0) += 1;
961                *frame.retries.entry((step_id.clone(), at)).or_insert(0) += 1;
962                if let Some(after) = after.filter(|after| *after > 0.0) {
963                    self.wait = Some(Duration::from_secs_f64(after));
964                }
965                match (target, workflow) {
966                    // "When used with `retry`, context transfers back
967                    // upon completion of the specified step" — so the
968                    // named step runs, then this one is tried again.
969                    (Some(target), _) => {
970                        let at = position_of(frame, &target)?;
971                        frame.detour = Some(frame.at);
972                        frame.at = at;
973                        Ok(())
974                    }
975                    // The same, for a workflow.
976                    (None, Some(workflow_id)) => {
977                        let workflow = self
978                            .description
979                            .workflows
980                            .iter()
981                            .find(|workflow| workflow.workflow_id == workflow_id)
982                            .ok_or_else(|| ExecutionError::UnknownWorkflow(workflow_id.clone()))?;
983                        let inputs = self.arguments(&arguments)?;
984                        self.enter(workflow, inputs, Some((step_id, Then::Retry)))
985                    }
986                    // Nothing named: this step, again.
987                    (None, None) => Ok(()),
988                }
989            }
990            Action::Goto {
991                step: Some(step_id),
992                ..
993            } => {
994                frame.at = position_of(frame, &step_id)?;
995                Ok(())
996            }
997            Action::Goto {
998                workflow: Some(workflow_id),
999                parameters: arguments,
1000                ..
1001            } => {
1002                let index = frame.order[frame.at];
1003                let step_id = frame.workflow.steps[index].step_id.clone();
1004                let workflow = self
1005                    .description
1006                    .workflows
1007                    .iter()
1008                    .find(|workflow| workflow.workflow_id == workflow_id)
1009                    .ok_or_else(|| ExecutionError::UnknownWorkflow(workflow_id.clone()))?;
1010                let inputs = self.arguments(&arguments)?;
1011                self.enter(workflow, inputs, Some((step_id, Then::EndCaller)))
1012            }
1013            Action::Goto { .. } => Ok(()),
1014        }
1015    }
1016
1017    /// The report for a run that has nothing left to do.
1018    fn finish(&mut self) -> ExecutionReport {
1019        ExecutionReport {
1020            workflow_id: self
1021                .options
1022                .workflow
1023                .clone()
1024                .or_else(|| {
1025                    self.description
1026                        .workflows
1027                        .first()
1028                        .map(|workflow| workflow.workflow_id.clone())
1029                })
1030                .unwrap_or_default(),
1031            outcome: Outcome::Succeeded,
1032            outputs: BTreeMap::new(),
1033            steps: std::mem::take(&mut self.records),
1034        }
1035    }
1036}
1037
1038/// Where a step sits in the order its workflow runs.
1039fn position_of(frame: &Frame<'_>, step_id: &str) -> Result<usize, ExecutionError> {
1040    let index = frame
1041        .workflow
1042        .steps
1043        .iter()
1044        .position(|step| step.step_id == step_id)
1045        .ok_or_else(|| ExecutionError::UnknownStep {
1046            workflow: frame.workflow.workflow_id.clone(),
1047            step: step_id.to_owned(),
1048        })?;
1049    Ok(frame
1050        .order
1051        .iter()
1052        .position(|&candidate| candidate == index)
1053        .unwrap_or(frame.order.len()))
1054}
1055
1056/// What a step's outcome asks the run to do.
1057#[derive(Clone, Debug)]
1058enum Action {
1059    Advance,
1060    End(Outcome),
1061    Retry {
1062        /// Which failure action asked, so its own budget is the one
1063        /// that is spent.
1064        at: usize,
1065        after: Option<f64>,
1066        /// A step to run before trying again, if the action names one.
1067        step: Option<String>,
1068        /// A workflow to run before trying again, if it names one.
1069        workflow: Option<String>,
1070        parameters: Vec<ReusableOr<Parameter>>,
1071    },
1072    Goto {
1073        step: Option<String>,
1074        workflow: Option<String>,
1075        parameters: Vec<ReusableOr<Parameter>>,
1076    },
1077}
1078
1079fn describe(action: &Action) -> Option<String> {
1080    match action {
1081        Action::Advance => None,
1082        Action::End(Outcome::Failed) => Some("ended, failed".to_owned()),
1083        Action::End(_) => Some("ended".to_owned()),
1084        Action::Retry {
1085            step: Some(step), ..
1086        } => Some(format!("retry via step `{step}`")),
1087        Action::Retry {
1088            workflow: Some(workflow),
1089            ..
1090        } => Some(format!("retry via workflow `{workflow}`")),
1091        Action::Retry { .. } => Some("retry".to_owned()),
1092        Action::Goto {
1093            step: Some(step), ..
1094        } => Some(format!("goto step `{step}`")),
1095        Action::Goto {
1096            workflow: Some(workflow),
1097            ..
1098        } => Some(format!("goto workflow `{workflow}`")),
1099        Action::Goto { .. } => None,
1100    }
1101}
1102
1103/// A parameter, resolved to a name, a place and a value.
1104struct Resolved {
1105    name: String,
1106    location: ParameterLocation,
1107    value: Value,
1108}
1109
1110/// Resolve a list of parameters, following `Reusable` references into
1111/// the description's components.
1112fn parameters(
1113    list: &[ReusableOr<Parameter>],
1114    description: &Description,
1115    scope: &Scope<'_>,
1116) -> Result<Vec<Resolved>, ExecutionError> {
1117    let mut resolved = Vec::with_capacity(list.len());
1118    for entry in list {
1119        let (parameter, overridden) = match entry {
1120            ReusableOr::Item(parameter) => (parameter.clone(), None),
1121            ReusableOr::Reusable(reusable) => {
1122                let name = reusable
1123                    .reference
1124                    .strip_prefix("$components.parameters.")
1125                    .ok_or_else(|| {
1126                        ExecutionError::Unsupported(format!(
1127                            "`{}` is not a component this executor can follow",
1128                            reusable.reference
1129                        ))
1130                    })?;
1131                let parameter = description
1132                    .components
1133                    .as_ref()
1134                    .and_then(|components| components.parameters.get(name))
1135                    .ok_or_else(|| {
1136                        ExecutionError::Unsupported(format!(
1137                            "`{}` names a component the description has not got",
1138                            reusable.reference
1139                        ))
1140                    })?;
1141                (parameter.clone(), reusable.value.clone())
1142            }
1143        };
1144        let value = match overridden {
1145            Some(value) => select::resolve(&value, scope)?,
1146            None => select::value_of(&parameter.value, scope)?,
1147        };
1148        resolved.push(Resolved {
1149            name: parameter.name,
1150            location: parameter.in_.unwrap_or(ParameterLocation::Query),
1151            value,
1152        });
1153    }
1154    Ok(resolved)
1155}
1156
1157/// A success action, following a `Reusable` into the components.
1158fn success_action(
1159    entry: &ReusableOr<roas_arazzo::v1_1::SuccessAction>,
1160    description: &Description,
1161) -> Result<roas_arazzo::v1_1::SuccessAction, ExecutionError> {
1162    match entry {
1163        ReusableOr::Item(action) => Ok(action.clone()),
1164        ReusableOr::Reusable(reusable) => reusable
1165            .reference
1166            .strip_prefix("$components.successActions.")
1167            .and_then(|name| {
1168                description
1169                    .components
1170                    .as_ref()
1171                    .and_then(|components| components.success_actions.get(name))
1172            })
1173            .cloned()
1174            .ok_or_else(|| {
1175                ExecutionError::Unsupported(format!(
1176                    "`{}` names a component the description has not got",
1177                    reusable.reference
1178                ))
1179            }),
1180    }
1181}
1182
1183/// A failure action, following a `Reusable` into the components.
1184fn failure_action(
1185    entry: &ReusableOr<roas_arazzo::v1_1::FailureAction>,
1186    description: &Description,
1187) -> Result<roas_arazzo::v1_1::FailureAction, ExecutionError> {
1188    match entry {
1189        ReusableOr::Item(action) => Ok(action.clone()),
1190        ReusableOr::Reusable(reusable) => reusable
1191            .reference
1192            .strip_prefix("$components.failureActions.")
1193            .and_then(|name| {
1194                description
1195                    .components
1196                    .as_ref()
1197                    .and_then(|components| components.failure_actions.get(name))
1198            })
1199            .cloned()
1200            .ok_or_else(|| {
1201                ExecutionError::Unsupported(format!(
1202                    "`{}` names a component the description has not got",
1203                    reusable.reference
1204                ))
1205            }),
1206    }
1207}
1208
1209/// Whether every criterion of an action holds. No criteria means the
1210/// action applies.
1211fn holds(criteria: &[Criterion], scope: &Scope<'_>) -> Result<bool, ExecutionError> {
1212    for criterion in criteria {
1213        if !criterion::passes(criterion, scope)? {
1214            return Ok(false);
1215        }
1216    }
1217    Ok(true)
1218}
1219
1220/// The values a set of `outputs` names, for a workflow that stopped
1221/// early.
1222///
1223/// An output naming a step or a workflow that never ran is expected and
1224/// skipped — that is what stopping early means. Nothing else is: an
1225/// input that was never given, a pointer into a body that has not got
1226/// it, a malformed selector, an unsupported expression — each is a
1227/// fault in the description, and a failed workflow is no reason to keep
1228/// quiet about it.
1229fn evaluate_what_ran(
1230    outputs: &BTreeMap<String, ValueOrSelector>,
1231    scope: &Scope<'_>,
1232) -> Result<BTreeMap<String, Value>, ExecutionError> {
1233    let mut named = BTreeMap::new();
1234    for (name, value) in outputs {
1235        match select::value_of(value, scope) {
1236            Ok(value) => {
1237                named.insert(name.clone(), value);
1238            }
1239            Err(SelectError::Expression(ExpressionError::NotRun { .. })) => {}
1240            Err(error) => return Err(error.into()),
1241        }
1242    }
1243    Ok(named)
1244}
1245
1246/// The values a set of `outputs` names.
1247fn evaluate_outputs(
1248    outputs: &BTreeMap<String, ValueOrSelector>,
1249    scope: &Scope<'_>,
1250) -> Result<BTreeMap<String, Value>, ExecutionError> {
1251    let mut resolved = BTreeMap::new();
1252    for (name, value) in outputs {
1253        resolved.insert(name.clone(), select::value_of(value, scope)?);
1254    }
1255    Ok(resolved)
1256}
1257
1258/// The body a step sends: what goes on the wire, what it means, and
1259/// what to call it.
1260struct Body {
1261    bytes: Vec<u8>,
1262    value: Value,
1263    content_type: String,
1264}
1265
1266/// The body a step sends, with its replacements applied.
1267fn body(step: &Step, scope: &Scope<'_>) -> Result<Option<Body>, ExecutionError> {
1268    let Some(request_body) = &step.request_body else {
1269        return Ok(None);
1270    };
1271    let mut payload = match &request_body.payload {
1272        Some(payload) => select::resolve(payload, scope)?,
1273        None => Value::Null,
1274    };
1275    for replacement in &request_body.replacements {
1276        let value = select::value_of(&replacement.value, scope)?;
1277        let language = match &replacement.target_selector_type {
1278            Some(type_) => select::kind_of(type_)?,
1279            None => select::Language::Pointer,
1280        };
1281        select::place(language, &replacement.target, &mut payload, value).map_err(|reason| {
1282            ExecutionError::BadRequest {
1283                step: step.step_id.clone(),
1284                reason,
1285            }
1286        })?;
1287    }
1288    let content_type = request_body
1289        .content_type
1290        .clone()
1291        .unwrap_or_else(|| "application/json".to_owned());
1292    // A string payload sent as anything but JSON goes as it is written;
1293    // everything else is JSON on the wire.
1294    let bytes = match (&payload, content_type.contains("json")) {
1295        (Value::String(text), false) => text.clone().into_bytes(),
1296        _ => payload.to_string().into_bytes(),
1297    };
1298    Ok(Some(Body {
1299        bytes,
1300        value: payload,
1301        content_type,
1302    }))
1303}
1304
1305/// The URL a request goes to: the server, the path with its parameters
1306/// filled in, and the query.
1307fn url(
1308    endpoint: &operation::Endpoint,
1309    path: &BTreeMap<String, Value>,
1310    query: &BTreeMap<String, Value>,
1311    querystring: Option<&str>,
1312) -> Result<String, String> {
1313    let mut filled = endpoint.path.clone();
1314    for (name, value) in path {
1315        filled = filled.replace(&format!("{{{name}}}"), &encode(&text(value)));
1316    }
1317    if let Some(start) = filled.find('{') {
1318        return Err(format!(
1319            "`{}` still has `{}` in it, which no parameter filled in",
1320            endpoint.path,
1321            &filled[start
1322                ..filled[start..]
1323                    .find('}')
1324                    .map_or(filled.len(), |end| start + end + 1)]
1325        ));
1326    }
1327    let mut url = format!("{}{filled}", endpoint.base);
1328    let pairs: Vec<String> = query
1329        .iter()
1330        .map(|(name, value)| format!("{}={}", encode(name), encode(&text(value))))
1331        .collect();
1332    let query = match (pairs.is_empty(), querystring) {
1333        (true, None) => String::new(),
1334        (true, Some(raw)) => raw.to_owned(),
1335        (false, None) => pairs.join("&"),
1336        (false, Some(raw)) => format!("{}&{raw}", pairs.join("&")),
1337    };
1338    if !query.is_empty() {
1339        url.push('?');
1340        url.push_str(&query);
1341    }
1342    url::Url::parse(&url).map_err(|error| format!("`{url}` is not a URL: {error}"))?;
1343    Ok(url)
1344}
1345
1346/// A value as the text that goes into a URL or a header: a string as it
1347/// stands, anything else as its JSON.
1348fn text(value: &Value) -> String {
1349    match value {
1350        Value::String(text) => text.clone(),
1351        other => other.to_string(),
1352    }
1353}
1354
1355/// Percent-encode everything a URL does not leave alone.
1356fn encode(text: &str) -> String {
1357    let mut encoded = String::with_capacity(text.len());
1358    for byte in text.bytes() {
1359        match byte {
1360            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
1361                encoded.push(byte as char);
1362            }
1363            other => encoded.push_str(&format!("%{other:02X}")),
1364        }
1365    }
1366    encoded
1367}
1368
1369/// The workflows to run, in an order that respects `dependsOn`, ending
1370/// with the one that was asked for.
1371fn ordered_workflows<'d>(
1372    description: &'d Description,
1373    wanted: &'d Workflow,
1374) -> Result<Vec<&'d Workflow>, ExecutionError> {
1375    let mut ordered = Vec::new();
1376    let mut visiting = BTreeSet::new();
1377    let mut done = BTreeSet::new();
1378    visit(description, wanted, &mut ordered, &mut visiting, &mut done)?;
1379    Ok(ordered)
1380}
1381
1382fn visit<'d>(
1383    description: &'d Description,
1384    workflow: &'d Workflow,
1385    ordered: &mut Vec<&'d Workflow>,
1386    visiting: &mut BTreeSet<String>,
1387    done: &mut BTreeSet<String>,
1388) -> Result<(), ExecutionError> {
1389    if done.contains(&workflow.workflow_id) {
1390        return Ok(());
1391    }
1392    if !visiting.insert(workflow.workflow_id.clone()) {
1393        return Err(ExecutionError::Circular(workflow.workflow_id.clone()));
1394    }
1395    for id in &workflow.depends_on {
1396        let dependency = description
1397            .workflows
1398            .iter()
1399            .find(|candidate| &candidate.workflow_id == id)
1400            .ok_or_else(|| ExecutionError::UnknownWorkflow(id.clone()))?;
1401        visit(description, dependency, ordered, visiting, done)?;
1402    }
1403    visiting.remove(&workflow.workflow_id);
1404    done.insert(workflow.workflow_id.clone());
1405    ordered.push(workflow);
1406    Ok(())
1407}
1408
1409/// Every step id a step's *expressions* read, which is a dependency
1410/// whether or not `dependsOn` says so.
1411///
1412/// "Tools MUST also treat runtime expression output references (e.g.,
1413/// `$steps.stepId.outputs.field`) as implicit dependencies" — so two
1414/// things matter. Only the fields that hold expressions are read, and
1415/// within them only what the runtime would really evaluate: a whole
1416/// `$…` string, a `{$…}` inside one, and in a condition the bare
1417/// operands its parser reads. A payload that merely mentions a step in
1418/// its text goes on the wire as text, and is no dependency at all.
1419///
1420/// A `Reusable` is followed into the components: where a parameter or
1421/// an action is written makes no difference to what it reads.
1422fn steps_named_by(step: &Step, description: &Description) -> BTreeSet<String> {
1423    let mut found = BTreeSet::new();
1424
1425    /// The step id an expression names, if it names one.
1426    fn named_in(expression: &str) -> Option<String> {
1427        let rest = expression.strip_prefix("$steps.")?;
1428        let (id, _) = rest.split_once('.')?;
1429        Some(id.to_owned())
1430    }
1431    fn read(text: &str, found: &mut BTreeSet<String>) {
1432        found.extend(
1433            expression::references(text)
1434                .into_iter()
1435                .filter_map(named_in),
1436        );
1437    }
1438    /// A condition is read the way its own type reads it: a `simple`
1439    /// one through the parser that evaluates it, anything else through
1440    /// the `{$…}` its engine has filled in first.
1441    fn read_condition(criterion: &Criterion, found: &mut BTreeSet<String>) {
1442        let simple = matches!(
1443            criterion.type_,
1444            None | Some(CriterionType::Simple(CriterionKind::Simple))
1445        );
1446        if simple {
1447            found.extend(
1448                criterion::expressions_in(&criterion.condition)
1449                    .iter()
1450                    .filter_map(|expression| named_in(expression)),
1451            );
1452        } else {
1453            // Only the `{$…}` — a pattern or a path is not evaluated as
1454            // an expression even when the whole of it starts with `$`,
1455            // which in a regex is an anchor.
1456            found.extend(
1457                expression::interpolations(&criterion.condition)
1458                    .into_iter()
1459                    .filter_map(named_in),
1460            );
1461        }
1462    }
1463    fn read_value(value: &ValueOrSelector, found: &mut BTreeSet<String>) {
1464        match value {
1465            ValueOrSelector::Literal(literal) => read_literal(literal, found),
1466            // A selector's context is an expression; its selector is a
1467            // JSONPath or a pointer, which the runtime does not
1468            // evaluate as one.
1469            ValueOrSelector::Selector(selector) => read(&selector.context, found),
1470        }
1471    }
1472    fn read_literal(value: &Value, found: &mut BTreeSet<String>) {
1473        match value {
1474            Value::String(text) => read(text, found),
1475            Value::Array(items) => items.iter().for_each(|item| read_literal(item, found)),
1476            Value::Object(members) => members
1477                .values()
1478                .for_each(|member| read_literal(member, found)),
1479            _ => {}
1480        }
1481    }
1482    fn read_parameters(
1483        list: &[ReusableOr<Parameter>],
1484        description: &Description,
1485        found: &mut BTreeSet<String>,
1486    ) {
1487        for entry in list {
1488            match entry {
1489                ReusableOr::Item(parameter) => read_value(&parameter.value, found),
1490                ReusableOr::Reusable(reusable) => {
1491                    // An override *replaces* the component's value, so
1492                    // it replaces what that value read too: the
1493                    // component's own dependency is not one here.
1494                    if let Some(overridden) = &reusable.value {
1495                        read_literal(overridden, found);
1496                    } else if let Some(parameter) = reusable
1497                        .reference
1498                        .strip_prefix("$components.parameters.")
1499                        .and_then(|name| {
1500                            description
1501                                .components
1502                                .as_ref()
1503                                .and_then(|components| components.parameters.get(name))
1504                        })
1505                    {
1506                        read_value(&parameter.value, found);
1507                    }
1508                }
1509            }
1510        }
1511    }
1512    fn read_criteria(list: &[Criterion], found: &mut BTreeSet<String>) {
1513        for criterion in list {
1514            if let Some(context) = &criterion.context {
1515                read(context, found);
1516            }
1517            read_condition(criterion, found);
1518        }
1519    }
1520
1521    read_parameters(&step.parameters, description, &mut found);
1522    read_criteria(&step.success_criteria, &mut found);
1523    for output in step.outputs.values() {
1524        read_value(output, &mut found);
1525    }
1526    if let Some(body) = &step.request_body {
1527        if let Some(payload) = &body.payload {
1528            read_literal(payload, &mut found);
1529        }
1530        for replacement in &body.replacements {
1531            read_value(&replacement.value, &mut found);
1532        }
1533    }
1534    for entry in &step.on_success {
1535        if let Ok(action) = success_action(entry, description) {
1536            read_criteria(&action.criteria, &mut found);
1537            read_parameters(&action.parameters, description, &mut found);
1538        }
1539    }
1540    for entry in &step.on_failure {
1541        if let Ok(action) = failure_action(entry, description) {
1542            read_criteria(&action.criteria, &mut found);
1543            read_parameters(&action.parameters, description, &mut found);
1544        }
1545    }
1546
1547    found.remove(&step.step_id);
1548    found
1549}
1550
1551/// Step indices in an order that respects `dependsOn` and the steps an
1552/// expression reads, keeping the document's order where neither says
1553/// anything.
1554fn ordered_steps(
1555    workflow: &Workflow,
1556    description: &Description,
1557) -> Result<Vec<usize>, ExecutionError> {
1558    let index: BTreeMap<&str, usize> = workflow
1559        .steps
1560        .iter()
1561        .enumerate()
1562        .map(|(at, step)| (step.step_id.as_str(), at))
1563        .collect();
1564    let mut ordered = Vec::with_capacity(workflow.steps.len());
1565    let mut visiting = BTreeSet::new();
1566    let mut done = BTreeSet::new();
1567    for step in &workflow.steps {
1568        visit_step(
1569            workflow,
1570            description,
1571            &index,
1572            step,
1573            &mut ordered,
1574            &mut visiting,
1575            &mut done,
1576        )?;
1577    }
1578    Ok(ordered)
1579}
1580
1581fn visit_step(
1582    workflow: &Workflow,
1583    description: &Description,
1584    index: &BTreeMap<&str, usize>,
1585    step: &Step,
1586    ordered: &mut Vec<usize>,
1587    visiting: &mut BTreeSet<String>,
1588    done: &mut BTreeSet<String>,
1589) -> Result<(), ExecutionError> {
1590    if done.contains(&step.step_id) {
1591        return Ok(());
1592    }
1593    if !visiting.insert(step.step_id.clone()) {
1594        return Err(ExecutionError::Circular(step.step_id.clone()));
1595    }
1596    for id in &step.depends_on {
1597        let at = index
1598            .get(id.as_str())
1599            .ok_or_else(|| ExecutionError::UnknownStep {
1600                workflow: workflow.workflow_id.clone(),
1601                step: id.clone(),
1602            })?;
1603        visit_step(
1604            workflow,
1605            description,
1606            index,
1607            &workflow.steps[*at],
1608            ordered,
1609            visiting,
1610            done,
1611        )?;
1612    }
1613    // The same for the steps this one reads. A name that is not a step
1614    // of this workflow is left alone: an expression may be wrong, and
1615    // saying so belongs where it is evaluated, with the whole context.
1616    for id in steps_named_by(step, description) {
1617        let Some(at) = index.get(id.as_str()) else {
1618            continue;
1619        };
1620        visit_step(
1621            workflow,
1622            description,
1623            index,
1624            &workflow.steps[*at],
1625            ordered,
1626            visiting,
1627            done,
1628        )?;
1629    }
1630    visiting.remove(&step.step_id);
1631    done.insert(step.step_id.clone());
1632    ordered.push(index[step.step_id.as_str()]);
1633    Ok(())
1634}