refactor: streamline stdout and stderr handling in process managers

crates/gpmcp-layer-core/src/process_manager_trait.rs

@@ -198,11 +198,14 @@ impl Decoder for Utf8Codec {
     }
 }
 pub async fn stream<A: AsyncReadExt + Unpin + 'static>(
-    io: &mut A,
+    io: Option<A>,
     out: impl Into<LayerStdio>,
 ) -> tokio::io::Result<()> {
-    let mut frames = FramedRead::with_capacity(io, Utf8Codec, 1024);
-    stream_frames(&mut frames, out.into()).await
+    if let Some(io) = io {
+        let mut frames = FramedRead::with_capacity(io, Utf8Codec, 1024);
+        return stream_frames(&mut frames, out.into()).await;
+    }
+    Ok(())
 }
 
 async fn stream_frames<R: AsyncReadExt + Unpin>(

crates/gpmcp-layer-unix/src/unix_process_manager.rs

@@ -216,23 +216,11 @@ impl ProcessManager for UnixProcessManager {
             );
         }
 
-        // Capture and stream stdout - the stream function will detect it's stdout and route accordingly
-        if let Some(mut stdout) = child.stdout.take() {
-            tokio::spawn(async move {
-                if let Err(e) = stream(&mut stdout, out).await {
-                    warn!("Error streaming stdout: {}", e);
-                }
-            });
-        }
+        let (stdout, stderr) = (child.stdout.take(), child.stderr.take());
 
-        // Capture and stream stderr - the stream function will detect it's stderr and route accordingly
-        if let Some(mut stderr) = child.stderr.take() {
-            tokio::spawn(async move {
-                if let Err(e) = stream(&mut stderr, err).await {
-                    warn!("Error streaming stderr: {}", e);
-                }
-            });
-        }
+        tokio::spawn(
+            async move { tokio::try_join!(stream(stdout, out), stream(stderr, err)).ok() },
+        );
         Ok(UnixProcessHandle::new(child, command.to_string()))
     }
 }

crates/gpmcp-layer-windows/src/windows_process_manager.rs

@@ -275,24 +275,11 @@ impl ProcessManager for WindowsProcessManager {
             );
         }
 
-        // Capture and stream stdout - the stream function will detect it's stdout and route accordingly
-        if let Some(mut stdout) = child.stdout.take() {
-            tokio::spawn(async move {
-                if let Err(e) = stream(&mut stdout, out).await {
-                    warn!("Error streaming stdout: {}", e);
-                }
-            });
-        }
-
-        // Capture and stream stderr - the stream function will detect it's stderr and route accordingly
-        if let Some(mut stderr) = child.stderr.take() {
-            tokio::spawn(async move {
-                if let Err(e) = stream(&mut stderr, err).await {
-                    warn!("Error streaming stderr: {}", e);
-                }
-            });
-        }
+        let (stdout, stderr) = (child.stdout.take(), child.stderr.take());
 
+        tokio::spawn(
+            async move { tokio::try_join!(stream(stdout, out), stream(stderr, err)).ok() },
+        );
         Ok(WindowsProcessHandle::new(child, command.to_string()))
     }
 }