scientific_workflow/runtime/
phase.rs

1//! First-class phase and task declarations owned by the runtime.
2
3use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::sync::Arc;
6
7use serde::de::DeserializeOwned;
8use serde_json::Value;
9use sha2::{Digest, Sha256};
10
11use super::error::ReportingError;
12use super::task::{TaskContext, TaskResult, Workload};
13use crate::configuration::{ProjectConfig, TaskConfig};
14use crate::project::ScientificProject;
15
16/// Stable numeric identity of one execution phase.
17#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
18pub struct PhaseId(u64);
19
20impl PhaseId {
21    /// Creates a phase identity suitable for command-line selections such as
22    /// `[2, 4, 5]`.
23    pub const fn new(value: u64) -> Self {
24        Self(value)
25    }
26
27    /// Returns the exact numeric phase identity.
28    pub const fn get(self) -> u64 {
29        self.0
30    }
31}
32
33impl From<u64> for PhaseId {
34    fn from(value: u64) -> Self {
35        Self::new(value)
36    }
37}
38
39impl fmt::Display for PhaseId {
40    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
41        self.0.fmt(formatter)
42    }
43}
44
45/// Stable task identity scoped to one phase.
46#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
47pub struct TaskId(Arc<str>);
48
49impl TaskId {
50    /// Creates a task ID. Empty or whitespace-only IDs are rejected when the
51    /// owning phase is built, keeping task construction infallible.
52    pub fn new(value: impl Into<String>) -> Self {
53        Self(value.into().into())
54    }
55
56    /// Borrows the exact ID text.
57    pub fn as_str(&self) -> &str {
58        &self.0
59    }
60}
61
62impl fmt::Display for TaskId {
63    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
64        self.0.fmt(formatter)
65    }
66}
67
68/// Exact runtime task lookup key.
69#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
70pub struct TaskKey {
71    phase: PhaseId,
72    task: TaskId,
73}
74
75impl TaskKey {
76    /// Creates one phase-qualified task key.
77    pub fn new(phase: impl Into<PhaseId>, task: TaskId) -> Self {
78        Self {
79            phase: phase.into(),
80            task,
81        }
82    }
83
84    /// Returns the owning phase ID.
85    pub const fn phase_id(&self) -> PhaseId {
86        self.phase
87    }
88
89    /// Borrows the phase-local task ID.
90    pub fn task_id(&self) -> &TaskId {
91        &self.task
92    }
93}
94
95impl fmt::Display for TaskKey {
96    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
97        write!(formatter, "{}/{}", self.phase, self.task)
98    }
99}
100
101/// Reporting shape of one task.
102#[derive(Clone, Copy, Debug, Eq, PartialEq)]
103#[non_exhaustive]
104pub enum TaskDisplayKind {
105    /// Iterative work with an optional target supplied when execution starts.
106    Progress,
107    /// One-shot lifecycle work without an artificial iteration counter.
108    Activity,
109}
110
111/// First-class immutable task declaration owned by exactly one [`Phase`].
112pub struct Task {
113    key: TaskKey,
114    kind: Arc<str>,
115    configuration: TaskConfig,
116    display_kind: TaskDisplayKind,
117    label: Arc<str>,
118    display_keys: Option<Arc<[Box<str>]>>,
119    workload: Option<Workload>,
120    reused: bool,
121}
122
123impl Task {
124    /// Returns the exact phase-qualified task key.
125    pub fn key(&self) -> &TaskKey {
126        &self.key
127    }
128
129    /// Returns the phase-local task ID.
130    pub fn id(&self) -> &TaskId {
131        self.key.task_id()
132    }
133
134    /// Returns the task kind/namespace.
135    pub fn kind(&self) -> &str {
136        &self.kind
137    }
138
139    /// Returns the automatically generated display label.
140    pub fn label(&self) -> &str {
141        &self.label
142    }
143
144    /// Returns whether the task renders iterative progress or activity status.
145    pub const fn display_kind(&self) -> TaskDisplayKind {
146        self.display_kind
147    }
148
149    /// Returns the originating deterministic configuration ordinal.
150    pub fn configuration_ordinal(&self) -> u64 {
151        self.configuration.task_ordinal()
152    }
153
154    /// Borrows the retained cheap task-configuration handle.
155    pub fn configuration(&self) -> &TaskConfig {
156        &self.configuration
157    }
158
159    /// Borrows one fixed or swept parameter by exact key.
160    pub fn value(&self, key: &str) -> Option<&Value> {
161        self.configuration.value(key)
162    }
163
164    /// Borrows one required task parameter.
165    pub fn require_value(&self, key: &str) -> Result<&Value, ReportingError> {
166        self.value(key)
167            .ok_or_else(|| ReportingError::UnknownManagedTaskParameter {
168                task: self.key.to_string(),
169                key: key.to_owned(),
170            })
171    }
172
173    /// Decodes one required task parameter without first cloning its JSON tree.
174    pub fn decode_value<T>(&self, key: &str) -> Result<T, ReportingError>
175    where
176        T: DeserializeOwned,
177    {
178        T::deserialize(self.require_value(key)?).map_err(|source| {
179            ReportingError::DecodeManagedTaskParameter {
180                task: self.key.to_string(),
181                key: key.to_owned(),
182                source,
183            }
184        })
185    }
186
187    /// Iterates every fixed or swept parameter without cloning values.
188    pub fn iter(&self) -> impl Iterator<Item = (&str, &Value)> + '_ {
189        self.configuration.parameters().iter()
190    }
191
192    /// Borrows the exact keys used to generate the current label.
193    pub fn display_keys(&self) -> Option<&[Box<str>]> {
194        self.display_keys.as_deref()
195    }
196
197    pub(crate) fn take_workload(&mut self) -> Option<Workload> {
198        self.workload.take()
199    }
200
201    pub(crate) fn has_workload(&self) -> bool {
202        self.workload.is_some()
203    }
204
205    pub(crate) const fn is_reused(&self) -> bool {
206        self.reused
207    }
208
209    fn attach_to_phase(&mut self, phase: PhaseId) {
210        self.key.phase = phase;
211    }
212
213    fn parameter_keys(&self) -> Vec<&str> {
214        self.iter().map(|(key, _)| key).collect()
215    }
216
217    fn regenerate_label(&mut self, keys: Arc<[Box<str>]>, include_id: bool) {
218        let mut parts = Vec::with_capacity(keys.len() + usize::from(include_id));
219        for key in keys.iter() {
220            let value = self
221                .value(key)
222                .expect("validated display parameters resolve for every applicable task");
223            parts.push(format!("{key}={}", compact_value(value)));
224        }
225        if include_id {
226            parts.push(format!("id={}", self.key.task_id()));
227        }
228        self.label = if parts.is_empty() {
229            Arc::clone(&self.kind)
230        } else {
231            format!("{} {}", self.kind, parts.join(" ")).into()
232        };
233        self.display_keys = Some(keys);
234    }
235}
236
237impl fmt::Debug for Task {
238    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
239        formatter
240            .debug_struct("Task")
241            .field("key", &self.key)
242            .field("kind", &self.kind)
243            .field("label", &self.label)
244            .field("display_kind", &self.display_kind)
245            .field("parameters", &self.iter().count())
246            .finish_non_exhaustive()
247    }
248}
249
250/// Immutable nonempty execution phase and reporter section.
251pub struct Phase {
252    id: PhaseId,
253    label: Arc<str>,
254    tasks: Vec<Task>,
255    max_concurrent_workloads: usize,
256    queue_capacity: usize,
257    dependencies: Vec<PhaseId>,
258    failure_policy: PhaseFailurePolicy,
259    require_confirm: bool,
260}
261
262/// Scheduling behavior after the first workload failure in a phase.
263#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
264#[non_exhaustive]
265pub enum PhaseFailurePolicy {
266    /// Stop new work and cooperatively cancel active workloads.
267    #[default]
268    FailFast,
269    /// Stop new work but allow already active workloads to finish.
270    FinishActive,
271}
272
273impl PhaseFailurePolicy {
274    /// Returns the stable uncolored policy name used by plain output.
275    pub const fn as_str(self) -> &'static str {
276        match self {
277            Self::FailFast => "fail-fast",
278            Self::FinishActive => "finish-active",
279        }
280    }
281}
282
283impl Phase {
284    /// Begins declaring a phase. Tasks must be added before [`PhaseBuilder::build`].
285    pub fn builder(id: impl Into<PhaseId>, label: impl Into<String>) -> PhaseBuilder {
286        PhaseBuilder {
287            id: id.into(),
288            label: label.into(),
289            tasks: Vec::new(),
290            display_by_kind: HashMap::new(),
291            max_concurrent_workloads: 1,
292            queue_capacity: 1,
293            dependencies: Vec::new(),
294            failure_policy: PhaseFailurePolicy::FailFast,
295            require_confirm: false,
296        }
297    }
298
299    /// Returns the stable phase identity.
300    pub const fn id(&self) -> PhaseId {
301        self.id
302    }
303
304    /// Returns the human-facing phase heading.
305    pub fn label(&self) -> &str {
306        &self.label
307    }
308
309    /// Returns every task in deterministic display/execution order.
310    pub fn tasks(&self) -> &[Task] {
311        &self.tasks
312    }
313
314    /// Returns the phase-local concurrent workload ceiling.
315    pub const fn max_concurrent_workloads(&self) -> usize {
316        self.max_concurrent_workloads
317    }
318
319    /// Returns the prepared-but-not-running workload ceiling.
320    pub const fn queue_capacity(&self) -> usize {
321        self.queue_capacity
322    }
323
324    /// Returns declared predecessor phases in declaration order.
325    pub fn dependencies(&self) -> &[PhaseId] {
326        &self.dependencies
327    }
328
329    /// Returns the behavior selected for the first workload failure.
330    pub const fn failure_policy(&self) -> PhaseFailurePolicy {
331        self.failure_policy
332    }
333
334    /// Reports whether successful completion requires confirmation before the
335    /// next selected phase may start.
336    pub const fn requires_confirmation(&self) -> bool {
337        self.require_confirm
338    }
339
340    pub(crate) fn into_tasks(self) -> Vec<Task> {
341        self.tasks
342    }
343
344    /// Returns one exact phase-local task by ID.
345    pub fn task(&self, id: &TaskId) -> Option<&Task> {
346        self.tasks.iter().find(|task| task.key.task_id() == id)
347    }
348
349    /// Returns the sole task matching an exact partial selector.
350    pub fn unique_task_matching(&self, selector: &TaskSelector) -> Result<&Task, ReportingError> {
351        let mut matches = self.tasks.iter().filter(|task| selector.matches(task));
352        let first = matches
353            .next()
354            .ok_or_else(|| ReportingError::ManagedTaskNotFound {
355                selector: selector.to_string(),
356            })?;
357        if let Some(second) = matches.next() {
358            return Err(ReportingError::ManagedTaskSelectorAmbiguous {
359                selector: selector.to_string(),
360                first: first.key.to_string(),
361                second: second.key.to_string(),
362            });
363        }
364        Ok(first)
365    }
366}
367
368/// Builder for one nonempty first-class phase.
369pub struct PhaseBuilder {
370    id: PhaseId,
371    label: String,
372    tasks: Vec<Task>,
373    display_by_kind: HashMap<String, Vec<String>>,
374    max_concurrent_workloads: usize,
375    queue_capacity: usize,
376    dependencies: Vec<PhaseId>,
377    failure_policy: PhaseFailurePolicy,
378    require_confirm: bool,
379}
380
381impl PhaseBuilder {
382    /// Adds one iterative workload backed by an already selected configuration.
383    pub fn progress_workload<W>(
384        mut self,
385        configuration: TaskConfig,
386        kind: impl Into<String>,
387        workload: W,
388    ) -> Self
389    where
390        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
391    {
392        self.push_configuration_task(
393            configuration,
394            kind.into(),
395            TaskDisplayKind::Progress,
396            Some(Box::new(workload)),
397            false,
398        );
399        self
400    }
401
402    /// Adds one activity workload backed by an already selected configuration.
403    pub fn activity_workload<W>(
404        mut self,
405        configuration: TaskConfig,
406        kind: impl Into<String>,
407        workload: W,
408    ) -> Self
409    where
410        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
411    {
412        self.push_configuration_task(
413            configuration,
414            kind.into(),
415            TaskDisplayKind::Activity,
416            Some(Box::new(workload)),
417            false,
418        );
419        self
420    }
421
422    /// Adds one application-verified reused activity task.
423    pub fn reused_activity(mut self, configuration: TaskConfig, kind: impl Into<String>) -> Self {
424        self.push_configuration_task(
425            configuration,
426            kind.into(),
427            TaskDisplayKind::Activity,
428            None,
429            true,
430        );
431        self
432    }
433
434    /// Generates iterative tasks that share one thread-safe workload.
435    ///
436    /// Use this concise form when the callable can borrow shared captured
437    /// state. Use [`Self::progress_workloads_from_project`] when each task must
438    /// instead own a separately constructed resource.
439    pub fn progress_tasks_from_project<W>(
440        self,
441        project: &ScientificProject,
442        kind: impl Into<String>,
443        workload: W,
444    ) -> Self
445    where
446        W: Fn(&TaskContext) -> TaskResult + Send + Sync + 'static,
447    {
448        self.progress_tasks_from_configuration(project.configuration(), kind, workload)
449    }
450
451    /// Generates iterative tasks from configuration with one shared workload.
452    pub fn progress_tasks_from_configuration<W>(
453        mut self,
454        configuration: &ProjectConfig,
455        kind: impl Into<String>,
456        workload: W,
457    ) -> Self
458    where
459        W: Fn(&TaskContext) -> TaskResult + Send + Sync + 'static,
460    {
461        self.extend_shared_workload(
462            configuration,
463            kind.into(),
464            TaskDisplayKind::Progress,
465            workload,
466        );
467        self
468    }
469
470    /// Generates executable iterative tasks from every project configuration.
471    pub fn progress_workloads_from_project<F, W>(
472        self,
473        project: &ScientificProject,
474        kind: impl Into<String>,
475        factory: F,
476    ) -> Self
477    where
478        F: Fn(&TaskConfig) -> W,
479        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
480    {
481        self.progress_workloads_from_configuration(project.configuration(), kind, factory)
482    }
483
484    /// Generates executable iterative tasks from lower-level configuration.
485    pub fn progress_workloads_from_configuration<F, W>(
486        mut self,
487        configuration: &ProjectConfig,
488        kind: impl Into<String>,
489        factory: F,
490    ) -> Self
491    where
492        F: Fn(&TaskConfig) -> W,
493        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
494    {
495        self.extend_configuration_workloads(
496            configuration,
497            kind.into(),
498            TaskDisplayKind::Progress,
499            factory,
500        );
501        self
502    }
503
504    /// Generates activity tasks that share one thread-safe workload.
505    ///
506    /// Use [`Self::activity_workloads_from_project`] when each task needs a
507    /// separately constructed single-owner resource.
508    pub fn activity_tasks_from_project<W>(
509        self,
510        project: &ScientificProject,
511        kind: impl Into<String>,
512        workload: W,
513    ) -> Self
514    where
515        W: Fn(&TaskContext) -> TaskResult + Send + Sync + 'static,
516    {
517        self.activity_tasks_from_configuration(project.configuration(), kind, workload)
518    }
519
520    /// Generates activity tasks from configuration with one shared workload.
521    pub fn activity_tasks_from_configuration<W>(
522        mut self,
523        configuration: &ProjectConfig,
524        kind: impl Into<String>,
525        workload: W,
526    ) -> Self
527    where
528        W: Fn(&TaskContext) -> TaskResult + Send + Sync + 'static,
529    {
530        self.extend_shared_workload(
531            configuration,
532            kind.into(),
533            TaskDisplayKind::Activity,
534            workload,
535        );
536        self
537    }
538
539    /// Generates executable activity tasks from every project configuration.
540    pub fn activity_workloads_from_project<F, W>(
541        self,
542        project: &ScientificProject,
543        kind: impl Into<String>,
544        factory: F,
545    ) -> Self
546    where
547        F: Fn(&TaskConfig) -> W,
548        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
549    {
550        self.activity_workloads_from_configuration(project.configuration(), kind, factory)
551    }
552
553    /// Generates executable activity tasks from lower-level configuration.
554    pub fn activity_workloads_from_configuration<F, W>(
555        mut self,
556        configuration: &ProjectConfig,
557        kind: impl Into<String>,
558        factory: F,
559    ) -> Self
560    where
561        F: Fn(&TaskConfig) -> W,
562        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
563    {
564        self.extend_configuration_workloads(
565            configuration,
566            kind.into(),
567            TaskDisplayKind::Activity,
568            factory,
569        );
570        self
571    }
572
573    /// Selects the exact parameter subset used to label one task kind.
574    pub fn display_tasks_by<I, S>(mut self, kind: impl Into<String>, keys: I) -> Self
575    where
576        I: IntoIterator<Item = S>,
577        S: Into<String>,
578    {
579        self.display_by_kind
580            .insert(kind.into(), keys.into_iter().map(Into::into).collect());
581        self
582    }
583
584    /// Sets the later scheduler's phase-local active workload ceiling.
585    pub fn max_concurrent_workloads(mut self, maximum: usize) -> Self {
586        self.max_concurrent_workloads = maximum;
587        self
588    }
589
590    /// Sets the later scheduler's prepared-work queue capacity.
591    pub fn queue_capacity(mut self, capacity: usize) -> Self {
592        self.queue_capacity = capacity;
593        self
594    }
595
596    /// Declares one phase that must be satisfied before this phase starts.
597    pub fn depends_on(mut self, dependency: impl Into<PhaseId>) -> Self {
598        self.dependencies.push(dependency.into());
599        self
600    }
601
602    /// Selects behavior after the first workload failure.
603    pub fn failure_policy(mut self, policy: PhaseFailurePolicy) -> Self {
604        self.failure_policy = policy;
605        self
606    }
607
608    /// Requires the user to type `yes` before advancing to the next phase.
609    ///
610    /// The default is `false`. This setting has no effect when this phase is
611    /// the final selected phase because there is no transition to confirm.
612    pub fn require_confirm(mut self, require: bool) -> Self {
613        self.require_confirm = require;
614        self
615    }
616
617    /// Validates and creates one immutable nonempty phase.
618    pub fn build(mut self) -> Result<Phase, ReportingError> {
619        if self.label.trim().is_empty() {
620            return Err(ReportingError::InvalidPhaseLabel { phase: self.id.0 });
621        }
622        if self.tasks.is_empty() {
623            return Err(ReportingError::EmptyPhase { phase: self.id.0 });
624        }
625        if self.max_concurrent_workloads == 0 {
626            return Err(ReportingError::InvalidPhaseWorkloadLimit { phase: self.id.0 });
627        }
628        if self.queue_capacity == 0 {
629            return Err(ReportingError::InvalidPhaseQueueCapacity { phase: self.id.0 });
630        }
631
632        let mut ids = HashSet::with_capacity(self.tasks.len());
633        for task in &mut self.tasks {
634            if task.key.task_id().as_str().trim().is_empty() {
635                return Err(ReportingError::InvalidManagedTaskId { phase: self.id.0 });
636            }
637            if task.kind.trim().is_empty() {
638                return Err(ReportingError::InvalidManagedTaskKind {
639                    task: task.key.task_id().to_string(),
640                });
641            }
642            if !ids.insert(task.key.task_id().clone()) {
643                return Err(ReportingError::DuplicateManagedTaskId {
644                    phase: self.id.0,
645                    task: task.key.task_id().to_string(),
646                });
647            }
648            task.attach_to_phase(self.id);
649        }
650
651        let mut dependencies = HashSet::with_capacity(self.dependencies.len());
652        for dependency in &self.dependencies {
653            if !dependencies.insert(*dependency) || *dependency == self.id {
654                return Err(ReportingError::PhaseDependencyCycle { phase: self.id.0 });
655            }
656        }
657
658        let kinds: HashSet<_> = self
659            .tasks
660            .iter()
661            .map(|task| task.kind.to_string())
662            .collect();
663        for requested in self.display_by_kind.keys() {
664            if !kinds.contains(requested) {
665                return Err(ReportingError::UnknownManagedTaskKind {
666                    kind: requested.clone(),
667                });
668            }
669        }
670        for kind in kinds {
671            self.generate_labels(&kind)?;
672        }
673
674        Ok(Phase {
675            id: self.id,
676            label: self.label.into(),
677            tasks: self.tasks,
678            max_concurrent_workloads: self.max_concurrent_workloads,
679            queue_capacity: self.queue_capacity,
680            dependencies: self.dependencies,
681            failure_policy: self.failure_policy,
682            require_confirm: self.require_confirm,
683        })
684    }
685
686    fn extend_configuration_workloads<F, W>(
687        &mut self,
688        configuration: &ProjectConfig,
689        kind: String,
690        display_kind: TaskDisplayKind,
691        factory: F,
692    ) where
693        F: Fn(&TaskConfig) -> W,
694        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
695    {
696        for config in configuration.task_configs() {
697            let workload = factory(&config);
698            self.push_configuration_task(
699                config,
700                kind.clone(),
701                display_kind,
702                Some(Box::new(workload)),
703                false,
704            );
705        }
706    }
707
708    fn extend_shared_workload<W>(
709        &mut self,
710        configuration: &ProjectConfig,
711        kind: String,
712        display_kind: TaskDisplayKind,
713        workload: W,
714    ) where
715        W: Fn(&TaskContext) -> TaskResult + Send + Sync + 'static,
716    {
717        let workload = Arc::new(workload);
718        for config in configuration.task_configs() {
719            let workload = Arc::clone(&workload);
720            self.push_configuration_task(
721                config,
722                kind.clone(),
723                display_kind,
724                Some(Box::new(move |context| workload(context))),
725                false,
726            );
727        }
728    }
729
730    fn push_configuration_task(
731        &mut self,
732        configuration: TaskConfig,
733        kind: String,
734        display_kind: TaskDisplayKind,
735        workload: Option<Workload>,
736        reused: bool,
737    ) {
738        let id = TaskId::new(format!("{kind}:{}", configuration.task_ordinal()));
739        self.tasks.push(Task {
740            key: TaskKey::new(self.id, id),
741            kind: kind.clone().into(),
742            configuration,
743            display_kind,
744            label: kind.into(),
745            display_keys: None,
746            workload,
747            reused,
748        });
749    }
750
751    fn generate_labels(&mut self, kind: &str) -> Result<(), ReportingError> {
752        let positions: Vec<_> = self
753            .tasks
754            .iter()
755            .enumerate()
756            .filter_map(|(position, task)| (task.kind() == kind).then_some(position))
757            .collect();
758        validate_parameter_keys(&self.tasks, &positions, kind)?;
759        let requested = self.display_by_kind.get(kind);
760        let keys = if let Some(requested) = requested {
761            validate_display_keys(&self.tasks, &positions, requested)?
762        } else {
763            varying_keys(&self.tasks, &positions)
764        };
765        let keys: Arc<[Box<str>]> = keys.into_iter().map(String::into_boxed_str).collect();
766        let include_id = keys.is_empty() && positions.len() > 1;
767        let mut labels = HashMap::<String, TaskKey>::with_capacity(positions.len());
768        for position in positions {
769            self.tasks[position].regenerate_label(Arc::clone(&keys), include_id);
770            let task = &self.tasks[position];
771            if let Some(first) = labels.insert(task.label().to_owned(), task.key().clone()) {
772                return Err(ReportingError::ManagedTaskDisplayCollision {
773                    label: task.label().to_owned(),
774                    first: first.to_string(),
775                    second: task.key().to_string(),
776                });
777            }
778        }
779        Ok(())
780    }
781}
782
783impl fmt::Debug for Phase {
784    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
785        formatter
786            .debug_struct("Phase")
787            .field("id", &self.id)
788            .field("label", &self.label)
789            .field("tasks", &self.tasks.len())
790            .field("max_concurrent_workloads", &self.max_concurrent_workloads)
791            .field("queue_capacity", &self.queue_capacity)
792            .field("dependencies", &self.dependencies)
793            .field("failure_policy", &self.failure_policy)
794            .field("require_confirm", &self.require_confirm)
795            .finish()
796    }
797}
798
799impl fmt::Debug for PhaseBuilder {
800    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
801        formatter
802            .debug_struct("PhaseBuilder")
803            .field("id", &self.id)
804            .field("label", &self.label)
805            .field("tasks", &self.tasks.len())
806            .field("max_concurrent_workloads", &self.max_concurrent_workloads)
807            .field("queue_capacity", &self.queue_capacity)
808            .field("dependencies", &self.dependencies)
809            .field("require_confirm", &self.require_confirm)
810            .finish_non_exhaustive()
811    }
812}
813
814/// Exact partial selector over phase, task kind, and structured parameters.
815#[derive(Clone, Debug, Default)]
816pub struct TaskSelector {
817    phase: Option<PhaseId>,
818    kind: Option<Arc<str>>,
819    parameters: Vec<(Box<str>, Value)>,
820}
821
822impl TaskSelector {
823    /// Creates an unconstrained selector.
824    pub fn new() -> Self {
825        Self::default()
826    }
827
828    /// Constrains selection to one phase.
829    pub fn phase(mut self, phase: impl Into<PhaseId>) -> Self {
830        self.phase = Some(phase.into());
831        self
832    }
833
834    /// Constrains selection to one task kind/namespace.
835    pub fn kind(mut self, kind: impl Into<String>) -> Self {
836        self.kind = Some(kind.into().into());
837        self
838    }
839
840    /// Adds or replaces one exact parameter constraint.
841    pub fn parameter(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
842        let key = key.into();
843        if let Some((_, current)) = self
844            .parameters
845            .iter_mut()
846            .find(|(candidate, _)| candidate.as_ref() == key)
847        {
848            *current = value.into();
849        } else {
850            self.parameters.push((key.into_boxed_str(), value.into()));
851        }
852        self
853    }
854
855    pub(crate) fn matches(&self, task: &Task) -> bool {
856        self.phase.is_none_or(|phase| task.key.phase == phase)
857            && self.kind.as_deref().is_none_or(|kind| task.kind() == kind)
858            && self
859                .parameters
860                .iter()
861                .all(|(key, value)| task.value(key) == Some(value))
862    }
863}
864
865impl fmt::Display for TaskSelector {
866    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
867        let mut fields = Vec::new();
868        if let Some(phase) = self.phase {
869            fields.push(format!("phase={phase}"));
870        }
871        if let Some(kind) = &self.kind {
872            fields.push(format!("kind={kind}"));
873        }
874        fields.extend(
875            self.parameters
876                .iter()
877                .map(|(key, value)| format!("{key}={}", compact_value(value))),
878        );
879        formatter.write_str(&fields.join(", "))
880    }
881}
882
883fn varying_keys(tasks: &[Task], positions: &[usize]) -> Vec<String> {
884    let Some(&first) = positions.first() else {
885        return Vec::new();
886    };
887    tasks[first]
888        .parameter_keys()
889        .into_iter()
890        .filter(|key| {
891            let first_value = tasks[first].value(key);
892            positions
893                .iter()
894                .skip(1)
895                .any(|&position| tasks[position].value(key) != first_value)
896        })
897        .map(str::to_owned)
898        .collect()
899}
900
901fn validate_parameter_keys(
902    tasks: &[Task],
903    positions: &[usize],
904    kind: &str,
905) -> Result<(), ReportingError> {
906    let Some(&first) = positions.first() else {
907        return Ok(());
908    };
909    let expected: HashSet<_> = tasks[first].parameter_keys().into_iter().collect();
910    for &position in positions.iter().skip(1) {
911        let actual: HashSet<_> = tasks[position].parameter_keys().into_iter().collect();
912        if actual != expected {
913            return Err(ReportingError::InconsistentManagedTaskParameters {
914                kind: kind.to_owned(),
915                first: tasks[first].key().to_string(),
916                second: tasks[position].key().to_string(),
917            });
918        }
919    }
920    Ok(())
921}
922
923fn validate_display_keys(
924    tasks: &[Task],
925    positions: &[usize],
926    requested: &[String],
927) -> Result<Vec<String>, ReportingError> {
928    let mut seen = HashSet::with_capacity(requested.len());
929    for key in requested {
930        if !seen.insert(key.as_str()) {
931            return Err(ReportingError::DuplicateIdentityParameter { key: key.clone() });
932        }
933        if positions
934            .iter()
935            .any(|&position| tasks[position].value(key).is_none())
936        {
937            return Err(ReportingError::UnknownIdentityParameter { key: key.clone() });
938        }
939    }
940    Ok(requested.to_vec())
941}
942
943fn compact_value(value: &Value) -> String {
944    match value {
945        Value::Array(values) => format!("<array:{}:{}>", values.len(), short_hash(value)),
946        Value::Object(values) => format!("<object:{}:{}>", values.len(), short_hash(value)),
947        _ => serde_json::to_string(value)
948            .expect("serde_json::Value always serializes to valid compact JSON"),
949    }
950}
951
952fn short_hash(value: &Value) -> String {
953    let bytes = serde_json::to_vec(value)
954        .expect("serde_json::Value always serializes to valid compact JSON");
955    let digest = Sha256::digest(bytes);
956    digest[..4]
957        .iter()
958        .map(|byte| format!("{byte:02x}"))
959        .collect::<Vec<_>>()
960        .join("")
961}
scientific_workflow/runtime/phase.rs

scientific_workflow/runtime/
phase.rs
