Revert "execute: remove auto-batching from pipeline, fan-out is always explicit"

This reverts commit 2117f14a53.

pkg/tools/execute/code.go

@@ -213,14 +213,38 @@ func dispatchExecuteCode(ctx context.Context, e *Server, args json.RawMessage) (
 		return e.executeWithStdin(ctx, code, lang, timeout, sandboxName, trusted, "")
 	}
 
-	// Pipeline: stages run sequentially, stdout piped to next stdin.
+	// Pipeline: run stages sequentially, auto-batching consecutive read-safe
+	// tool steps into parallel groups. Write/global steps serialize.
 	var input string
-	for i, stage := range stages {
-		out, err := e.runStage(ctx, i, stage, timeout, sandboxName, input)
-		if err != nil {
-			return out, fmt.Errorf("step %d: %w", i, err)
+	for i := 0; i < len(stages); {
+		if classifyStep(stages[i]) == lockClassRead {
+			// Collect run of consecutive read-safe stages.
+			j := i + 1
+			for j < len(stages) && classifyStep(stages[j]) == lockClassRead {
+				j++
+			}
+			out, err := e.runReadBatch(ctx, i, stages[i:j], timeout, sandboxName, input)
+			if err != nil {
+				return out, err
+			}
+			input = out
+			i = j
+		} else {
+			// Write or global: acquire exclusive lock, run single stage.
+			lf, err := acquireFlock(e.lockDir, "rw", true)
+			if err != nil {
+				return "", fmt.Errorf("step %d: lock: %w", i, err)
+			}
+			out, runErr := e.runStage(ctx, i, stages[i], timeout, sandboxName, input)
+			if lf != nil {
+				lf.Close()
+			}
+			if runErr != nil {
+				return out, fmt.Errorf("step %d: %w", i, runErr)
+			}
+			input = out
+			i++
 		}
-		input = out
 	}
 	return input, nil
 }

pkg/tools/execute/execute_test.go

@@ -1050,3 +1050,46 @@ func TestSetLockDir(t *testing.T) {
 		t.Errorf("lockDir = %q; want %q", s.lockDir, dir)
 	}
 }
+
+func TestRunReadBatch_Single(t *testing.T) {
+	s := newServer(t)
+	ctx := context.Background()
+	stages := []CodeStep{{Code: "echo batch", Tool: ""}}
+	out, err := s.runReadBatch(ctx, 0, stages, 10, "default", "")
+	if err != nil {
+		t.Fatalf("runReadBatch: %v", err)
+	}
+	if !strings.Contains(out, "batch") {
+		t.Errorf("output = %q; want 'batch'", out)
+	}
+}
+
+func TestRunReadBatch_Multi(t *testing.T) {
+	s := newServer(t)
+	ctx := context.Background()
+	stages := []CodeStep{
+		{Code: "echo one"},
+		{Code: "echo two"},
+	}
+	out, err := s.runReadBatch(ctx, 0, stages, 10, "default", "")
+	if err != nil {
+		t.Fatalf("runReadBatch multi: %v", err)
+	}
+	if !strings.Contains(out, "one") || !strings.Contains(out, "two") {
+		t.Errorf("output = %q; want 'one' and 'two'", out)
+	}
+}
+
+func TestRunReadBatch_WithLockDir(t *testing.T) {
+	s := newServer(t)
+	s.SetLockDir(t.TempDir())
+	ctx := context.Background()
+	stages := []CodeStep{{Code: "echo locked"}}
+	out, err := s.runReadBatch(ctx, 0, stages, 10, "default", "")
+	if err != nil {
+		t.Fatalf("runReadBatch with lockdir: %v", err)
+	}
+	if !strings.Contains(out, "locked") {
+		t.Errorf("output = %q; want 'locked'", out)
+	}
+}

pkg/tools/execute/pipe.go

@@ -60,3 +60,53 @@ func (e *Server) runStage(ctx context.Context, idx int, stage CodeStep, timeout
 
 	return "", fmt.Errorf("stage %d: requires code, tool, or parallel", idx)
 }
+
+// runReadBatch runs a contiguous slice of read-safe stages concurrently,
+// concatenating their outputs in submission order. All stages receive the
+// same stdinData (read-safe tools don't depend on stdin chaining).
+// If e.lockDir is set, each goroutine acquires LOCK_SH before running.
+func (e *Server) runReadBatch(ctx context.Context, startIdx int, stages []CodeStep, timeout int, sandboxName, stdinData string) (string, error) {
+	if len(stages) == 1 {
+		lf, err := acquireFlock(e.lockDir, "rw", false)
+		if err != nil {
+			return "", fmt.Errorf("step %d: lock: %w", startIdx, err)
+		}
+		if lf != nil {
+			defer lf.Close()
+		}
+		out, err := e.runStage(ctx, startIdx, stages[0], timeout, sandboxName, stdinData)
+		if err != nil {
+			return out, fmt.Errorf("step %d: %w", startIdx, err)
+		}
+		return out, nil
+	}
+
+	results := make([]string, len(stages))
+	errs := make([]error, len(stages))
+	var wg sync.WaitGroup
+	for i, stage := range stages {
+		wg.Add(1)
+		go func(j int, s CodeStep) {
+			defer wg.Done()
+			lf, lerr := acquireFlock(e.lockDir, "rw", false)
+			if lerr != nil {
+				errs[j] = fmt.Errorf("step %d: lock: %w", startIdx+j, lerr)
+				return
+			}
+			if lf != nil {
+				defer lf.Close()
+			}
+			results[j], errs[j] = e.runStage(ctx, startIdx+j, s, timeout, sandboxName, stdinData)
+		}(i, stage)
+	}
+	wg.Wait()
+
+	var out string
+	for i, r := range results {
+		out += r
+		if errs[i] != nil {
+			return out, errs[i]
+		}
+	}
+	return out, nil
+}