perf(tools): centralize batch parallelism with panic-safe helper (#33)

* perf(tools): centralize batch parallelism with panic-safe helper

The eight batch tools each hand-rolled the same semaphore + drain pattern,
and three of them (batch_read, http_batch, parallel_shell) ran their worker
bodies with NO per-goroutine panic recovery. A panic in a worker goroutine
is not caught by the deferred recover in the spawning Call — it crashes the
entire agent process. So a single malformed input to one of those tools
could take odek down.

- Add parallelMap[I,O]: bounded-concurrency indexed map that recovers a
  panic in any worker and routes it through a caller-supplied handler,
  guaranteeing one bad task can never crash the process. Results stay in
  input order. Replaces ~8 copies of the semaphore/drain boilerplate.
- Add toolConcurrency(): CPU-aware worker count clamped to [4,8] (was a
  hard-coded 4), overlapping more I/O on multi-core machines without
  thrashing.
- Migrate batch_read, count_lines, checksum, word_count, multi_grep,
  head_tail, parallel_shell, http_batch to the helper. parallel_shell and
  http_batch keep their serial up-front approval pre-check (which may
  prompt) and only parallelize the side-effect-free execution; their inline
  worker bodies are extracted into runOne/fetchOne.

Recoverability is the headline fix; the concurrency bump and de-duplication
come along for free. Behavior (outputs, ordering, approval semantics) is
unchanged — verified by the existing per-tool tests under -race plus new
parallelMap tests covering ordering, panic recovery, the concurrency
bound, and limit clamping.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* harden(parallel): make crash-prevention guarantee unconditional

vprotocol axis 2.5/2.8: parallelMap recovers a worker panic and calls the
caller's recovered() handler — but if that handler itself panicked, the
second panic escaped the goroutine and crashed the process, defeating the
helper's whole purpose. Recover around the handler too, so no worker can
crash the process under any input. Added a test with a panicking handler.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: jkyberneees

cmd/odek/file_tool.go

@@ -916,21 +916,10 @@ func (t *batchReadTool) Call(argsJSON string) (result string, err error) {
 		return jsonError(fmt.Sprintf("max %d files per batch_read call", maxBatchFiles))
 	}
 
-	results := make([]batchReadFileResult, len(args.Files))
-	sem := make(chan struct{}, 4) // bounded concurrency for file I/O
-
-	for i, f := range args.Files {
-		sem <- struct{}{}
-		go func(idx int, fileArg batchReadFileArg) {
-			defer func() { <-sem }()
-			results[idx] = t.readSingle(fileArg)
-		}(i, f)
-	}
-
-	// Drain — wait for all goroutines
-	for i := 0; i < cap(sem); i++ {
-		sem <- struct{}{}
-	}
+	results := parallelMap(args.Files, toolConcurrency(), t.readSingle,
+		func(f batchReadFileArg, p any) batchReadFileResult {
+			return batchReadFileResult{Path: f.Path, Error: fmt.Sprintf("internal error: %v", p)}
+		})
 
 	return jsonResult(batchReadResult{Results: results})
 }

cmd/odek/parallel.go

@@ -0,0 +1,73 @@
+package main
+
+import (
+	"runtime"
+	"sync"
+)
+
+// toolConcurrency is the worker count the batch tools use for bounded
+// parallelism. It scales with the machine (GOMAXPROCS) but stays in a modest
+// [4, 8] band: enough to overlap file/network/subprocess I/O on a multi-core
+// dev machine, low enough to avoid thrashing disks, sockets, or the process
+// table. The batch tools previously hard-coded 4; this never drops below that.
+func toolConcurrency() int {
+	n := runtime.GOMAXPROCS(0)
+	switch {
+	case n < 4:
+		return 4
+	case n > 8:
+		return 8
+	default:
+		return n
+	}
+}
+
+// parallelMap applies work to every item with at most `limit` workers running
+// at once, returning results in input order: out[i] == work(items[i]).
+//
+// It recovers a panic in any single worker and routes it through
+// recovered(item, panicValue) instead of letting it unwind. This is the whole
+// point of centralising the pattern: a panic in a goroutine is NOT caught by a
+// deferred recover in the function that spawned it — it crashes the entire
+// process. Every batch tool fans work out to goroutines, so without this guard
+// one malformed input (a nil deref, an out-of-range slice) in a single worker
+// takes down the whole agent. Three of the hand-rolled copies this replaces
+// were missing per-worker recovery entirely.
+//
+// A nil `recovered` is tolerated: a panicked slot is left as the zero value.
+func parallelMap[I, O any](items []I, limit int, work func(I) O, recovered func(I, any) O) []O {
+	out := make([]O, len(items))
+	if len(items) == 0 {
+		return out
+	}
+	if limit < 1 {
+		limit = 1
+	}
+	if limit > len(items) {
+		limit = len(items)
+	}
+
+	var wg sync.WaitGroup
+	sem := make(chan struct{}, limit)
+	for i := range items {
+		sem <- struct{}{} // acquire before spawning — bounds live goroutines too
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			defer func() {
+				if r := recover(); r != nil && recovered != nil {
+					// The handler runs while unwinding a worker panic; if it
+					// panics too, that second panic must not escape either — the
+					// guarantee is that NO worker can crash the process. A
+					// panicked handler leaves the slot at its zero value.
+					defer func() { _ = recover() }()
+					out[idx] = recovered(items[idx], r)
+				}
+			}()
+			out[idx] = work(items[idx])
+		}(i)
+	}
+	wg.Wait()
+	return out
+}

cmd/odek/parallel_test.go

@@ -0,0 +1,200 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestParallelMap_OrderAndCorrectness(t *testing.T) {
+	items := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+	out := parallelMap(items, 3,
+		func(i int) int { return i * i },
+		func(i int, _ any) int { return -1 })
+	for i, v := range out {
+		if v != i*i {
+			t.Errorf("out[%d] = %d, want %d", i, v, i*i)
+		}
+	}
+}
+
+func TestParallelMap_Empty(t *testing.T) {
+	out := parallelMap(nil, 4,
+		func(i int) int { return i },
+		func(i int, _ any) int { return -1 })
+	if len(out) != 0 {
+		t.Errorf("expected empty output, got %d", len(out))
+	}
+}
+
+// TestParallelMap_RecoversWorkerPanic is the core recoverability contract: a
+// panic in one worker must NOT unwind (which would crash the process) and must
+// not affect the other slots — the panicked slot gets the recovered() value.
+func TestParallelMap_RecoversWorkerPanic(t *testing.T) {
+	items := []int{0, 1, 2, 3, 4}
+	out := parallelMap(items, 3,
+		func(i int) string {
+			if i == 2 {
+				panic("boom")
+			}
+			return fmt.Sprintf("ok-%d", i)
+		},
+		func(i int, r any) string {
+			return fmt.Sprintf("recovered-%d-%v", i, r)
+		})
+
+	want := []string{"ok-0", "ok-1", "recovered-2-boom", "ok-3", "ok-4"}
+	for i := range want {
+		if out[i] != want[i] {
+			t.Errorf("out[%d] = %q, want %q", i, out[i], want[i])
+		}
+	}
+}
+
+func TestParallelMap_MultiplePanicsAllRecovered(t *testing.T) {
+	items := []int{0, 1, 2, 3, 4, 5}
+	out := parallelMap(items, 4,
+		func(i int) int {
+			if i%2 == 0 {
+				panic(i)
+			}
+			return i
+		},
+		func(i int, _ any) int { return -i })
+	for i := range items {
+		want := i
+		if i%2 == 0 {
+			want = -i
+		}
+		if out[i] != want {
+			t.Errorf("out[%d] = %d, want %d", i, out[i], want)
+		}
+	}
+}
+
+// TestParallelMap_NilRecoveredLeavesZero verifies a nil recovered handler is
+// tolerated: the panicked slot is left as the zero value, and crucially the
+// panic still does not crash the process.
+func TestParallelMap_NilRecoveredLeavesZero(t *testing.T) {
+	out := parallelMap([]int{0, 1, 2}, 2,
+		func(i int) string {
+			if i == 1 {
+				panic("x")
+			}
+			return "v"
+		}, nil)
+	if out[0] != "v" || out[2] != "v" {
+		t.Errorf("non-panicking slots wrong: %v", out)
+	}
+	if out[1] != "" {
+		t.Errorf("panicked slot should be zero value, got %q", out[1])
+	}
+}
+
+// TestParallelMap_PanickingHandlerDoesNotCrash verifies the guarantee is
+// unconditional: even a recovered() handler that itself panics cannot crash the
+// process — the worker's slot is just left at its zero value.
+func TestParallelMap_PanickingHandlerDoesNotCrash(t *testing.T) {
+	out := parallelMap([]int{0, 1, 2}, 2,
+		func(i int) string {
+			if i == 1 {
+				panic("work boom")
+			}
+			return "ok"
+		},
+		func(i int, _ any) string {
+			panic("handler boom") // must not escape the goroutine
+		})
+	if out[0] != "ok" || out[2] != "ok" {
+		t.Errorf("non-panicking slots wrong: %v", out)
+	}
+	if out[1] != "" {
+		t.Errorf("slot with panicking handler should be zero value, got %q", out[1])
+	}
+}
+
+// TestParallelMap_ConcurrencyBounded verifies no more than `limit` workers run
+// at once, even with far more items than the limit.
+func TestParallelMap_ConcurrencyBounded(t *testing.T) {
+	const n, limit = 30, 4
+	var cur int64
+	var mu sync.Mutex
+	peak := int64(0)
+	release := make(chan struct{})
+
+	work := func(i int) int {
+		c := atomic.AddInt64(&cur, 1)
+		mu.Lock()
+		if c > peak {
+			peak = c
+		}
+		mu.Unlock()
+		<-release // hold the worker so concurrency can be observed
+		atomic.AddInt64(&cur, -1)
+		return i
+	}
+
+	items := make([]int, n)
+	for i := range items {
+		items[i] = i
+	}
+
+	doneCh := make(chan []int, 1)
+	go func() {
+		doneCh <- parallelMap(items, limit, work, func(int, any) int { return -1 })
+	}()
+
+	// Wait until the limit is saturated.
+	deadline := time.After(2 * time.Second)
+	for atomic.LoadInt64(&cur) < int64(limit) {
+		select {
+		case <-deadline:
+			close(release)
+			t.Fatal("workers never reached the concurrency limit")
+		default:
+			time.Sleep(time.Millisecond)
+		}
+	}
+	// Give any over-scheduled workers a moment to (incorrectly) start.
+	time.Sleep(50 * time.Millisecond)
+
+	mu.Lock()
+	gotPeak := peak
+	mu.Unlock()
+	if gotPeak > limit {
+		close(release)
+		t.Fatalf("peak concurrency %d exceeded limit %d", gotPeak, limit)
+	}
+
+	close(release)
+	out := <-doneCh
+	for i, v := range out {
+		if v != i {
+			t.Errorf("out[%d] = %d, want %d", i, v, i)
+		}
+	}
+}
+
+func TestParallelMap_LimitClamping(t *testing.T) {
+	items := []int{0, 1, 2}
+	// limit larger than len and limit < 1 must both still produce correct output.
+	for _, limit := range []int{0, -5, 100} {
+		out := parallelMap(items, limit,
+			func(i int) int { return i + 1 },
+			func(i int, _ any) int { return -1 })
+		for i := range items {
+			if out[i] != i+1 {
+				t.Errorf("limit=%d: out[%d] = %d, want %d", limit, i, out[i], i+1)
+			}
+		}
+	}
+}
+
+func TestToolConcurrency_InBand(t *testing.T) {
+	c := toolConcurrency()
+	if c < 4 || c > 8 {
+		t.Errorf("toolConcurrency() = %d, want in [4,8]", c)
+	}
+}