refactor(pipeline): add PipelineError::reconstruct to de-duplicate outcome rematch

src/lib/pipeline/core/builder.rs

@@ -751,7 +751,6 @@ impl Pipeline {
             build_chain_contexts, build_worker_storage, is_fusible_chain, run_fused_single_thread,
             run_worker_loop,
         };
-        use super::signal::PipelineError;
         use super::step::StepKind;
         use super::topology::StepIdx;
 
@@ -1022,17 +1021,7 @@ impl Pipeline {
         // (io::Error isn't Clone); reconstruct from the recorded outcome.
         let _ = graph;
         match signal.outcome() {
-            Some(PipelineError::Cancelled) => Err(PipelineError::Cancelled),
-            Some(PipelineError::Io { step, source }) => Err(PipelineError::Io {
-                step,
-                source: std::io::Error::new(source.kind(), format!("{source}")),
-            }),
-            Some(PipelineError::NotEnoughThreads { required, available }) => {
-                Err(PipelineError::NotEnoughThreads { required: *required, available: *available })
-            }
-            Some(PipelineError::TimedOut { stalled_secs }) => {
-                Err(PipelineError::TimedOut { stalled_secs: *stalled_secs })
-            }
+            Some(err) => Err(err.reconstruct()),
             None => Ok(()),
         }
     }

src/lib/pipeline/core/runtime/fused.rs

@@ -189,19 +189,9 @@ pub fn run_fused_single_thread(
     }
 
     // Map the recorded outcome to the run result (same shape as `Pipeline::run`;
-    // `PipelineError` is not `Clone`, so reconstruct the `Io` variant).
+    // `PipelineError` is not `Clone`, so reconstruct via `PipelineError::reconstruct`).
     match signal.outcome() {
-        Some(PipelineError::Cancelled) => Err(PipelineError::Cancelled),
-        Some(PipelineError::Io { step, source }) => Err(PipelineError::Io {
-            step,
-            source: std::io::Error::new(source.kind(), format!("{source}")),
-        }),
-        Some(PipelineError::NotEnoughThreads { required, available }) => {
-            Err(PipelineError::NotEnoughThreads { required: *required, available: *available })
-        }
-        Some(PipelineError::TimedOut { stalled_secs }) => {
-            Err(PipelineError::TimedOut { stalled_secs: *stalled_secs })
-        }
+        Some(err) => Err(err.reconstruct()),
         None => Ok(()),
     }
 }

src/lib/pipeline/core/signal.rs

@@ -26,6 +26,29 @@ pub enum PipelineError {
     TimedOut { stalled_secs: u64 },
 }
 
+impl PipelineError {
+    /// Reconstruct an owned copy of this error.
+    ///
+    /// `PipelineError` is not `Clone` because its `Io` variant holds an
+    /// [`io::Error`], which is not `Clone`; this rebuilds that source from its
+    /// `kind()` + display string. Used to turn the borrowed `signal.outcome()`
+    /// into the owned error returned from the run drivers (`Pipeline::run` and
+    /// the fused single-thread driver), which previously open-coded the same
+    /// per-variant rematch in two places.
+    pub(crate) fn reconstruct(&self) -> PipelineError {
+        match self {
+            Self::Cancelled => Self::Cancelled,
+            Self::Io { step, source } => {
+                Self::Io { step, source: io::Error::new(source.kind(), format!("{source}")) }
+            }
+            Self::NotEnoughThreads { required, available } => {
+                Self::NotEnoughThreads { required: *required, available: *available }
+            }
+            Self::TimedOut { stalled_secs } => Self::TimedOut { stalled_secs: *stalled_secs },
+        }
+    }
+}
+
 impl std::fmt::Display for PipelineError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
@@ -210,4 +233,28 @@ mod tests {
         assert!(handle.is_cancelled());
         assert!(signal.is_cancelled());
     }
+
+    #[test]
+    fn reconstruct_preserves_every_variant() {
+        // `reconstruct` rebuilds an owned error from a borrow (PipelineError
+        // isn't Clone). Each variant must round-trip; the `Io` source is
+        // rebuilt from kind + display, the rest are copied verbatim.
+        let io = PipelineError::Io { step: "s", source: io::Error::other("boom") }.reconstruct();
+        match io {
+            PipelineError::Io { step, source } => {
+                assert_eq!(step, "s");
+                assert_eq!(source.kind(), io::ErrorKind::Other);
+                assert_eq!(source.to_string(), "boom");
+            }
+            other => panic!("expected Io, got {other:?}"),
+        }
+
+        assert!(matches!(PipelineError::Cancelled.reconstruct(), PipelineError::Cancelled));
+
+        let net = PipelineError::NotEnoughThreads { required: 4, available: 2 }.reconstruct();
+        assert!(matches!(net, PipelineError::NotEnoughThreads { required: 4, available: 2 }));
+
+        let to = PipelineError::TimedOut { stalled_secs: 60 }.reconstruct();
+        assert!(matches!(to, PipelineError::TimedOut { stalled_secs: 60 }));
+    }
 }