feat(agent): stream step output chunks over CommandStream

agent/internal/exec/exec.go

@@ -2,22 +2,44 @@ package exec
 
 import (
 	"bufio"
-	"bytes"
 	"context"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/mrhid6/vantage/agent/internal/grpc/pb"
 )
 
+// streamWriter forwards every write to emit() as an ordered chunk. Used as both
+// Stdout and Stderr so output interleaves in real execution order. The mutex
+// ensures a single stdout/stderr write is not interleaved mid-slice with another.
+type streamWriter struct {
+	mu   sync.Mutex
+	seq  uint64
+	emit func(seq uint64, data []byte)
+}
+
+func (w *streamWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.emit != nil {
+		buf := make([]byte, len(p))
+		copy(buf, p)
+		w.emit(w.seq, buf)
+		w.seq++
+	}
+	return len(p), nil
+}
+
 // RunStep writes the script to a temp file, provides a WORKFLOW_ENV file for
 // the script to append KEY=value output to, executes it under the requested
-// interpreter, and returns captured output plus parsed output env.
-func RunStep(cmd *pb.RunStepCmd) *pb.StepResult {
+// interpreter, and streams output via emit, returning the terminal result
+// with empty stdout/stderr but populated exit_code/output_env.
+func RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *pb.StepResult {
 	res := &pb.StepResult{CommandId: "", OutputEnv: map[string]string{}}
 
 	dir, err := os.MkdirTemp("", "vantage-step-")
@@ -74,21 +96,20 @@ func RunStep(cmd *pb.RunStepCmd) *pb.StepResult {
 		c.Env = append(c.Env, k+"="+v)
 	}
 
-	var stdout, stderr bytes.Buffer
-	c.Stdout = &stdout
-	c.Stderr = &stderr
+	sw := &streamWriter{emit: emit}
+	c.Stdout = sw
+	c.Stderr = sw
 	runErr := c.Run()
 
-	res.Stdout = stdout.String()
-	res.Stderr = stderr.String()
+	// stdout/stderr are streamed via emit, not returned in the result.
 	if ctx.Err() == context.DeadlineExceeded {
 		res.ExitCode = 124
-		res.Stderr += "\n[vantage] step timed out"
+		res.Stderr = "[vantage] step timed out"
 	} else if ee, ok := runErr.(*exec.ExitError); ok {
 		res.ExitCode = ee.ExitCode()
 	} else if runErr != nil {
 		res.ExitCode = 1
-		res.Stderr += "\n[vantage] " + runErr.Error()
+		res.Stderr = "[vantage] " + runErr.Error()
 	}
 
 	res.OutputEnv = parseEnvFile(envFile)