scientific_workflow/study/
renderer.rs

1//! Private thread-safe task progress state and 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::HashSet;
9use std::fmt;
10use std::io::{self, IsTerminal};
11use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};
12use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender};
13use std::sync::{Arc, Mutex};
14use std::thread::{self, JoinHandle};
15use std::time::Duration;
16
17use serde_json::Value;
18
19use super::command::StudyCommand;
20use super::error::StudyError;
21use super::phase::{Phase, TaskKey, TaskMode};
22use super::tui::{RenderSnapshot, TaskView, TerminalUi};
23
24const REFRESH_INTERVAL: Duration = Duration::from_millis(100);
25const MESSAGE_CAPACITY: usize = 256;
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 [`TaskProgressHandle`] 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    /// The task cooperatively stopped after cancellation was requested.
41    Cancelled,
42    /// The task was never started because its phase stopped admitting work.
43    Skipped,
44}
45
46impl TaskStatus {
47    /// Returns the stable uncolored lifecycle label used by logs and APIs.
48    pub const fn as_str(self) -> &'static str {
49        match self {
50            Self::Pending => "pending",
51            Self::Running => "running",
52            Self::Completed => "completed",
53            Self::Failed => "failed",
54            Self::Cancelled => "cancelled",
55            Self::Skipped => "skipped",
56        }
57    }
58
59    fn encode(self) -> u8 {
60        match self {
61            Self::Pending => 0,
62            Self::Running => 1,
63            Self::Completed => 2,
64            Self::Failed => 3,
65            Self::Cancelled => 4,
66            Self::Skipped => 5,
67        }
68    }
69
70    fn decode(value: u8) -> Self {
71        match value {
72            0 => Self::Pending,
73            1 => Self::Running,
74            2 => Self::Completed,
75            3 => Self::Failed,
76            4 => Self::Cancelled,
77            5 => Self::Skipped,
78            _ => unreachable!("task status is written only through TaskStatus::encode"),
79        }
80    }
81
82    fn label(self) -> &'static str {
83        self.as_str()
84    }
85}
86
87/// Exact parameter-derived identity of one task.
88///
89/// The identity contains only the caller-selected parameter fields. Its label
90/// is deterministic compact JSON text intended for reporting, while equality
91/// validation uses the retained JSON values themselves.
92#[derive(Clone, Debug)]
93pub struct TaskIdentity {
94    label: Arc<str>,
95    key: TaskKey,
96    metadata: Arc<std::collections::BTreeMap<String, Value>>,
97}
98
99impl TaskIdentity {
100    /// Returns the terminal label derived from exact parameter key/value pairs.
101    pub fn label(&self) -> &str {
102        &self.label
103    }
104
105    /// Returns the number of parameter fields forming this identity.
106    pub fn len(&self) -> usize {
107        self.metadata.len()
108    }
109
110    /// Reports whether the identity contains no parameter fields.
111    pub fn is_empty(&self) -> bool {
112        self.metadata.is_empty()
113    }
114
115    /// Borrows one exact identity value by parameter name.
116    pub fn value(&self, key: &str) -> Option<&Value> {
117        self.metadata.get(key)
118    }
119
120    /// Iterates identity fields in the configured display order.
121    pub fn iter(&self) -> Box<dyn Iterator<Item = (&str, &Value)> + '_> {
122        Box::new(
123            self.metadata
124                .iter()
125                .map(|(key, value)| (key.as_str(), value)),
126        )
127    }
128
129    /// Returns the exact first-class task key when this identity came from a
130    /// phase declaration.
131    pub fn task_key(&self) -> &TaskKey {
132        &self.key
133    }
134}
135
136/// Immutable aggregate captured when centralized reporting ends.
137#[derive(Clone, Debug, Eq, PartialEq)]
138pub struct ProgressSummary {
139    total: u64,
140    pending: u64,
141    running: u64,
142    completed: u64,
143    failed: u64,
144    cancelled: u64,
145    skipped: u64,
146}
147
148impl ProgressSummary {
149    /// Returns the number of registered tasks.
150    pub fn total(&self) -> u64 {
151        self.total
152    }
153
154    /// Returns the number of tasks that never started.
155    pub fn pending(&self) -> u64 {
156        self.pending
157    }
158
159    /// Returns the number of tasks still running at capture time.
160    pub fn running(&self) -> u64 {
161        self.running
162    }
163
164    /// Returns the number of successfully completed tasks.
165    pub fn completed(&self) -> u64 {
166        self.completed
167    }
168
169    /// Returns the number of failed or interrupted tasks.
170    pub fn failed(&self) -> u64 {
171        self.failed
172    }
173
174    /// Returns the number of active tasks that cooperatively cancelled.
175    pub fn cancelled(&self) -> u64 {
176        self.cancelled
177    }
178
179    /// Returns the number of tasks that were never started.
180    pub fn skipped(&self) -> u64 {
181        self.skipped
182    }
183
184    /// Reports whether every registered task completed successfully.
185    pub fn is_success(&self) -> bool {
186        self.completed == self.total
187            && self.pending == 0
188            && self.running == 0
189            && self.failed == 0
190            && self.cancelled == 0
191            && self.skipped == 0
192    }
193}
194
195/// Output policy selected before the renderer starts.
196#[derive(Clone, Copy, Debug, Eq, PartialEq)]
197enum OutputMode {
198    Auto,
199    Terminal,
200    Plain,
201    Hidden,
202}
203
204/// Builder that makes the renderer observe existing first-class phases/tasks.
205pub(crate) struct StudyRendererBuilder {
206    slots: Arc<[Arc<ProgressSlot>]>,
207    output: OutputMode,
208    cancellation: Option<CancellationToken>,
209}
210
211impl StudyRendererBuilder {
212    pub(crate) fn cancellation_token(mut self, cancellation: CancellationToken) -> Self {
213        self.cancellation = Some(cancellation);
214        self
215    }
216    /// Forces cursor-controlled isolated-screen rendering.
217    pub fn terminal(mut self) -> Self {
218        self.output = OutputMode::Terminal;
219        self
220    }
221
222    /// Forces stable line-oriented output.
223    pub fn plain(mut self) -> Self {
224        self.output = OutputMode::Plain;
225        self
226    }
227
228    /// Suppresses rendering while retaining lifecycle validation.
229    pub fn hidden(mut self) -> Self {
230        self.output = OutputMode::Hidden;
231        self
232    }
233
234    /// Validates phase uniqueness and starts one renderer.
235    pub fn start(self) -> Result<StudyRenderer, StudyError> {
236        start_renderer(self.slots, self.output, self.cancellation)
237    }
238}
239
240impl fmt::Debug for StudyRendererBuilder {
241    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
242        formatter
243            .debug_struct("StudyRendererBuilder")
244            .field("tasks", &self.slots.len())
245            .field("output", &self.output)
246            .finish_non_exhaustive()
247    }
248}
249
250/// Central progress registry and exclusive human-facing terminal owner.
251pub(crate) struct StudyRenderer {
252    inner: Arc<RendererInner>,
253    renderer: Option<JoinHandle<()>>,
254    finished: bool,
255}
256
257impl StudyRenderer {
258    /// Observes tasks already owned and identified by first-class phases.
259    pub fn for_phase(phase: &Phase, heading: &str) -> Result<StudyRendererBuilder, StudyError> {
260        Ok(StudyRendererBuilder {
261            slots: build_phase_slots(std::slice::from_ref(phase), Some(heading))?,
262            output: OutputMode::Auto,
263            cancellation: None,
264        })
265    }
266
267    /// Starts one exact first-class iterative task.
268    pub(crate) fn start_progress(
269        &self,
270        key: &TaskKey,
271        initial_iteration: u64,
272        target_iteration: Option<u64>,
273    ) -> Result<TaskProgressHandle, StudyError> {
274        let slot = managed_slot(&self.inner.slots, key)?;
275        if slot.mode != TaskMode::Progress {
276            return Err(mode_mismatch(&slot, "progress"));
277        }
278        start_slot(&self.inner, slot, initial_iteration, target_iteration)
279    }
280
281    /// Starts one exact first-class lifecycle-only one-shot task.
282    pub(crate) fn start_one_shot(&self, key: &TaskKey) -> Result<OneShotTaskHandle, StudyError> {
283        let slot = managed_slot(&self.inner.slots, key)?;
284        if slot.mode != TaskMode::OneShot {
285            return Err(mode_mismatch(&slot, "one-shot"));
286        }
287        start_one_shot_slot(&self.inner, slot)
288    }
289
290    /// Marks one exact first-class task complete through verified reuse.
291    pub fn mark_completed(&self, key: &TaskKey) -> Result<(), StudyError> {
292        let slot = managed_slot(&self.inner.slots, key)?;
293        mark_slot_completed(&slot)
294    }
295
296    pub(crate) fn mark_skipped(&self, key: &TaskKey) -> Result<(), StudyError> {
297        let slot = managed_slot(&self.inner.slots, key)?;
298        mark_pending_terminal(&slot, TaskStatus::Skipped, "skipped")
299    }
300
301    pub(crate) fn mark_cancelled(&self, key: &TaskKey) -> Result<(), StudyError> {
302        let slot = managed_slot(&self.inner.slots, key)?;
303        mark_pending_terminal(&slot, TaskStatus::Cancelled, "cancelled")
304    }
305
306    pub(crate) fn mark_delayed(&self, key: &TaskKey, rank: usize) -> Result<(), StudyError> {
307        let slot = managed_slot(&self.inner.slots, key)?;
308        if TaskStatus::decode(slot.status.load(Ordering::Acquire)) != TaskStatus::Pending {
309            return Err(StudyError::TaskAlreadyStarted {
310                identity: slot.identity.label().to_owned(),
311            });
312        }
313        *lock(&slot.detail) = format!("delayed start (rank {rank})").into_boxed_str();
314        Ok(())
315    }
316
317    pub(crate) fn request_cancellation(&self) {
318        self.inner.cancelled.store(true, Ordering::Release);
319    }
320
321    pub(crate) fn is_cancelled(&self) -> bool {
322        self.inner.cancelled.load(Ordering::Acquire)
323    }
324
325    pub(crate) fn cancellation_flag(&self) -> Arc<AtomicBool> {
326        Arc::clone(&self.inner.cancelled)
327    }
328
329    /// Returns a non-blocking snapshot of all task lifecycle counts.
330    pub fn summary(&self) -> ProgressSummary {
331        summarize(&self.inner.slots)
332    }
333
334    pub(crate) fn task_execution_snapshots(&self) -> Vec<TaskExecutionSnapshot> {
335        self.inner
336            .slots
337            .iter()
338            .map(|slot| TaskExecutionSnapshot {
339                key: slot.identity.task_key().clone(),
340                status: TaskStatus::decode(slot.status.load(Ordering::Acquire)),
341                current_iteration: (slot.mode == TaskMode::Progress)
342                    .then(|| slot.current.load(Ordering::Relaxed)),
343                target_iteration: (slot.mode == TaskMode::Progress
344                    && slot.target_known.load(Ordering::Acquire))
345                .then(|| slot.target.load(Ordering::Relaxed)),
346            })
347            .collect()
348    }
349
350    /// Finishes a successful session and emits its final summary and message.
351    ///
352    /// Every task must already be completed. The method stops and joins the
353    /// renderer and releases exclusive terminal ownership.
354    pub fn complete(mut self, message: impl Into<String>) -> Result<ProgressSummary, StudyError> {
355        let summary = self.summary();
356        if !summary.is_success() {
357            self.stop(false, "study did not complete".to_owned())?;
358            return Err(StudyError::IncompleteProgress {
359                pending: summary.pending,
360                running: summary.running,
361                failed: summary.failed,
362            });
363        }
364        self.stop(true, message.into())?;
365        Ok(summary)
366    }
367
368    /// Finishes an unsuccessful session while preserving all task statuses.
369    pub fn fail(mut self, message: impl Into<String>) -> Result<ProgressSummary, StudyError> {
370        let summary = self.summary();
371        self.stop(false, message.into())?;
372        Ok(summary)
373    }
374
375    fn stop(&mut self, success: bool, message: String) -> Result<(), StudyError> {
376        self.inner
377            .events
378            .send(RenderEvent::Stop { success, message })
379            .map_err(|_| StudyError::RendererUnavailable)?;
380        self.finished = true;
381        if self
382            .renderer
383            .take()
384            .expect("an unfinished study renderer owns one thread")
385            .join()
386            .is_err()
387        {
388            return Err(StudyError::RendererPanicked);
389        }
390        Ok(())
391    }
392}
393
394pub(crate) struct TaskExecutionSnapshot {
395    pub(crate) key: TaskKey,
396    pub(crate) status: TaskStatus,
397    pub(crate) current_iteration: Option<u64>,
398    pub(crate) target_iteration: Option<u64>,
399}
400
401impl fmt::Debug for StudyRenderer {
402    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
403        formatter
404            .debug_struct("StudyRenderer")
405            .field("tasks", &self.inner.slots.len())
406            .field("summary", &self.summary())
407            .finish_non_exhaustive()
408    }
409}
410
411impl Drop for StudyRenderer {
412    fn drop(&mut self) {
413        if self.finished {
414            return;
415        }
416        let _ = self.inner.events.send(RenderEvent::Stop {
417            success: false,
418            message: "study renderer dropped before completion".to_owned(),
419        });
420        if let Some(renderer) = self.renderer.take() {
421            let _ = renderer.join();
422        }
423    }
424}
425
426/// Non-clone task-local progress handle.
427///
428/// Dropping a running handle without calling [`TaskProgressHandle::complete`] or
429/// [`TaskProgressHandle::fail`] marks the task failed, making ordinary `?` returns
430/// safe without a separate cleanup branch.
431pub(crate) struct TaskProgressHandle {
432    slot: Arc<ProgressSlot>,
433    events: SyncSender<RenderEvent>,
434    cancelled: Arc<AtomicBool>,
435    active: bool,
436}
437
438/// Shared, cheap cancellation observation for embedded schedulers and tasks.
439#[derive(Clone, Debug)]
440pub struct CancellationToken(Arc<AtomicBool>);
441
442impl CancellationToken {
443    pub(crate) fn new() -> Self {
444        Self(Arc::new(AtomicBool::new(false)))
445    }
446
447    pub(crate) fn shared(&self) -> Arc<AtomicBool> {
448        Arc::clone(&self.0)
449    }
450
451    /// Reports whether interactive Ctrl-C requested termination.
452    pub fn is_cancelled(&self) -> bool {
453        self.0.load(Ordering::Acquire)
454    }
455
456    /// Requests cooperative termination.
457    pub fn cancel(&self) {
458        self.0.store(true, Ordering::Release);
459    }
460}
461
462impl TaskProgressHandle {
463    /// Reports whether the owning renderer requested cooperative termination.
464    pub fn is_cancelled(&self) -> bool {
465        self.cancelled.load(Ordering::Acquire)
466    }
467
468    /// Borrows the exact parameter-derived task identity.
469    pub fn identity(&self) -> &TaskIdentity {
470        &self.slot.identity
471    }
472
473    /// Returns the latest reported absolute simulation iteration.
474    pub fn current_iteration(&self) -> u64 {
475        self.slot.current.load(Ordering::Relaxed)
476    }
477
478    /// Returns the known absolute target iteration, if one exists.
479    pub fn target_iteration(&self) -> Option<u64> {
480        if self.slot.target_known.load(Ordering::Acquire) {
481            Some(self.slot.target.load(Ordering::Relaxed))
482        } else {
483            None
484        }
485    }
486
487    /// Sets or replaces the absolute target for this running task.
488    pub fn set_target_iteration(&self, target: u64) -> Result<(), StudyError> {
489        let current = self.current_iteration();
490        if current > target {
491            return Err(StudyError::InitialIterationBeyondTarget {
492                identity: self.identity().label().to_owned(),
493                initial: current,
494                target,
495            });
496        }
497        self.slot.target.store(target, Ordering::Relaxed);
498        self.slot.target_known.store(true, Ordering::Release);
499        Ok(())
500    }
501
502    /// Synchronizes progress to an authoritative absolute simulation iteration.
503    ///
504    /// The atomic update never allocates or locks. Regressions and movement
505    /// beyond a known target are rejected without modifying the counter.
506    pub fn set_iteration(&self, iteration: u64) -> Result<(), StudyError> {
507        if let Some(target) = self.target_iteration().filter(|target| iteration > *target) {
508            return Err(StudyError::IterationBeyondTarget {
509                identity: self.identity().label().to_owned(),
510                iteration,
511                target,
512            });
513        }
514        let previous = self.slot.current.fetch_max(iteration, Ordering::Relaxed);
515        if iteration < previous {
516            return Err(StudyError::IterationRegressed {
517                identity: self.identity().label().to_owned(),
518                current: previous,
519                attempted: iteration,
520            });
521        }
522        Ok(())
523    }
524
525    /// Synchronizes the authoritative iteration and applies the configured
526    /// continuation target.
527    ///
528    /// An indeterminate task returns `true`. A task with a target returns
529    /// `true` below it and `false` exactly at it. Movement beyond the target or
530    /// backwards is rejected through the same validation as [`Self::set_iteration`].
531    pub fn should_continue(&self, iteration: u64) -> Result<bool, StudyError> {
532        self.set_iteration(iteration)?;
533        Ok(!self.is_cancelled()
534            && self
535                .target_iteration()
536                .is_none_or(|target| iteration < target))
537    }
538
539    /// Updates one infrequent human-readable detail such as `evolving` or
540    /// `validating`. Detail updates may lock; iteration updates do not.
541    pub fn set_detail(&self, detail: impl Into<String>) {
542        *lock(&self.slot.detail) = detail.into().into_boxed_str();
543    }
544
545    /// Sends one task-scoped message through the sole renderer.
546    pub fn report(&self, message: impl Into<String>) -> Result<(), StudyError> {
547        self.events
548            .send(RenderEvent::TaskMessage {
549                identity: self.identity().label().to_owned(),
550                message: message.into(),
551            })
552            .map_err(|_| StudyError::RendererUnavailable)
553    }
554
555    /// Marks the task successful and consumes this handle.
556    ///
557    /// `reason == None` means the configured target must have been reached.
558    /// `Some(reason)` records an intentional scientific early-completion reason
559    /// and permits completion before that generic target.
560    pub fn complete(mut self, reason: Option<String>) -> Result<(), StudyError> {
561        if reason.is_none()
562            && let Some(target) = self.target_iteration()
563        {
564            let current = self.current_iteration();
565            if current != target {
566                *lock(&self.slot.detail) = "target not reached".into();
567                self.slot
568                    .status
569                    .store(TaskStatus::Failed.encode(), Ordering::Release);
570                self.active = false;
571                return Err(StudyError::TargetIterationNotReached {
572                    identity: self.identity().label().to_owned(),
573                    current,
574                    target,
575                });
576            }
577        }
578        *lock(&self.slot.detail) = reason
579            .unwrap_or_else(|| "completed".to_owned())
580            .into_boxed_str();
581        self.slot
582            .status
583            .store(TaskStatus::Completed.encode(), Ordering::Release);
584        self.active = false;
585        Ok(())
586    }
587
588    /// Marks the task failed, records a concise detail, and consumes the handle.
589    pub fn fail(mut self, reason: impl Into<String>) {
590        *lock(&self.slot.detail) = reason.into().into_boxed_str();
591        self.slot
592            .status
593            .store(TaskStatus::Failed.encode(), Ordering::Release);
594        self.active = false;
595    }
596
597    pub(crate) fn cancel(mut self, reason: impl Into<String>) {
598        *lock(&self.slot.detail) = reason.into().into_boxed_str();
599        self.slot
600            .status
601            .store(TaskStatus::Cancelled.encode(), Ordering::Release);
602        self.active = false;
603    }
604}
605
606impl fmt::Debug for TaskProgressHandle {
607    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
608        formatter
609            .debug_struct("TaskProgressHandle")
610            .field("identity", &self.identity().label())
611            .field("current_iteration", &self.current_iteration())
612            .field("target_iteration", &self.target_iteration())
613            .field("active", &self.active)
614            .finish_non_exhaustive()
615    }
616}
617
618impl Drop for TaskProgressHandle {
619    fn drop(&mut self) {
620        if self.active {
621            *lock(&self.slot.detail) = "interrupted".into();
622            self.slot
623                .status
624                .store(TaskStatus::Failed.encode(), Ordering::Release);
625        }
626    }
627}
628
629/// Non-clone task-local handle for lifecycle-only work.
630///
631/// One-shot tasks deliberately expose no iteration or target operations. Dropping
632/// an active handle marks only its reporting task failed.
633pub(crate) struct OneShotTaskHandle {
634    slot: Arc<ProgressSlot>,
635    events: SyncSender<RenderEvent>,
636    cancelled: Arc<AtomicBool>,
637    active: bool,
638}
639
640impl OneShotTaskHandle {
641    /// Reports whether the owning renderer requested cooperative termination.
642    pub fn is_cancelled(&self) -> bool {
643        self.cancelled.load(Ordering::Acquire)
644    }
645
646    /// Borrows the task identity supplied by the owning phase.
647    pub fn identity(&self) -> &TaskIdentity {
648        &self.slot.identity
649    }
650
651    /// Returns the current lifecycle status.
652    pub fn status(&self) -> TaskStatus {
653        TaskStatus::decode(self.slot.status.load(Ordering::Acquire))
654    }
655
656    /// Updates one infrequent human-readable execution detail.
657    pub fn set_detail(&self, detail: impl Into<String>) {
658        *lock(&self.slot.detail) = detail.into().into_boxed_str();
659    }
660
661    /// Sends one task-scoped message through the sole renderer.
662    pub fn report(&self, message: impl Into<String>) -> Result<(), StudyError> {
663        self.events
664            .send(RenderEvent::TaskMessage {
665                identity: self.identity().label().to_owned(),
666                message: message.into(),
667            })
668            .map_err(|_| StudyError::RendererUnavailable)
669    }
670
671    /// Marks this one-shot successful and consumes its handle.
672    pub fn complete(mut self) {
673        *lock(&self.slot.detail) = "completed".into();
674        self.slot
675            .status
676            .store(TaskStatus::Completed.encode(), Ordering::Release);
677        self.active = false;
678    }
679
680    /// Marks this one-shot failed and consumes its handle.
681    pub fn fail(mut self, reason: impl Into<String>) {
682        *lock(&self.slot.detail) = reason.into().into_boxed_str();
683        self.slot
684            .status
685            .store(TaskStatus::Failed.encode(), Ordering::Release);
686        self.active = false;
687    }
688
689    pub(crate) fn cancel(mut self, reason: impl Into<String>) {
690        *lock(&self.slot.detail) = reason.into().into_boxed_str();
691        self.slot
692            .status
693            .store(TaskStatus::Cancelled.encode(), Ordering::Release);
694        self.active = false;
695    }
696}
697
698impl fmt::Debug for OneShotTaskHandle {
699    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
700        formatter
701            .debug_struct("OneShotTaskHandle")
702            .field("identity", &self.identity().label())
703            .field("status", &self.status())
704            .field("active", &self.active)
705            .finish_non_exhaustive()
706    }
707}
708
709impl Drop for OneShotTaskHandle {
710    fn drop(&mut self) {
711        if self.active {
712            *lock(&self.slot.detail) = "interrupted".into();
713            self.slot
714                .status
715                .store(TaskStatus::Failed.encode(), Ordering::Release);
716        }
717    }
718}
719
720struct RendererInner {
721    slots: Arc<[Arc<ProgressSlot>]>,
722    events: SyncSender<RenderEvent>,
723    cancelled: Arc<AtomicBool>,
724}
725
726struct ProgressSlot {
727    identity: TaskIdentity,
728    phase_label: Option<Arc<str>>,
729    mode: TaskMode,
730    current: AtomicU64,
731    target: AtomicU64,
732    target_known: AtomicBool,
733    started: AtomicBool,
734    status: AtomicU8,
735    detail: Mutex<Box<str>>,
736}
737
738enum RenderEvent {
739    TaskMessage { identity: String, message: String },
740    Stop { success: bool, message: String },
741}
742
743struct TerminalLease;
744
745impl Drop for TerminalLease {
746    fn drop(&mut self) {
747        TERMINAL_OWNED.store(false, Ordering::Release);
748    }
749}
750
751fn acquire_terminal() -> Result<(), StudyError> {
752    TERMINAL_OWNED
753        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
754        .map(|_| ())
755        .map_err(|_| StudyError::TerminalAlreadyOwned)
756}
757
758fn resolve_output(output: OutputMode) -> OutputMode {
759    match output {
760        OutputMode::Auto if io::stderr().is_terminal() && io::stdin().is_terminal() => {
761            OutputMode::Terminal
762        }
763        OutputMode::Auto => OutputMode::Plain,
764        explicit => explicit,
765    }
766}
767
768fn start_renderer(
769    slots: Arc<[Arc<ProgressSlot>]>,
770    requested_output: OutputMode,
771    cancellation: Option<CancellationToken>,
772) -> Result<StudyRenderer, StudyError> {
773    acquire_terminal()?;
774    let lease = TerminalLease;
775    let cancelled = cancellation
776        .map(|token| token.shared())
777        .unwrap_or_else(|| Arc::new(AtomicBool::new(false)));
778    let mut output = resolve_output(requested_output);
779    let terminal = if output == OutputMode::Terminal {
780        match TerminalUi::enter(slots.len()) {
781            Ok(terminal) => Some(terminal),
782            Err(_) if requested_output == OutputMode::Auto => {
783                output = OutputMode::Plain;
784                None
785            }
786            Err(error) => {
787                return Err(StudyError::TerminalSetup {
788                    operation: error.operation,
789                    source: error.source,
790                });
791            }
792        }
793    } else {
794        None
795    };
796    let (events, receiver) = mpsc::sync_channel(MESSAGE_CAPACITY);
797    let renderer_slots = Arc::clone(&slots);
798    let renderer_cancelled = Arc::clone(&cancelled);
799    let renderer = match thread::Builder::new()
800        .name("scientific-workflow-progress".to_owned())
801        .spawn(move || {
802            render(
803                receiver,
804                renderer_slots,
805                output,
806                terminal,
807                renderer_cancelled,
808                lease,
809            )
810        }) {
811        Ok(renderer) => renderer,
812        Err(source) => return Err(StudyError::StartRenderer { source }),
813    };
814    Ok(StudyRenderer {
815        inner: Arc::new(RendererInner {
816            slots,
817            events,
818            cancelled,
819        }),
820        renderer: Some(renderer),
821        finished: false,
822    })
823}
824
825fn managed_slot(
826    slots: &[Arc<ProgressSlot>],
827    key: &TaskKey,
828) -> Result<Arc<ProgressSlot>, StudyError> {
829    slots
830        .iter()
831        .find(|slot| slot.identity.task_key() == key)
832        .cloned()
833        .ok_or_else(|| StudyError::UnknownTask {
834            task: key.to_string(),
835        })
836}
837
838fn mode_mismatch(slot: &ProgressSlot, requested: &'static str) -> StudyError {
839    StudyError::TaskModeMismatch {
840        task: slot.identity.task_key().to_string(),
841        requested,
842        actual: match slot.mode {
843            TaskMode::Progress => "progress",
844            TaskMode::OneShot => "one-shot",
845        },
846    }
847}
848
849fn mark_slot_completed(slot: &ProgressSlot) -> Result<(), StudyError> {
850    slot.status
851        .compare_exchange(
852            TaskStatus::Pending.encode(),
853            TaskStatus::Completed.encode(),
854            Ordering::AcqRel,
855            Ordering::Acquire,
856        )
857        .map_err(|_| StudyError::TaskAlreadyStarted {
858            identity: slot.identity.label().to_owned(),
859        })?;
860    *lock(&slot.detail) = "already completed".into();
861    Ok(())
862}
863
864fn mark_pending_terminal(
865    slot: &ProgressSlot,
866    status: TaskStatus,
867    detail: &'static str,
868) -> Result<(), StudyError> {
869    slot.status
870        .compare_exchange(
871            TaskStatus::Pending.encode(),
872            status.encode(),
873            Ordering::AcqRel,
874            Ordering::Acquire,
875        )
876        .map_err(|_| StudyError::TaskAlreadyStarted {
877            identity: slot.identity.label().to_owned(),
878        })?;
879    *lock(&slot.detail) = detail.into();
880    Ok(())
881}
882
883fn start_slot(
884    inner: &RendererInner,
885    slot: Arc<ProgressSlot>,
886    initial_iteration: u64,
887    target_iteration: Option<u64>,
888) -> Result<TaskProgressHandle, StudyError> {
889    if let Some(target) = target_iteration.filter(|target| initial_iteration > *target) {
890        return Err(StudyError::InitialIterationBeyondTarget {
891            identity: slot.identity.label().to_owned(),
892            initial: initial_iteration,
893            target,
894        });
895    }
896    slot.status
897        .compare_exchange(
898            TaskStatus::Pending.encode(),
899            TaskStatus::Running.encode(),
900            Ordering::AcqRel,
901            Ordering::Acquire,
902        )
903        .map_err(|_| StudyError::TaskAlreadyStarted {
904            identity: slot.identity.label().to_owned(),
905        })?;
906    slot.started.store(true, Ordering::Release);
907    slot.current.store(initial_iteration, Ordering::Relaxed);
908    if let Some(target) = target_iteration {
909        slot.target.store(target, Ordering::Relaxed);
910        slot.target_known.store(true, Ordering::Release);
911    } else {
912        slot.target_known.store(false, Ordering::Release);
913    }
914    *lock(&slot.detail) = "running".into();
915    Ok(TaskProgressHandle {
916        slot,
917        events: inner.events.clone(),
918        cancelled: Arc::clone(&inner.cancelled),
919        active: true,
920    })
921}
922
923fn start_one_shot_slot(
924    inner: &RendererInner,
925    slot: Arc<ProgressSlot>,
926) -> Result<OneShotTaskHandle, StudyError> {
927    slot.status
928        .compare_exchange(
929            TaskStatus::Pending.encode(),
930            TaskStatus::Running.encode(),
931            Ordering::AcqRel,
932            Ordering::Acquire,
933        )
934        .map_err(|_| StudyError::TaskAlreadyStarted {
935            identity: slot.identity.label().to_owned(),
936        })?;
937    slot.started.store(true, Ordering::Release);
938    slot.target_known.store(false, Ordering::Release);
939    *lock(&slot.detail) = "running".into();
940    Ok(OneShotTaskHandle {
941        slot,
942        events: inner.events.clone(),
943        cancelled: Arc::clone(&inner.cancelled),
944        active: true,
945    })
946}
947
948fn build_phase_slots(
949    phases: &[Phase],
950    heading: Option<&str>,
951) -> Result<Arc<[Arc<ProgressSlot>]>, StudyError> {
952    if phases.is_empty() {
953        return Err(StudyError::EmptyPhaseSet);
954    }
955    let mut phase_ids = HashSet::with_capacity(phases.len());
956    let capacity = phases.iter().map(|phase| phase.tasks().len()).sum();
957    let mut slots = Vec::with_capacity(capacity);
958    for phase in phases {
959        if !phase_ids.insert(phase.id()) {
960            return Err(StudyError::DuplicatePhaseId {
961                phase: phase.id().get(),
962            });
963        }
964        let phase_label: Arc<str> = heading.unwrap_or_else(|| phase.label()).into();
965        for task in phase.tasks() {
966            slots.push(Arc::new(ProgressSlot {
967                identity: TaskIdentity {
968                    label: task.label().into(),
969                    key: task.key().clone(),
970                    metadata: task.metadata_map(),
971                },
972                phase_label: Some(Arc::clone(&phase_label)),
973                mode: task.mode(),
974                current: AtomicU64::new(0),
975                target: AtomicU64::new(0),
976                target_known: AtomicBool::new(false),
977                started: AtomicBool::new(false),
978                status: AtomicU8::new(TaskStatus::Pending.encode()),
979                detail: Mutex::new("pending".into()),
980            }));
981        }
982    }
983    Ok(slots.into())
984}
985
986fn summarize(slots: &[Arc<ProgressSlot>]) -> ProgressSummary {
987    let mut summary = ProgressSummary {
988        total: u64::try_from(slots.len()).expect("slot count originated from a u64 task count"),
989        pending: 0,
990        running: 0,
991        completed: 0,
992        failed: 0,
993        cancelled: 0,
994        skipped: 0,
995    };
996    for slot in slots {
997        match TaskStatus::decode(slot.status.load(Ordering::Acquire)) {
998            TaskStatus::Pending => summary.pending += 1,
999            TaskStatus::Running => summary.running += 1,
1000            TaskStatus::Completed => summary.completed += 1,
1001            TaskStatus::Failed => summary.failed += 1,
1002            TaskStatus::Cancelled => summary.cancelled += 1,
1003            TaskStatus::Skipped => summary.skipped += 1,
1004        }
1005    }
1006    summary
1007}
1008
1009fn render(
1010    receiver: Receiver<RenderEvent>,
1011    slots: Arc<[Arc<ProgressSlot>]>,
1012    output: OutputMode,
1013    mut terminal: Option<TerminalUi>,
1014    cancelled: Arc<AtomicBool>,
1015    _lease: TerminalLease,
1016) {
1017    let mut last_statuses = vec![TaskStatus::Pending; slots.len()];
1018    let mut input_failed = false;
1019    loop {
1020        if let Some(display) = &mut terminal {
1021            if !input_failed {
1022                match display.poll_command() {
1023                    Ok(Some(StudyCommand::Exit)) => {
1024                        cancelled.store(true, Ordering::Release);
1025                        display.mark_exit_requested();
1026                    }
1027                    Ok(None) => {}
1028                    Err(error) => {
1029                        cancelled.store(true, Ordering::Release);
1030                        display.push_message(format!("study: terminal input failed: {error}"));
1031                        display.mark_exit_requested();
1032                        input_failed = true;
1033                    }
1034                }
1035            }
1036            let snapshot = render_snapshot(&slots);
1037            let _ = display.draw(&snapshot);
1038        }
1039        match receiver.recv_timeout(REFRESH_INTERVAL) {
1040            Ok(RenderEvent::TaskMessage { identity, message }) => {
1041                let message = format!("{identity}: {message}");
1042                if let Some(display) = &mut terminal {
1043                    display.push_message(message);
1044                } else {
1045                    write_message(output, &message);
1046                }
1047            }
1048            Ok(RenderEvent::Stop { success, message }) => {
1049                if let Some(display) = &mut terminal {
1050                    let snapshot = render_snapshot(&slots);
1051                    let _ = display.draw(&snapshot);
1052                }
1053                if output == OutputMode::Plain {
1054                    write_plain_transitions(&slots, &mut last_statuses);
1055                }
1056                drop(terminal.take());
1057                write_final(output, &slots, success, &message);
1058                break;
1059            }
1060            Err(RecvTimeoutError::Timeout) => {
1061                if output == OutputMode::Plain {
1062                    write_plain_transitions(&slots, &mut last_statuses);
1063                }
1064            }
1065            Err(RecvTimeoutError::Disconnected) => break,
1066        }
1067    }
1068}
1069
1070fn render_snapshot(slots: &[Arc<ProgressSlot>]) -> RenderSnapshot {
1071    RenderSnapshot {
1072        heading: slots
1073            .first()
1074            .and_then(|slot| slot.phase_label.as_deref())
1075            .unwrap_or("Study")
1076            .to_owned(),
1077        summary: summarize(slots),
1078        tasks: terminal_task_views(slots),
1079    }
1080}
1081
1082fn terminal_task_views(slots: &[Arc<ProgressSlot>]) -> Vec<TaskView> {
1083    slots
1084        .iter()
1085        .map(|slot| TaskView {
1086            label: slot.identity.label().to_owned(),
1087            mode: slot.mode,
1088            current: slot.current.load(Ordering::Relaxed),
1089            target: slot
1090                .target_known
1091                .load(Ordering::Acquire)
1092                .then(|| slot.target.load(Ordering::Relaxed)),
1093            started: slot.started.load(Ordering::Acquire),
1094            status: TaskStatus::decode(slot.status.load(Ordering::Acquire)),
1095            detail: lock(&slot.detail).to_string(),
1096        })
1097        .collect()
1098}
1099
1100fn write_message(output: OutputMode, message: &str) {
1101    match output {
1102        OutputMode::Plain => eprintln!("[progress] {message}"),
1103        OutputMode::Terminal | OutputMode::Hidden | OutputMode::Auto => {}
1104    }
1105}
1106
1107fn write_plain_transitions(slots: &[Arc<ProgressSlot>], previous: &mut [TaskStatus]) {
1108    for (slot, old) in slots.iter().zip(previous) {
1109        let status = TaskStatus::decode(slot.status.load(Ordering::Acquire));
1110        if status != *old {
1111            let detail = lock(&slot.detail);
1112            eprintln!(
1113                "[task] identity={} status={} detail={} iteration={} target={}",
1114                slot.identity.task_key(),
1115                status.label(),
1116                detail.as_ref(),
1117                slot.current.load(Ordering::Relaxed),
1118                format_target(slot)
1119            );
1120            *old = status;
1121        }
1122    }
1123}
1124
1125fn write_final(output: OutputMode, slots: &[Arc<ProgressSlot>], success: bool, message: &str) {
1126    if output == OutputMode::Hidden || (output == OutputMode::Terminal && success) {
1127        return;
1128    }
1129    let summary = summarize(slots);
1130    if !success {
1131        for slot in slots {
1132            let detail = lock(&slot.detail);
1133            eprintln!(
1134                "[task-final] task={} status={} detail={}",
1135                slot.identity.task_key(),
1136                TaskStatus::decode(slot.status.load(Ordering::Acquire)).label(),
1137                detail.as_ref(),
1138            );
1139        }
1140    }
1141    eprintln!(
1142        "[study] status={} tasks={} completed={} failed={} cancelled={} skipped={} pending={} message={}",
1143        if success { "completed" } else { "failed" },
1144        summary.total,
1145        summary.completed,
1146        summary.failed,
1147        summary.cancelled,
1148        summary.skipped,
1149        summary.pending,
1150        message
1151    );
1152}
1153
1154fn format_target(slot: &ProgressSlot) -> String {
1155    if slot.target_known.load(Ordering::Acquire) {
1156        slot.target.load(Ordering::Relaxed).to_string()
1157    } else {
1158        "unknown".to_owned()
1159    }
1160}
1161
1162fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
1163    mutex
1164        .lock()
1165        .unwrap_or_else(std::sync::PoisonError::into_inner)
1166}
scientific_workflow/study/renderer.rs

scientific_workflow/study/
renderer.rs
