fix(concurrency): address gadfly review — fail-open pool + doc drift

Gadfly's own review of #27 surfaced a real robustness cluster (5 models,
error-handling) plus stale comments I missed.

Robustness — the flock permit pool could hang forever:
- tryAcquire swallowed os.OpenFile errors and treated every flock error as
  "busy", so a broken/missing pool dir (or a filesystem without flock) would
  spin-poll indefinitely; the fanout context is uncancellable and the per-lens
  timeout only starts AFTER acquire returns. Can't trigger in the deploy
  (entrypoint mkdir -p's the dir) but fixed defensively.
- tryAcquire now returns a structural error, distinguished from a healthy-full
  pool (EWOULDBLOCK = busy → keep polling). acquire FAILS OPEN on a structural
  error: logs once and runs the lens unthrottled rather than hanging the review.
- acquire uses time.NewTimer + Stop() (no per-poll timer leak on cancellation).
- activeLensSem warns on stderr when GADFLY_LENS_SEM_DIR is set but the size is
  invalid (was a silent degrade to unthrottled).
- New test: a broken pool dir fails open promptly.

Doc drift (stale references to the removed model cap):
- main.go defaultLensConcurrency + runSpecialists doc, entrypoint.sh status
  pre-seed + lane-launch comments, and the pre-existing lens_concurrency_test.go
  header all updated to the provider-wide-budget wording.

Accepted (graded real, not changed): all-models-start-at-once startup burst
(intended tradeoff) and index-0 sweep bias (cosmetic). One false positive
(one model using the whole budget is the intended lone-model behavior).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

cmd/gadfly/lens_concurrency_test.go

@@ -156,7 +156,7 @@ func TestRunSpecialists_PerProviderFanOut(t *testing.T) {
 
 // TestLensConcurrency covers the resolution matrix: scalar default, scalar
 // override, and per-provider override keyed by the model's resolved lane (same
-// lane rule entrypoint.sh uses for GADFLY_PROVIDER_CONCURRENCY).
+// lane rule entrypoint.sh uses for GADFLY_PROVIDER_LENS_CONCURRENCY).
 func TestLensConcurrency(t *testing.T) {
 	tests := []struct {
 		name     string

cmd/gadfly/lenssem.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -43,44 +44,72 @@ func activeLensSem() *lensSem {
 	}
 	n := envInt("GADFLY_LENS_SEM_SIZE", 0)
 	if n < 1 {
+		// A dir was requested but the size is missing/blank/invalid. Rather than
+		// silently run unthrottled, say so on stderr — it's almost certainly a
+		// misconfiguration in entrypoint.sh (the two are set together).
+		fmt.Fprintf(os.Stderr, "gadfly: GADFLY_LENS_SEM_DIR set but GADFLY_LENS_SEM_SIZE=%q invalid; lenses run unthrottled\n", os.Getenv("GADFLY_LENS_SEM_SIZE"))
 		return nil
 	}
 	return &lensSem{dir: dir, size: n}
 }
 
 // acquire blocks until a permit is free (or ctx is done) and returns a release
-// func. On cancellation it returns a no-op release plus ctx.Err(), so callers can
+// func. It FAILS OPEN: if the pool directory is structurally unusable (missing,
-// surface the lens as "did not run" rather than leaking a permit.
+// unwritable, or on a filesystem without flock) it logs once and returns a no-op
+// release with a nil error, so the lens runs UNTHROTTLED rather than hanging the
+// whole review forever — this is an advisory reviewer, a slightly oversubscribed
+// backend beats a stuck one. Only a full-but-healthy pool actually blocks; on ctx
+// cancellation it returns a no-op release plus ctx.Err() so the caller can surface
+// the lens as "did not run" instead of leaking a permit.
 func (s *lensSem) acquire(ctx context.Context) (func(), error) {
 	for {
-		if release, ok := s.tryAcquire(); ok {
+		release, ok, err := s.tryAcquire()
+		if ok {
 			return release, nil
 		}
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "gadfly: lens permit pool %q unusable (%v); proceeding without a permit\n", s.dir, err)
+			return func() {}, nil
+		}
+		timer := time.NewTimer(lensSemPollInterval)
 		select {
 		case <-ctx.Done():
+			timer.Stop() // don't leak the timer when we bail on cancellation
 			return func() {}, ctx.Err()
-		case <-time.After(lensSemPollInterval):
+		case <-timer.C:
 		}
 	}
 }
 
-// tryAcquire makes one non-blocking sweep over the permit files, returning a
+// tryAcquire makes one non-blocking sweep over the permit files. It returns a
-// release func for the first one it locks. The permit files are created lazily
+// release func for the first one it locks; otherwise it distinguishes a healthy
-// (entrypoint.sh only guarantees the directory exists), so a fresh pool needs no
+// FULL pool (some permit was openable but flock-busy → ok=false, err=nil → keep
-// seeding step beyond mkdir. Closing the *os.File in the returned func drops the
+// polling) from a structurally BROKEN pool (no permit was even acquirable and none
-// flock (the lock lives on the open file description).
+// was merely busy → err set → caller fails open). Permit files are created lazily
-func (s *lensSem) tryAcquire() (func(), bool) {
+// (entrypoint.sh only guarantees the directory exists). Closing the *os.File in
+// the returned func drops the flock (the lock lives on the open file description).
+func (s *lensSem) tryAcquire() (func(), bool, error) {
+	sawBusy := false
+	var structural error
 	for i := 0; i < s.size; i++ {
 		path := filepath.Join(s.dir, fmt.Sprintf("permit.%d", i))
 		f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
 		if err != nil {
+			structural = err // e.g. the pool dir is gone or unwritable
 			continue
 		}
-		if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
+		if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err == nil {
+			return func() { f.Close() }, true, nil
+		} else if errors.Is(err, syscall.EWOULDBLOCK) {
+			sawBusy = true // this permit is held by someone else — pool has capacity
+			f.Close()
+		} else {
+			structural = err // flock unsupported / other → not a transient "busy"
 			f.Close()
-			continue
 		}
-		return func() { f.Close() }, true
 	}
-	return nil, false
+	if sawBusy {
+		return nil, false, nil // full but healthy: another lens will free a permit
+	}
+	return nil, false, structural
 }

cmd/gadfly/lenssem_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -29,27 +30,52 @@ func TestActiveLensSemUnset(t *testing.T) {
 func TestLensSemExhaustionAndRelease(t *testing.T) {
 	s := &lensSem{dir: t.TempDir(), size: 2}
 
-	r1, ok := s.tryAcquire()
+	r1, ok, err := s.tryAcquire()
-	if !ok {
+	if !ok || err != nil {
-		t.Fatal("first acquire should succeed")
+		t.Fatalf("first acquire should succeed: ok=%v err=%v", ok, err)
 	}
-	r2, ok := s.tryAcquire()
+	r2, ok, err := s.tryAcquire()
-	if !ok {
+	if !ok || err != nil {
-		t.Fatal("second acquire should succeed")
+		t.Fatalf("second acquire should succeed: ok=%v err=%v", ok, err)
 	}
-	if _, ok := s.tryAcquire(); ok {
+	// Pool full but healthy: ok=false with NO error (keep polling), not a
-		t.Fatal("third acquire should fail: pool of 2 is exhausted")
+	// structural failure.
+	if _, ok, err := s.tryAcquire(); ok || err != nil {
+		t.Fatalf("third acquire should be full-but-healthy: ok=%v err=%v", ok, err)
 	}
 
 	r1() // free one permit
-	r3, ok := s.tryAcquire()
+	r3, ok, err := s.tryAcquire()
-	if !ok {
+	if !ok || err != nil {
-		t.Fatal("acquire should succeed after a release")
+		t.Fatalf("acquire should succeed after a release: ok=%v err=%v", ok, err)
 	}
 	r2()
 	r3()
 }
 
+// A structurally broken pool (dir missing/unwritable) must FAIL OPEN — tryAcquire
+// surfaces the error and acquire returns promptly with a no-op release and no
+// error, so a lens runs unthrottled instead of spinning forever.
+func TestLensSemBrokenPoolFailsOpen(t *testing.T) {
+	s := &lensSem{dir: filepath.Join(t.TempDir(), "does", "not", "exist"), size: 2}
+
+	if _, ok, err := s.tryAcquire(); ok || err == nil {
+		t.Fatalf("tryAcquire on a broken pool should report a structural error: ok=%v err=%v", ok, err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	start := time.Now()
+	release, err := s.acquire(ctx)
+	if err != nil {
+		t.Fatalf("acquire should fail open (nil err), got %v", err)
+	}
+	if waited := time.Since(start); waited > time.Second {
+		t.Fatalf("acquire on a broken pool should return promptly, waited %v", waited)
+	}
+	release() // must be a safe no-op
+}
+
 func TestLensSemAcquireBlocksThenCancels(t *testing.T) {
 	s := &lensSem{dir: t.TempDir(), size: 1}
 

cmd/gadfly/main.go

@@ -97,9 +97,11 @@ const (
 	// calls and then hard-failing with "max steps reached without a final
 	// answer" — it always has a few steps left to wrap up.
 	defaultWrapUpReserve = 4
-	// defaultLensConcurrency is how many specialist lenses run at once within a
+	// defaultLensConcurrency is the fallback lens budget when neither
-	// single model. 1 keeps the suite sequential (the historical behavior);
+	// GADFLY_PROVIDER_LENS_CONCURRENCY nor GADFLY_LENS_CONCURRENCY is set. 1 keeps
-	// higher values overlap the independent per-lens passes. See runSpecialists.
+	// the suite sequential (the historical behavior); higher values overlap the
+	// independent per-lens passes. Under entrypoint.sh the resolved value is the
+	// provider-wide budget shared across the provider's models. See runSpecialists.
 	defaultLensConcurrency = 1
 )
 
@@ -245,17 +247,15 @@ func run() error {
 
 // runSpecialists reviews the diff through each lens and returns the results in
 // the SAME order as specialists, regardless of finish order. It uses executus's
-// fanout primitive: up to GADFLY_LENS_CONCURRENCY lenses run concurrently (the
+// fanout primitive to overlap independent lens passes (fanout.Run returns one
-// default of 1 keeps the suite sequential, exactly as before), and fanout.Run
+// result per lens in input order); each lens runs under its own per-lens timeout
-// returns one result per lens in input order. Each lens already runs under its
+// (reviewWithSpecialist) and the lenses only read the immutable repoFS.
-// own per-lens timeout (reviewWithSpecialist) and the lenses only read the
-// immutable repoFS, so concurrency simply overlaps independent passes.
 //
-// Provider-wide throttling: when entrypoint.sh runs several of a provider's
+// Throttling: when entrypoint.sh runs several of a provider's models at once it
-// models at once it seeds a shared lens-permit pool (activeLensSem) that every
+// seeds a shared lens-permit pool (activeLensSem) that every model's lenses draw
-// model's lenses draw from, so the real cap is total lens passes in flight per
+// from, so the real cap is total lens passes in flight per PROVIDER — not
-// provider — not (models at once) × (lenses at once). Absent that pool (local
+// (models at once) × (lenses at once). Absent that pool (local runs, tests) the
-// runs, tests) this falls back to the in-process lensConcurrency() limit alone.
+// in-process GADFLY_LENS_CONCURRENCY limit applies alone (default 1 = sequential).
 func runSpecialists(eng reviewEngine, base string, specialists []Specialist, task, diff string) []specialistResult {
 	// Optional live status board: publishes this model's per-lens progress to a
 	// file the entrypoint board renders. Inert (no-op) unless GADFLY_STATUS_FILE
@@ -265,10 +265,11 @@ func runSpecialists(eng reviewEngine, base string, specialists []Specialist, tas
 	// The cross-process pool (if any) is the real ceiling; size the in-process
 	// fanout to it so a lone model in its lane can use the whole provider budget,
 	// while extra goroutines simply block in sem.acquire until a permit frees.
+	// Absent a pool, fall back to the in-process lens limit.
 	sem := activeLensSem()
 	maxConcurrent := lensConcurrency()
 	if sem != nil {
-		maxConcurrent = sem.size
+		maxConcurrent = sem.size // the shared pool is the real ceiling
 	}
 
 	fanResults := fanout.Run(context.Background(), specialists, fanout.Options[Specialist]{

entrypoint.sh

@@ -354,8 +354,8 @@ BOARD_PID=""
 if [ "${GADFLY_STATUS_BOARD:-1}" != "0" ]; then
   rm -rf "$STATUS_DIR"; mkdir -p "$STATUS_DIR"
   # Pre-seed every model as queued so the board shows the full swarm from t=0,
-  # even models still waiting on their provider lane's concurrency cap. Each
+  # even models whose lenses are still waiting on their provider's lens budget.
-  # binary overwrites its own file with real per-lens detail once it starts.
+  # Each binary overwrites its own file with real per-lens detail once it starts.
   for m in "${MODEL_LIST[@]}"; do
     jq -n --arg model "$m" --arg provider "$(provider_of "$m")" \
       '{model:$model, provider:$provider, started:0, updated:0, done:false, lenses:[]}' \
@@ -394,9 +394,9 @@ if [ "${GADFLY_PR_BUDGET_SECS:-0}" -gt 0 ] 2>/dev/null; then
 fi
 
 log "providers: ${PROVIDERS:-none}"
-# Each provider lane runs in parallel; cap is enforced within each lane. Track
+# Each provider lane runs in parallel; the shared lens budget throttles within
-# the lane PIDs so we wait ONLY for the review work — not the status board,
+# each lane. Track the lane PIDs so we wait ONLY for the review work — not the
-# which intentionally runs until we signal it below.
+# status board, which intentionally runs until we signal it below.
 LANE_PIDS=()
 for p in $PROVIDERS; do
   run_lane "$p" &