fix(envd): prevent output truncation on fast commands

cmd.StdoutPipe/StderrPipe are managed by cmd.Wait which closes the
pipe read-ends on return, racing with readers that haven't finished.
Replace them with manual os.Pipe so we control the lifecycle:
write-ends are closed after Start (child inherited them), read-ends
stay open until readers finish naturally via EOF.

After cmd.Wait reaps the child, call Drain() on the data multiplexer
to disable back-pressure, letting stuck readers unblock and see EOF.
Then wait for all readers to exit before proceeding.

Author: arkamar

packages/envd/internal/services/process/handler/handler.go

@@ -54,8 +54,9 @@ type Handler struct {
 	outCtx    context.Context //nolint:containedctx // todo: refactor so this can be removed
 	outCancel context.CancelFunc
 
-	stdinMu sync.Mutex
-	stdin   io.WriteCloser
+	stdinMu  sync.Mutex
+	stdin    io.WriteCloser
+	pipeEnds []*os.File // write-ends of stdout/stderr pipes; closed after Start
 
 	DataEvent *MultiplexedChannel[rpc.ProcessEvent_Data]
 	EndEvent  *MultiplexedChannel[rpc.ProcessEvent_End]
@@ -235,11 +236,14 @@ func New(
 
 		h.tty = tty
 	} else {
-		stdout, err := cmd.StdoutPipe()
+		stdoutR, stdoutW, err := os.Pipe()
 		if err != nil {
 			return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("error creating stdout pipe for command '%s': %w", userCmd, err))
 		}
 
+		cmd.Stdout = stdoutW
+		stdout := stdoutR
+
 		outWg.Go(func() {
 			stdoutLogs := make(chan []byte, outputBufferSize)
 			defer close(stdoutLogs)
@@ -283,11 +287,16 @@ func New(
 			}
 		})
 
-		stderr, err := cmd.StderrPipe()
+		stderrR, stderrW, err := os.Pipe()
 		if err != nil {
 			return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("error creating stderr pipe for command '%s': %w", userCmd, err))
 		}
 
+		cmd.Stderr = stderrW
+		stderr := stderrR
+
+		h.pipeEnds = []*os.File{stdoutW, stderrW}
+
 		outWg.Go(func() {
 			stderrLogs := make(chan []byte, outputBufferSize)
 			defer close(stderrLogs)
@@ -444,6 +453,12 @@ func (p *Handler) Start(requestTimeout time.Duration) (uint32, error) {
 		if err != nil {
 			return 0, fmt.Errorf("error starting process '%s': %w", p.userCommand(), err)
 		}
+
+		// Close parent's copy of the pipe write-ends; the child
+		// inherited them. Readers will see EOF when the child exits.
+		for _, f := range p.pipeEnds {
+			f.Close()
+		}
 	}
 
 	p.logger.
@@ -458,11 +473,15 @@ func (p *Handler) Start(requestTimeout time.Duration) (uint32, error) {
 }
 
 func (p *Handler) Wait() {
-	// cmd.Wait reaps the child and closes the pipe read-ends.
-	// Then we cancel outCtx to unblock any reader goroutine that
-	// is blocked on a full Source channel send (back-pressure).
+	// Reap the child. With manual os.Pipe() this does NOT close the
+	// read-ends, so readers can still drain remaining pipe data.
 	err := p.cmd.Wait()
-	p.outCancel()
+
+	// Disable back-pressure so the fan-out drains Source even with
+	// no subscribers. Readers will see EOF (child closed write-end
+	// on exit), flush their last chunk, and exit.
+	p.DataEvent.Drain()
+	<-p.outCtx.Done()
 
 	p.tty.Close()
 

packages/envd/internal/services/process/handler/multiplex.go

@@ -131,6 +131,15 @@ func (m *MultiplexedChannel[T]) receiveWhenReady() (v T, ok bool) {
 	}
 }
 
+// Drain disables back-pressure so the fan-out drains Source even
+// with no subscribers. Call this when the producer is done (e.g.
+// child process exited) so readers can flush their last chunks
+// without blocking. Source remains open for further sends.
+func (m *MultiplexedChannel[T]) Drain() {
+	m.closed.Store(true)
+	m.NotifySubscriberChange()
+}
+
 // CloseSource closes the Source channel and wakes the fan-out loop.
 func (m *MultiplexedChannel[T]) CloseSource() {
 	m.closed.Store(true)

packages/envd/internal/services/process/start_test.go

@@ -173,6 +173,44 @@ func TestStart_ProcessSurvivesClientDisconnect(t *testing.T) {
 	_ = proc.Kill()
 }
 
+// TestStart_FastCommandOutputNotTruncated verifies that a fast command's
+// stdout is fully delivered. This catches the bug where cmd.Wait() closing
+// the pipe races with outCancel(), causing the reader to drop its last chunk.
+func TestStart_FastCommandOutputNotTruncated(t *testing.T) {
+	t.Parallel()
+
+	client, cleanup := newTestService(t)
+	defer cleanup()
+
+	const expected = "hello-truncation-test\n"
+
+	for i := range 50 {
+		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+
+		stream, err := client.Start(ctx, connect.NewRequest(&rpc.StartRequest{
+			Process: &rpc.ProcessConfig{
+				Cmd:  "echo",
+				Args: []string{"hello-truncation-test"},
+			},
+		}))
+		require.NoError(t, err)
+
+		var stdout string
+		for stream.Receive() {
+			if data := stream.Msg().GetEvent().GetData(); data != nil {
+				if out := data.GetStdout(); out != nil {
+					stdout += string(out)
+				}
+			}
+		}
+		require.NoError(t, stream.Err())
+		_ = stream.Close()
+		cancel()
+
+		require.Equalf(t, expected, stdout, "iteration %d: output mismatch", i)
+	}
+}
+
 // processAlive checks whether a process with the given PID exists.
 func processAlive(pid int) bool {
 	// /proc/<pid>/stat exists iff the process is alive (Linux-specific).