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

scientific_workflow/runtime/
phase.rs
