From 2117f14a534174ad1a1277561946ff12f3604af9 Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Mon, 11 May 2026 13:40:47 +0200 Subject: [PATCH] execute: remove auto-batching from pipeline, fan-out is always explicit steps is always a pipe (sequential, stdout chains to next stdin). Concurrent fan-out requires explicit use of the parallel field. The ollie:parallel annotation still drives agent-level tool call parallelism via IsParallelRead, but no longer hijacks pipe semantics. --- pkg/tools/execute/code.go | 36 ++++------------------ pkg/tools/execute/execute_test.go | 43 -------------------------- pkg/tools/execute/pipe.go | 50 ------------------------------- 3 files changed, 6 insertions(+), 123 deletions(-) diff --git a/pkg/tools/execute/code.go b/pkg/tools/execute/code.go index a7cf912..feb9f50 100644 --- a/pkg/tools/execute/code.go +++ b/pkg/tools/execute/code.go @@ -213,38 +213,14 @@ func dispatchExecuteCode(ctx context.Context, e *Server, args json.RawMessage) ( return e.executeWithStdin(ctx, code, lang, timeout, sandboxName, trusted, "") } - // Pipeline: run stages sequentially, auto-batching consecutive read-safe - // tool steps into parallel groups. Write/global steps serialize. + // Pipeline: stages run sequentially, stdout piped to next stdin. var input string - 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++ + 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) } + input = out } return input, nil } diff --git a/pkg/tools/execute/execute_test.go b/pkg/tools/execute/execute_test.go index 7654032..8689701 100644 --- a/pkg/tools/execute/execute_test.go +++ b/pkg/tools/execute/execute_test.go @@ -1050,46 +1050,3 @@ 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) - } -} diff --git a/pkg/tools/execute/pipe.go b/pkg/tools/execute/pipe.go index cf1bf7a..c97560d 100644 --- a/pkg/tools/execute/pipe.go +++ b/pkg/tools/execute/pipe.go @@ -60,53 +60,3 @@ 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 -}