refactor(execute): decompose spawn into single-concern modules (#343)

## Summary

- Split the ~200-line `spawn()` in `session::execute::spawn` (which
mixed fspy/tokio command construction, Windows Job Object, FD_CLOEXEC
workaround, stdio piping, and cancellation) into four focused files
under `session::execute`:
- `spawn.rs` — now just abstracts fspy vs tokio, returning `ChildHandle
{ stdout, stderr, wait }` with a cancellation-aware `wait` future.
- `pipe.rs` — `pipe_stdio()` drains stdout/stderr, with optional capture
for cache replay. Houses `StdOutput` / `OutputKind`.
- `tracked_accesses.rs` — `TrackedPathAccesses::from_raw()` normalizes
raw fspy accesses. Houses `PathRead`.
  - `win_job.rs` — Windows Job Object RAII, moved out of `mod.rs`.
- `execute_spawn` composes the pieces and measures duration end-to-end
(spawn no longer tracks time).
- Group pipe-mode state into a local `Pipe` enum (`Piped { stdio_config,
capture }` / `Inherited`) so related state lives in one variant. The
`Inherited` arm drops `stdio_config` eagerly, matching existing
behavior.
- Compute `SpawnStdio` directly from the cache/suggestion condition (no
intermediate `use_piped: bool`). `SpawnStdio` now derives
`Copy/Clone/PartialEq/Eq/Debug`.

## Test plan

- [x] `just check`
- [x] `just lint`
- [x] `cargo test` (all library/unit tests, including
`malformed_windows_drive_path_after_workspace_strip_is_ignored` which
moved with `normalize_tracked_workspace_path`)
- [x] `cargo test -p vite_task_bin --test e2e_snapshots` — full e2e
coverage (148 tests) exercising piped + fspy + cache miss/hit, inherited
stdio, cancellation/process-tree kill, and read/write-overlap not-cached
paths
- [ ] Cross-target lint: `just lint-linux` / `just lint-windows` (CI
will cover)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: branchseer

crates/vite_task/src/session/cache/mod.rs

@@ -23,7 +23,7 @@ use wincode::{
     io::{Reader, Writer},
 };
 
-use super::execute::{fingerprint::PostRunFingerprint, spawn::StdOutput};
+use super::execute::{fingerprint::PostRunFingerprint, pipe::StdOutput};
 
 /// Cache lookup key identifying a task's execution configuration.
 ///

crates/vite_task/src/session/execute/fingerprint.rs

@@ -16,7 +16,7 @@ use vite_path::{AbsolutePath, RelativePathBuf};
 use vite_str::Str;
 use wincode::{SchemaRead, SchemaWrite};
 
-use super::spawn::PathRead;
+use super::tracked_accesses::PathRead;
 use crate::{collections::HashMap, session::cache::InputChangeKind};
 
 /// Post-run fingerprint capturing file state after execution.
@@ -53,7 +53,8 @@ pub enum DirEntryKind {
 impl PostRunFingerprint {
     /// Creates a new fingerprint from path accesses after task execution.
     ///
-    /// Negative glob filtering is done upstream in `spawn_with_tracking`.
+    /// Negative glob filtering is done upstream (see
+    /// [`super::tracked_accesses::TrackedPathAccesses::from_raw`]).
     /// Paths already present in `globbed_inputs` are skipped — they are
     /// already tracked by the prerun glob fingerprint, and the read-write
     /// overlap check in `execute_spawn` guarantees the task did not modify

crates/vite_task/src/session/execute/mod.rs

@@ -0,0 +1,104 @@
+//! Drain child stdout/stderr concurrently to writers, with optional capture.
+
+use std::io::Write;
+
+use serde::Serialize;
+use tokio::{
+    io::AsyncReadExt as _,
+    process::{ChildStderr, ChildStdout},
+};
+use tokio_util::sync::CancellationToken;
+use wincode::{SchemaRead, SchemaWrite};
+
+/// Output kind for stdout/stderr
+#[derive(Debug, PartialEq, Eq, Clone, Copy, SchemaWrite, SchemaRead, Serialize)]
+pub enum OutputKind {
+    StdOut,
+    StdErr,
+}
+
+/// Output chunk with stream kind
+#[derive(Debug, SchemaWrite, SchemaRead, Serialize, Clone)]
+pub struct StdOutput {
+    pub kind: OutputKind,
+    pub content: Vec<u8>,
+}
+
+/// Downstream destinations for bytes read from the child's stdout/stderr:
+/// two pass-through writers plus an optional capture buffer (populated in
+/// place during drain for cache replay).
+pub struct PipeSinks<'a> {
+    pub stdout_writer: &'a mut dyn Write,
+    pub stderr_writer: &'a mut dyn Write,
+    pub capture: Option<&'a mut Vec<StdOutput>>,
+}
+
+/// Drain the child's stdout/stderr concurrently into `sinks`.
+///
+/// Bytes are written through `sinks.stdout_writer` / `sinks.stderr_writer` in
+/// real time and, when `sinks.capture` is `Some`, also appended (with adjacent
+/// same-kind chunks coalesced) for cache replay.
+///
+/// On cancellation: returns `Ok(())` without killing the child — the caller
+/// drives the child's cancellation-aware `wait` future next, which observes the
+/// same already-fired token and performs the kill. Dropping `stdout`/`stderr`
+/// closes the pipe read ends (EPIPE on Unix, `ERROR_BROKEN_PIPE` on Windows).
+#[tracing::instrument(level = "debug", skip_all)]
+pub async fn pipe_stdio(
+    mut stdout: ChildStdout,
+    mut stderr: ChildStderr,
+    mut sinks: PipeSinks<'_>,
+    cancellation_token: CancellationToken,
+) -> std::io::Result<()> {
+    let mut stdout_buf = [0u8; 8192];
+    let mut stderr_buf = [0u8; 8192];
+    let mut stdout_done = false;
+    let mut stderr_done = false;
+
+    loop {
+        if stdout_done && stderr_done {
+            return Ok(());
+        }
+        tokio::select! {
+            result = stdout.read(&mut stdout_buf), if !stdout_done => {
+                match result? {
+                    0 => stdout_done = true,
+                    n => {
+                        let bytes = &stdout_buf[..n];
+                        sinks.stdout_writer.write_all(bytes)?;
+                        sinks.stdout_writer.flush()?;
+                        if let Some(capture) = &mut sinks.capture {
+                            append_output_chunk(capture, OutputKind::StdOut, bytes);
+                        }
+                    }
+                }
+            }
+            result = stderr.read(&mut stderr_buf), if !stderr_done => {
+                match result? {
+                    0 => stderr_done = true,
+                    n => {
+                        let bytes = &stderr_buf[..n];
+                        sinks.stderr_writer.write_all(bytes)?;
+                        sinks.stderr_writer.flush()?;
+                        if let Some(capture) = &mut sinks.capture {
+                            append_output_chunk(capture, OutputKind::StdErr, bytes);
+                        }
+                    }
+                }
+            }
+            () = cancellation_token.cancelled() => {
+                return Ok(());
+            }
+        }
+    }
+}
+
+fn append_output_chunk(capture: &mut Vec<StdOutput>, kind: OutputKind, bytes: &[u8]) {
+    if let Some(last) = capture.last_mut()
+        && last.kind == kind
+    {
+        last.content.extend_from_slice(bytes);
+    } else {
+        capture.push(StdOutput { kind, content: bytes.to_vec() });
+    }
+}