scientific_workflow/reporting/
progress.rs

1//! Thread-safe task progress state and the sole terminal renderer.
2//!
3//! Workers publish only atomics on the hot path. A single renderer thread polls
4//! those slots at a bounded frequency and owns every human-facing terminal
5//! write for the session. Progress never mutates or replaces scientific time;
6//! callers synchronize it from their authoritative model state.
7
8use std::collections::{HashMap, HashSet};
9use std::fmt;
10use std::io::{self, IsTerminal};
11use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};
12use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
13use std::sync::{Arc, Mutex};
14use std::thread::{self, JoinHandle};
15use std::time::Duration;
16
17use console::Term;
18use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle};
19use serde_json::Value;
20
21use super::error::ReportingError;
22use crate::configuration::{ProjectConfig, TaskConfig};
23use crate::project::ScientificProject;
24
25const REFRESH_INTERVAL: Duration = Duration::from_millis(100);
26static TERMINAL_OWNED: AtomicBool = AtomicBool::new(false);
27
28/// Lifecycle status of one independently executing scientific task.
29#[derive(Clone, Copy, Debug, Eq, PartialEq)]
30#[non_exhaustive]
31pub enum TaskStatus {
32    /// The task is registered but has not started.
33    Pending,
34    /// The task currently owns an active [`TaskProgress`] handle.
35    Running,
36    /// Evolution, persistence, and caller-defined validation completed.
37    Completed,
38    /// The task explicitly failed or dropped its active handle prematurely.
39    Failed,
40}
41
42impl TaskStatus {
43    fn encode(self) -> u8 {
44        match self {
45            Self::Pending => 0,
46            Self::Running => 1,
47            Self::Completed => 2,
48            Self::Failed => 3,
49        }
50    }
51
52    fn decode(value: u8) -> Self {
53        match value {
54            0 => Self::Pending,
55            1 => Self::Running,
56            2 => Self::Completed,
57            3 => Self::Failed,
58            _ => unreachable!("task status is written only through TaskStatus::encode"),
59        }
60    }
61
62    fn label(self) -> &'static str {
63        match self {
64            Self::Pending => "pending",
65            Self::Running => "running",
66            Self::Completed => "completed",
67            Self::Failed => "failed",
68        }
69    }
70}
71
72/// Exact parameter-derived identity of one task.
73///
74/// The identity contains only the caller-selected parameter fields. Its label
75/// is deterministic compact JSON text intended for reporting, while equality
76/// validation uses the retained JSON values themselves.
77#[derive(Clone, Debug)]
78pub struct TaskIdentity {
79    task: TaskConfig,
80    keys: Arc<[Box<str>]>,
81    label: Arc<str>,
82}
83
84impl TaskIdentity {
85    /// Returns the terminal label derived from exact parameter key/value pairs.
86    pub fn label(&self) -> &str {
87        &self.label
88    }
89
90    /// Returns the number of parameter fields forming this identity.
91    pub fn len(&self) -> usize {
92        self.keys.len()
93    }
94
95    /// Reports whether the identity contains no parameter fields.
96    pub fn is_empty(&self) -> bool {
97        self.keys.is_empty()
98    }
99
100    /// Borrows one exact identity value by parameter name.
101    pub fn value(&self, key: &str) -> Option<&Value> {
102        self.keys.iter().find_map(|name| {
103            (name.as_ref() == key)
104                .then(|| self.task.value(key))
105                .flatten()
106        })
107    }
108
109    /// Iterates identity fields in the configured display order.
110    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&str, &Value)> {
111        self.keys.iter().map(|name| {
112            (
113                name.as_ref(),
114                self.task
115                    .value(name)
116                    .expect("validated identity keys resolve for every task"),
117            )
118        })
119    }
120
121    fn matches(&self, task: &TaskConfig) -> bool {
122        self.keys
123            .iter()
124            .all(|key| task.value(key) == self.task.value(key))
125    }
126}
127
128/// Immutable aggregate captured when centralized reporting ends.
129#[derive(Clone, Debug, Eq, PartialEq)]
130pub struct ProgressSummary {
131    total: u64,
132    pending: u64,
133    running: u64,
134    completed: u64,
135    failed: u64,
136}
137
138impl ProgressSummary {
139    /// Returns the number of registered tasks.
140    pub fn total(&self) -> u64 {
141        self.total
142    }
143
144    /// Returns the number of tasks that never started.
145    pub fn pending(&self) -> u64 {
146        self.pending
147    }
148
149    /// Returns the number of tasks still running at capture time.
150    pub fn running(&self) -> u64 {
151        self.running
152    }
153
154    /// Returns the number of successfully completed tasks.
155    pub fn completed(&self) -> u64 {
156        self.completed
157    }
158
159    /// Returns the number of failed or interrupted tasks.
160    pub fn failed(&self) -> u64 {
161        self.failed
162    }
163
164    /// Reports whether every registered task completed successfully.
165    pub fn is_success(&self) -> bool {
166        self.completed == self.total && self.pending == 0 && self.running == 0 && self.failed == 0
167    }
168}
169
170/// Output policy selected before the renderer starts.
171#[derive(Clone, Copy, Debug, Eq, PartialEq)]
172enum OutputMode {
173    Auto,
174    Terminal,
175    Plain,
176    Hidden,
177}
178
179/// Builder for one process-wide progress-reporting session.
180pub struct ProgressReporterBuilder {
181    configuration: ProjectConfig,
182    identity_keys: Option<Vec<String>>,
183    output: OutputMode,
184}
185
186impl ProgressReporterBuilder {
187    /// Selects the exact parameter combination used to identify every task.
188    ///
189    /// Keys may refer to fixed or swept parameters, but the complete selected
190    /// tuple must be unique for every generated task. Calling this method with
191    /// no keys is valid only for a one-task project.
192    pub fn identify_tasks_by<I, S>(mut self, keys: I) -> Self
193    where
194        I: IntoIterator<Item = S>,
195        S: Into<String>,
196    {
197        self.identity_keys = Some(keys.into_iter().map(Into::into).collect());
198        self
199    }
200
201    /// Forces cursor-controlled terminal rendering even when stderr is not a
202    /// terminal. Most applications should retain automatic output selection.
203    pub fn terminal(mut self) -> Self {
204        self.output = OutputMode::Terminal;
205        self
206    }
207
208    /// Forces stable line-oriented output suitable for logs and CI systems.
209    pub fn plain(mut self) -> Self {
210        self.output = OutputMode::Plain;
211        self
212    }
213
214    /// Suppresses rendering while retaining counters and lifecycle validation.
215    ///
216    /// This is useful for tests and embedding applications with their own
217    /// non-terminal presentation layer. It does not disable progress tracking.
218    pub fn hidden(mut self) -> Self {
219        self.output = OutputMode::Hidden;
220        self
221    }
222
223    /// Validates identities, acquires exclusive terminal ownership, and starts
224    /// the centralized renderer.
225    pub fn start(self) -> Result<ProgressReporter, ReportingError> {
226        let identity_keys: Arc<[Box<str>]> =
227            validate_identity_keys(&self.configuration, self.identity_keys)?
228                .into_iter()
229                .map(String::into_boxed_str)
230                .collect();
231        let slots = build_slots(&self.configuration, Arc::clone(&identity_keys))?;
232        let output = match self.output {
233            OutputMode::Auto if io::stderr().is_terminal() => OutputMode::Terminal,
234            OutputMode::Auto => OutputMode::Plain,
235            explicit => explicit,
236        };
237
238        acquire_terminal()?;
239        let lease = TerminalLease;
240        let (events, receiver) = mpsc::channel();
241        let renderer_slots = Arc::clone(&slots);
242        let renderer = thread::Builder::new()
243            .name("scientific-workflow-progress".to_owned())
244            .spawn(move || render(receiver, renderer_slots, output, lease))
245            .map_err(|source| ReportingError::StartRenderer { source })?;
246
247        Ok(ProgressReporter {
248            inner: Arc::new(ReporterInner {
249                slots,
250                identity_keys,
251                events,
252            }),
253            renderer: Some(renderer),
254            finished: false,
255        })
256    }
257}
258
259impl fmt::Debug for ProgressReporterBuilder {
260    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
261        formatter
262            .debug_struct("ProgressReporterBuilder")
263            .field("tasks", &self.configuration.task_count())
264            .field("identity_keys", &self.identity_keys)
265            .field("output", &self.output)
266            .finish_non_exhaustive()
267    }
268}
269
270/// Central progress registry and exclusive human-facing terminal owner.
271pub struct ProgressReporter {
272    inner: Arc<ReporterInner>,
273    renderer: Option<JoinHandle<()>>,
274    finished: bool,
275}
276
277impl ProgressReporter {
278    /// Creates a builder from a conventional four-file scientific project.
279    ///
280    /// By default all sweep keys, in declaration order, form task identity.
281    pub fn for_project(project: &ScientificProject) -> ProgressReporterBuilder {
282        Self::for_configuration(project.configuration())
283    }
284
285    /// Creates a builder from the lower-level three-file project configuration.
286    pub fn for_configuration(configuration: &ProjectConfig) -> ProgressReporterBuilder {
287        ProgressReporterBuilder {
288            configuration: configuration.clone(),
289            identity_keys: None,
290            output: OutputMode::Auto,
291        }
292    }
293
294    /// Starts tracking one task using identity and ordering from `TaskConfig`.
295    ///
296    /// `initial_iteration` and `target_iteration` are absolute scientific
297    /// coordinates. A missing target creates an indeterminate progress display.
298    /// Callers never supply an ordinal or terminal label.
299    pub fn start_task(
300        &self,
301        task: &TaskConfig,
302        initial_iteration: u64,
303        target_iteration: Option<u64>,
304    ) -> Result<TaskProgress, ReportingError> {
305        let ordinal = task.task_ordinal();
306        let index = usize::try_from(ordinal)
307            .ok()
308            .filter(|index| *index < self.inner.slots.len())
309            .ok_or(ReportingError::UnknownTaskOrdinal {
310                task_ordinal: ordinal,
311            })?;
312        let slot = Arc::clone(&self.inner.slots[index]);
313        if !slot.identity.matches(task) {
314            return Err(ReportingError::TaskIdentityMismatch {
315                task_ordinal: ordinal,
316            });
317        }
318        if let Some(target) = target_iteration.filter(|target| initial_iteration > *target) {
319            return Err(ReportingError::InitialIterationBeyondTarget {
320                identity: slot.identity.label().to_owned(),
321                initial: initial_iteration,
322                target,
323            });
324        }
325        slot.status
326            .compare_exchange(
327                TaskStatus::Pending.encode(),
328                TaskStatus::Running.encode(),
329                Ordering::AcqRel,
330                Ordering::Acquire,
331            )
332            .map_err(|_| ReportingError::TaskAlreadyStarted {
333                identity: slot.identity.label().to_owned(),
334            })?;
335        slot.current.store(initial_iteration, Ordering::Relaxed);
336        if let Some(target) = target_iteration {
337            slot.target.store(target, Ordering::Relaxed);
338            slot.target_known.store(true, Ordering::Release);
339        } else {
340            slot.target_known.store(false, Ordering::Release);
341        }
342        *lock(&slot.phase) = "running".into();
343
344        Ok(TaskProgress {
345            slot,
346            events: self.inner.events.clone(),
347            active: true,
348        })
349    }
350
351    /// Sends one application-wide message through the sole renderer.
352    pub fn report(&self, message: impl Into<String>) -> Result<(), ReportingError> {
353        self.inner
354            .events
355            .send(RenderEvent::Message(message.into()))
356            .map_err(|_| ReportingError::RendererUnavailable)
357    }
358
359    /// Returns a non-blocking snapshot of all task lifecycle counts.
360    pub fn summary(&self) -> ProgressSummary {
361        summarize(&self.inner.slots)
362    }
363
364    /// Finishes a successful session and emits its final summary and message.
365    ///
366    /// Every task must already be completed. The method stops and joins the
367    /// renderer and releases exclusive terminal ownership.
368    pub fn complete(
369        mut self,
370        message: impl Into<String>,
371    ) -> Result<ProgressSummary, ReportingError> {
372        let summary = self.summary();
373        if !summary.is_success() {
374            self.stop(false, "workflow did not complete".to_owned())?;
375            return Err(ReportingError::IncompleteProgress {
376                pending: summary.pending,
377                running: summary.running,
378                failed: summary.failed,
379            });
380        }
381        self.stop(true, message.into())?;
382        Ok(summary)
383    }
384
385    /// Finishes an unsuccessful session while preserving all task statuses.
386    pub fn fail(mut self, message: impl Into<String>) -> Result<ProgressSummary, ReportingError> {
387        let summary = self.summary();
388        self.stop(false, message.into())?;
389        Ok(summary)
390    }
391
392    /// Emits one terminal error through the reporting subsystem.
393    ///
394    /// This helper is intended for failures occurring before a reporting
395    /// session starts or after it has released the terminal lease.
396    pub fn report_error(message: impl fmt::Display) {
397        eprintln!("[error] {message}");
398    }
399
400    fn stop(&mut self, success: bool, message: String) -> Result<(), ReportingError> {
401        self.inner
402            .events
403            .send(RenderEvent::Stop { success, message })
404            .map_err(|_| ReportingError::RendererUnavailable)?;
405        self.finished = true;
406        if self
407            .renderer
408            .take()
409            .expect("an unfinished reporter owns one renderer")
410            .join()
411            .is_err()
412        {
413            return Err(ReportingError::RendererPanicked);
414        }
415        Ok(())
416    }
417}
418
419impl fmt::Debug for ProgressReporter {
420    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
421        formatter
422            .debug_struct("ProgressReporter")
423            .field("tasks", &self.inner.slots.len())
424            .field("identity_keys", &self.inner.identity_keys)
425            .field("summary", &self.summary())
426            .finish_non_exhaustive()
427    }
428}
429
430impl Drop for ProgressReporter {
431    fn drop(&mut self) {
432        if self.finished {
433            return;
434        }
435        let _ = self.inner.events.send(RenderEvent::Stop {
436            success: false,
437            message: "progress reporter dropped before completion".to_owned(),
438        });
439        if let Some(renderer) = self.renderer.take() {
440            let _ = renderer.join();
441        }
442    }
443}
444
445/// Non-clone task-local progress handle.
446///
447/// Dropping a running handle without calling [`TaskProgress::complete`] or
448/// [`TaskProgress::fail`] marks the task failed, making ordinary `?` returns
449/// safe without a separate cleanup branch.
450pub struct TaskProgress {
451    slot: Arc<ProgressSlot>,
452    events: Sender<RenderEvent>,
453    active: bool,
454}
455
456impl TaskProgress {
457    /// Borrows the exact parameter-derived task identity.
458    pub fn identity(&self) -> &TaskIdentity {
459        &self.slot.identity
460    }
461
462    /// Returns the latest reported absolute simulation iteration.
463    pub fn current_iteration(&self) -> u64 {
464        self.slot.current.load(Ordering::Relaxed)
465    }
466
467    /// Returns the known absolute target iteration, if one exists.
468    pub fn target_iteration(&self) -> Option<u64> {
469        if self.slot.target_known.load(Ordering::Acquire) {
470            Some(self.slot.target.load(Ordering::Relaxed))
471        } else {
472            None
473        }
474    }
475
476    /// Returns this task's current lifecycle status.
477    pub fn status(&self) -> TaskStatus {
478        TaskStatus::decode(self.slot.status.load(Ordering::Acquire))
479    }
480
481    /// Synchronizes progress to an authoritative absolute simulation iteration.
482    ///
483    /// The atomic update never allocates or locks. Regressions and movement
484    /// beyond a known target are rejected without modifying the counter.
485    pub fn set_iteration(&self, iteration: u64) -> Result<(), ReportingError> {
486        if let Some(target) = self.target_iteration().filter(|target| iteration > *target) {
487            return Err(ReportingError::IterationBeyondTarget {
488                identity: self.identity().label().to_owned(),
489                iteration,
490                target,
491            });
492        }
493        let previous = self.slot.current.fetch_max(iteration, Ordering::Relaxed);
494        if iteration < previous {
495            return Err(ReportingError::IterationRegressed {
496                identity: self.identity().label().to_owned(),
497                current: previous,
498                attempted: iteration,
499            });
500        }
501        Ok(())
502    }
503
504    /// Updates one infrequent human-readable phase such as `evolving` or
505    /// `validating`. Phase updates may lock; iteration updates do not.
506    pub fn set_phase(&self, phase: impl Into<String>) {
507        *lock(&self.slot.phase) = phase.into().into_boxed_str();
508    }
509
510    /// Sends one task-scoped message through the sole renderer.
511    pub fn report(&self, message: impl Into<String>) -> Result<(), ReportingError> {
512        self.events
513            .send(RenderEvent::TaskMessage {
514                identity: self.identity().label().to_owned(),
515                message: message.into(),
516            })
517            .map_err(|_| ReportingError::RendererUnavailable)
518    }
519
520    /// Marks the complete task workflow successful and consumes this handle.
521    pub fn complete(mut self) -> Result<(), ReportingError> {
522        if let Some(target) = self.target_iteration() {
523            let current = self.current_iteration();
524            if current != target {
525                *lock(&self.slot.phase) = "target not reached".into();
526                self.slot
527                    .status
528                    .store(TaskStatus::Failed.encode(), Ordering::Release);
529                self.active = false;
530                return Err(ReportingError::TargetIterationNotReached {
531                    identity: self.identity().label().to_owned(),
532                    current,
533                    target,
534                });
535            }
536        }
537        *lock(&self.slot.phase) = "completed".into();
538        self.slot
539            .status
540            .store(TaskStatus::Completed.encode(), Ordering::Release);
541        self.active = false;
542        Ok(())
543    }
544
545    /// Marks the task failed, records a concise phase, and consumes the handle.
546    pub fn fail(mut self, reason: impl Into<String>) {
547        *lock(&self.slot.phase) = reason.into().into_boxed_str();
548        self.slot
549            .status
550            .store(TaskStatus::Failed.encode(), Ordering::Release);
551        self.active = false;
552    }
553}
554
555impl fmt::Debug for TaskProgress {
556    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
557        formatter
558            .debug_struct("TaskProgress")
559            .field("identity", &self.identity().label())
560            .field("current_iteration", &self.current_iteration())
561            .field("target_iteration", &self.target_iteration())
562            .field("active", &self.active)
563            .finish_non_exhaustive()
564    }
565}
566
567impl Drop for TaskProgress {
568    fn drop(&mut self) {
569        if self.active {
570            *lock(&self.slot.phase) = "interrupted".into();
571            self.slot
572                .status
573                .store(TaskStatus::Failed.encode(), Ordering::Release);
574        }
575    }
576}
577
578struct ReporterInner {
579    slots: Arc<[Arc<ProgressSlot>]>,
580    identity_keys: Arc<[Box<str>]>,
581    events: Sender<RenderEvent>,
582}
583
584struct ProgressSlot {
585    identity: TaskIdentity,
586    current: AtomicU64,
587    target: AtomicU64,
588    target_known: AtomicBool,
589    status: AtomicU8,
590    phase: Mutex<Box<str>>,
591}
592
593enum RenderEvent {
594    Message(String),
595    TaskMessage { identity: String, message: String },
596    Stop { success: bool, message: String },
597}
598
599struct TerminalLease;
600
601impl Drop for TerminalLease {
602    fn drop(&mut self) {
603        TERMINAL_OWNED.store(false, Ordering::Release);
604    }
605}
606
607fn acquire_terminal() -> Result<(), ReportingError> {
608    TERMINAL_OWNED
609        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
610        .map(|_| ())
611        .map_err(|_| ReportingError::TerminalAlreadyOwned)
612}
613
614fn validate_identity_keys(
615    configuration: &ProjectConfig,
616    requested: Option<Vec<String>>,
617) -> Result<Vec<String>, ReportingError> {
618    let keys = requested.unwrap_or_else(|| {
619        configuration
620            .parameters()
621            .sweep_keys()
622            .map(str::to_owned)
623            .collect()
624    });
625    let mut seen = HashSet::with_capacity(keys.len());
626    for key in &keys {
627        if !seen.insert(key.as_str()) {
628            return Err(ReportingError::DuplicateIdentityParameter { key: key.clone() });
629        }
630        if !configuration.parameters().contains_parameter(key) {
631            return Err(ReportingError::UnknownIdentityParameter { key: key.clone() });
632        }
633    }
634    Ok(keys)
635}
636
637fn build_slots(
638    configuration: &ProjectConfig,
639    keys: Arc<[Box<str>]>,
640) -> Result<Arc<[Arc<ProgressSlot>]>, ReportingError> {
641    let capacity = usize::try_from(configuration.task_count()).map_err(|_| {
642        ReportingError::TaskCountTooLarge {
643            task_count: configuration.task_count(),
644        }
645    })?;
646    let mut slots = Vec::with_capacity(capacity);
647    let mut identities = HashMap::<String, u64>::with_capacity(capacity);
648    for task in configuration.task_configs() {
649        let label = render_identity(&task, &keys);
650        if let Some(first_ordinal) = identities.insert(label.clone(), task.task_ordinal()) {
651            return Err(ReportingError::NonUniqueTaskIdentity {
652                identity: label,
653                first_ordinal,
654                second_ordinal: task.task_ordinal(),
655            });
656        }
657        slots.push(Arc::new(ProgressSlot {
658            identity: TaskIdentity {
659                task,
660                keys: Arc::clone(&keys),
661                label: label.into(),
662            },
663            current: AtomicU64::new(0),
664            target: AtomicU64::new(0),
665            target_known: AtomicBool::new(false),
666            status: AtomicU8::new(TaskStatus::Pending.encode()),
667            phase: Mutex::new("pending".into()),
668        }));
669    }
670    Ok(slots.into())
671}
672
673fn render_identity(task: &TaskConfig, keys: &[Box<str>]) -> String {
674    if keys.is_empty() {
675        return "task".to_owned();
676    }
677    keys.iter()
678        .map(|key| {
679            let value = task
680                .value(key)
681                .expect("validated identity keys resolve for every task");
682            let value = serde_json::to_string(value)
683                .expect("serde_json::Value always serializes to valid JSON");
684            format!("{key}={value}")
685        })
686        .collect::<Vec<_>>()
687        .join(", ")
688}
689
690fn summarize(slots: &[Arc<ProgressSlot>]) -> ProgressSummary {
691    let mut summary = ProgressSummary {
692        total: u64::try_from(slots.len()).expect("slot count originated from a u64 task count"),
693        pending: 0,
694        running: 0,
695        completed: 0,
696        failed: 0,
697    };
698    for slot in slots {
699        match TaskStatus::decode(slot.status.load(Ordering::Acquire)) {
700            TaskStatus::Pending => summary.pending += 1,
701            TaskStatus::Running => summary.running += 1,
702            TaskStatus::Completed => summary.completed += 1,
703            TaskStatus::Failed => summary.failed += 1,
704        }
705    }
706    summary
707}
708
709fn render(
710    receiver: Receiver<RenderEvent>,
711    slots: Arc<[Arc<ProgressSlot>]>,
712    output: OutputMode,
713    _lease: TerminalLease,
714) {
715    // Clearing happens inside the renderer after the exclusive lease has been
716    // transferred to it. No application or worker thread writes to the
717    // terminal during the reporting session.
718    if output == OutputMode::Terminal {
719        let _ = Term::stderr().clear_screen();
720    }
721    let mut terminal = (output == OutputMode::Terminal).then(|| TerminalDisplay::new(&slots));
722    let mut last_statuses = vec![TaskStatus::Pending; slots.len()];
723    loop {
724        if let Some(display) = &mut terminal {
725            display.refresh(&slots);
726        }
727        match receiver.recv_timeout(REFRESH_INTERVAL) {
728            Ok(RenderEvent::Message(message)) => write_message(output, terminal.as_ref(), &message),
729            Ok(RenderEvent::TaskMessage { identity, message }) => {
730                write_message(output, terminal.as_ref(), &format!("{identity}: {message}"));
731            }
732            Ok(RenderEvent::Stop { success, message }) => {
733                if let Some(display) = &mut terminal {
734                    display.refresh(&slots);
735                    display.finish(&slots);
736                }
737                if output == OutputMode::Plain {
738                    write_plain_transitions(&slots, &mut last_statuses);
739                }
740                write_final(output, &slots, success, &message);
741                break;
742            }
743            Err(RecvTimeoutError::Timeout) => {
744                if output == OutputMode::Plain {
745                    write_plain_transitions(&slots, &mut last_statuses);
746                }
747            }
748            Err(RecvTimeoutError::Disconnected) => break,
749        }
750    }
751}
752
753struct TerminalDisplay {
754    multi: MultiProgress,
755    bars: Vec<ProgressBar>,
756    statuses: Vec<TaskStatus>,
757    known_style: ProgressStyle,
758    unknown_style: ProgressStyle,
759}
760
761impl TerminalDisplay {
762    fn new(slots: &[Arc<ProgressSlot>]) -> Self {
763        let multi = MultiProgress::with_draw_target(ProgressDrawTarget::stderr());
764        let known_style = ProgressStyle::with_template(
765            "{prefix:.bold} [{msg}] {wide_bar:.cyan/blue} {pos}/{len} elapsed {elapsed_precise} ETA {eta_precise}",
766        )
767        .expect("hard-coded progress template is valid");
768        let unknown_style = ProgressStyle::with_template(
769            "{prefix:.bold} [{msg}] {spinner:.cyan} iteration {pos} elapsed {elapsed_precise} ETA unknown",
770        )
771        .expect("hard-coded spinner template is valid");
772        let bars: Vec<_> = slots
773            .iter()
774            .map(|slot| {
775                let bar = multi.add(ProgressBar::new_spinner());
776                bar.set_prefix(slot.identity.label().to_owned());
777                bar.set_style(unknown_style.clone());
778                bar.set_message("pending");
779                bar
780            })
781            .collect();
782
783        // MultiProgress throttles draws globally. Force each bar's initial
784        // state once so pending tasks are materialized instead of being
785        // starved by earlier rows updated in the same renderer pass.
786        for bar in &bars {
787            bar.force_draw();
788        }
789
790        Self {
791            multi,
792            bars,
793            statuses: vec![TaskStatus::Pending; slots.len()],
794            known_style,
795            unknown_style,
796        }
797    }
798
799    fn refresh(&mut self, slots: &[Arc<ProgressSlot>]) {
800        for ((bar, previous_status), slot) in self.bars.iter().zip(&mut self.statuses).zip(slots) {
801            if !slot.target_known.load(Ordering::Acquire) {
802                bar.set_style(self.unknown_style.clone());
803            } else {
804                let target = slot.target.load(Ordering::Relaxed);
805                bar.set_style(self.known_style.clone());
806                bar.set_length(target);
807            }
808            bar.set_position(slot.current.load(Ordering::Relaxed));
809            let status = TaskStatus::decode(slot.status.load(Ordering::Acquire));
810            if *previous_status == TaskStatus::Pending && status == TaskStatus::Running {
811                // Pending time is queueing time, not task execution time. Start
812                // elapsed and ETA measurement from the first running state.
813                bar.reset_elapsed();
814            }
815            let phase = lock(&slot.phase);
816            if phase.is_empty() || phase.as_ref() == status.label() {
817                bar.set_message(status.label());
818            } else {
819                bar.set_message(format!("{}: {}", status.label(), phase.as_ref()));
820            }
821            bar.tick();
822            *previous_status = status;
823        }
824    }
825
826    fn finish(&self, slots: &[Arc<ProgressSlot>]) {
827        for (bar, slot) in self.bars.iter().zip(slots) {
828            let status = TaskStatus::decode(slot.status.load(Ordering::Acquire));
829            bar.finish_with_message(status.label());
830        }
831        let _ = self.multi.clear();
832    }
833}
834
835fn write_message(output: OutputMode, terminal: Option<&TerminalDisplay>, message: &str) {
836    match output {
837        OutputMode::Terminal => {
838            if let Some(display) = terminal {
839                let _ = display.multi.println(message);
840            }
841        }
842        OutputMode::Plain => eprintln!("[progress] {message}"),
843        OutputMode::Hidden | OutputMode::Auto => {}
844    }
845}
846
847fn write_plain_transitions(slots: &[Arc<ProgressSlot>], previous: &mut [TaskStatus]) {
848    for (slot, old) in slots.iter().zip(previous) {
849        let status = TaskStatus::decode(slot.status.load(Ordering::Acquire));
850        if status != *old {
851            let phase = lock(&slot.phase);
852            eprintln!(
853                "[task] identity={} status={} phase={} iteration={} target={}",
854                slot.identity.label(),
855                status.label(),
856                phase.as_ref(),
857                slot.current.load(Ordering::Relaxed),
858                format_target(slot)
859            );
860            *old = status;
861        }
862    }
863}
864
865fn write_final(output: OutputMode, slots: &[Arc<ProgressSlot>], success: bool, message: &str) {
866    if output == OutputMode::Hidden {
867        return;
868    }
869    let summary = summarize(slots);
870    eprintln!(
871        "[workflow] status={} tasks={} completed={} failed={} pending={} message={}",
872        if success { "completed" } else { "failed" },
873        summary.total,
874        summary.completed,
875        summary.failed,
876        summary.pending,
877        message
878    );
879}
880
881fn format_target(slot: &ProgressSlot) -> String {
882    if slot.target_known.load(Ordering::Acquire) {
883        slot.target.load(Ordering::Relaxed).to_string()
884    } else {
885        "unknown".to_owned()
886    }
887}
888
889fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
890    mutex
891        .lock()
892        .unwrap_or_else(std::sync::PoisonError::into_inner)
893}
scientific_workflow/reporting/progress.rs

scientific_workflow/reporting/
progress.rs
