diff --git a/.gitea/workflows/review-reusable.yml b/.gitea/workflows/review-reusable.yml index 29e140f..730a136 100644 --- a/.gitea/workflows/review-reusable.yml +++ b/.gitea/workflows/review-reusable.yml @@ -44,7 +44,7 @@ on: # # Owner-set user-scope variables (see README "Central config via variables"): # GADFLY_DEFAULT_MODELS, GADFLY_DEFAULT_SPECIALISTS, - # GADFLY_DEFAULT_PROVIDER_CONCURRENCY, GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY, + # GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY (the provider-wide lens budget), # GADFLY_ENDPOINT_NETHERSTORM (the local GPU box endpoint). # An unset variable + no input → the image default (one model, default suite), # so a public consumer with neither still gets a sane minimal review. @@ -53,8 +53,8 @@ on: specialists: { type: string, default: "" } # GADFLY_SPECIALISTS — empty falls back to user var GADFLY_DEFAULT_SPECIALISTS provider: { type: string, default: "" } # GADFLY_PROVIDER base_url: { type: string, default: "" } # GADFLY_BASE_URL - provider_concurrency: { type: string, default: "" } # GADFLY_PROVIDER_CONCURRENCY — empty falls back to user var GADFLY_DEFAULT_PROVIDER_CONCURRENCY - provider_lens_concurrency: { type: string, default: "" } # GADFLY_PROVIDER_LENS_CONCURRENCY — empty falls back to user var GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY + provider_concurrency: { type: string, default: "" } # DEPRECATED / ignored — the per-provider MODEL cap was removed; the lens budget below is the single throttle. Kept so existing callers don't error. + provider_lens_concurrency: { type: string, default: "" } # GADFLY_PROVIDER_LENS_CONCURRENCY — the per-provider lens budget (shared across the provider's models); empty falls back to user var GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY timeout_secs: { type: string, default: "600" } # GADFLY_TIMEOUT_SECS (per lens) max_steps: { type: string, default: "14" } # GADFLY_MAX_STEPS worker_model: { type: string, default: "" } # GADFLY_WORKER_MODEL @@ -151,7 +151,9 @@ jobs: GADFLY_SPECIALISTS: ${{ inputs.specialists || vars.GADFLY_DEFAULT_SPECIALISTS }} GADFLY_PROVIDER: ${{ inputs.provider }} GADFLY_BASE_URL: ${{ inputs.base_url }} - GADFLY_PROVIDER_CONCURRENCY: ${{ inputs.provider_concurrency || vars.GADFLY_DEFAULT_PROVIDER_CONCURRENCY }} + # NB: GADFLY_PROVIDER_CONCURRENCY (the old model cap) is intentionally no + # longer forwarded — entrypoint.sh ignores it. The lens budget is the one + # throttle now, shared across a provider's models. GADFLY_PROVIDER_LENS_CONCURRENCY: ${{ inputs.provider_lens_concurrency || vars.GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY }} GADFLY_TIMEOUT_SECS: ${{ inputs.timeout_secs }} GADFLY_MAX_STEPS: ${{ inputs.max_steps }} diff --git a/CLAUDE.md b/CLAUDE.md index 8df96f4..ddb8d20 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -57,7 +57,7 @@ Dockerfile multi-stage; private-module creds via BuildKit secrets ne .gitea/workflows/build-image.yml push main → :latest; tag v* → :+:latest; PR → build-only .gitea/workflows/review-reusable.yml reusable (workflow_call) review job; resolves swarm config at RUNTIME: consumer `with:` input → owner user-scope var (GADFLY_DEFAULT_MODELS / - _SPECIALISTS / _PROVIDER_CONCURRENCY / _PROVIDER_LENS_CONCURRENCY, + + _SPECIALISTS / _PROVIDER_LENS_CONCURRENCY, + GADFLY_ENDPOINT_RAGNAROS) → image default. Vars are injected per-run, so editing one var retunes the whole fleet even though long-lived act_runners CACHE this file by ref (a moved tag is NOT re-fetched — only a runtime value or a fresh @ @@ -154,10 +154,19 @@ are actually exercised. OpenAI/Anthropic/Google come from majordomo's abstractio NOT re-pulled, so the job silently runs the previous image. For a run that must use a specific build (e.g. validating a just-pushed fix), pin the consumer stub to the immutable `:sha-` tag the build publishes, not `:latest`. -- **Concurrency is per-provider** (`entrypoint.sh`): each provider is a lane, lanes run in - parallel, `cap` (from `GADFLY_PROVIDER_CONCURRENCY` else `GADFLY_CONCURRENCY`, default 1) bounds - models-at-once within a lane. The review timeout (`GADFLY_TIMEOUT_SECS`) is **per-lens**, not - shared across the suite — a slow model can't starve later lenses (the original timeout bug). +- **Concurrency is one per-provider lens budget** (`entrypoint.sh` + `cmd/gadfly/lenssem.go`): + each provider is a lane, lanes run in parallel, and within a lane ALL of the provider's models + run at once — the only throttle is a **provider-wide lens budget** (max lens passes in flight, + a lens = one specialist's review+recheck). The budget comes from + `GADFLY_PROVIDER_LENS_CONCURRENCY` (`provider=N` map) else `GADFLY_LENS_CONCURRENCY` (default 1). + Because models are separate processes, the budget is a **cross-process permit pool**: entrypoint + seeds a per-lane dir of N flock files; each model's binary acquires one before a lens pass and + releases it after (flock auto-drops on process death). This replaced the old two-level + `GADFLY_PROVIDER_CONCURRENCY` MODEL cap × per-model `GADFLY_LENS_CONCURRENCY`, which multiplied + and let a model hold its slot through its last lens — stalling the next model with idle lens + capacity. Those two model-cap vars are now **ignored**. The review timeout + (`GADFLY_TIMEOUT_SECS`) is **per-lens**, not shared across the suite — a slow lens can't starve + the others (the original timeout bug). - **Large-PR token burn**: the agent loop re-sends the whole transcript every step, so a giant diff (the old `get_diff` dumped it untruncated, and it was embedded in both the review and recheck task) was re-transmitted ~steps × lenses × passes × models times — a ~250 K-token PR diff --git a/README.md b/README.md index a7dc0ef..3a333b3 100644 --- a/README.md +++ b/README.md @@ -272,38 +272,32 @@ Unset = no delegation (current behavior). ### Concurrency (per-provider lanes) With multiple models, each **provider** is its own lane and lanes run in **parallel**, so a fast -cloud provider isn't stuck behind a slow local box. Within a lane, at most `cap` models run at -once — `cap` comes from `GADFLY_PROVIDER_CONCURRENCY` (a `provider=N` map) else `GADFLY_CONCURRENCY` -(default `1`). The timeout is **per-lens** (`GADFLY_TIMEOUT_SECS`), so a slow model on one lens -can't starve the others. +cloud provider isn't stuck behind a slow local box. There is **one throttle**: a per-provider +**lens budget** — the max number of lens passes (a lens = one specialist's review+recheck) in +flight at once for that provider. Every model in the lane runs concurrently and its lenses draw +from that single shared budget, so nothing else caps how many models run. The budget comes from +`GADFLY_PROVIDER_LENS_CONCURRENCY` (a `provider=N` map) else the `GADFLY_LENS_CONCURRENCY` scalar +(default `1`). The timeout is **per-lens** (`GADFLY_TIMEOUT_SECS`), so a slow lens can't starve +the others. ```yaml -# One local box (serial — it serves one model at a time) + 3 cloud reviews at once, -# both lanes running concurrently: -GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=3,m1pro=1" +# The local box gets 1 lens at a time (serial); the cloud lane runs up to 3 lens passes at once, +# shared across ALL its models. Both lanes run concurrently. +GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1pro=1" GADFLY_MODELS: "m1pro/qwen3:14b,qwen3-coder:480b-cloud,gpt-oss:120b-cloud" ``` A model's provider is the spec's first segment (`m1pro/…` → `m1pro`), or `GADFLY_PROVIDER`/ -`ollama-cloud` for a bare id. Default (`cap 1`) keeps a single-provider pool fully sequential. +`ollama-cloud` for a bare id. The budget is **shared across the provider's models**: with a +budget of 3 and two cloud models, you get 3 lens passes in flight in any mix — as one model +finishes a lens, the freed slot immediately goes to another model's next lens, so a model +winding down to its last lens never stalls the others (the pre-2026-07 design capped *models* +separately and did stall — that `GADFLY_PROVIDER_CONCURRENCY`/`GADFLY_CONCURRENCY` model cap is +**gone**; those vars are now ignored). Default (budget `1`) keeps a provider fully sequential. -**Lens fan-out (within a model).** By default the specialist lenses run **sequentially** inside -each model (`GADFLY_LENS_CONCURRENCY=1`). Raise it to overlap the independent per-lens -review+recheck passes — the model then posts its consolidated comment as soon as its lenses -finish (so with sequential models, results stream in per model and per-model timings stay -clean). Like the model cap, it's **per-provider configurable**: `GADFLY_PROVIDER_LENS_CONCURRENCY` -takes a `provider=N` map keyed by the **same provider lanes** as `GADFLY_PROVIDER_CONCURRENCY`, -falling back to the `GADFLY_LENS_CONCURRENCY` scalar (default `1`). **It multiplies with the -model cap:** total in-flight requests ≈ *models-at-once × lenses-at-once*, so to fan lenses out -without oversubscribing a backend, keep its model cap low and raise its lens cap: - -```yaml -# Per provider: cloud runs one model at a time but fans its 3 lenses out (3 concurrent requests); -# the slow local box stays fully serial. Both provider lanes still run in parallel. -GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=1,m1=1" -GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1=1" -GADFLY_SPECIALISTS: "security,correctness,error-handling" -``` +> Under the hood the shared budget is a small cross-process permit pool (flock files, seeded per +> lane by `entrypoint.sh`); permits release automatically if a model process dies, so a crashed +> lens can't leak budget. ### Live status board @@ -418,8 +412,7 @@ on its next review **without** a re-pin or a tag move: |---|---| | `GADFLY_DEFAULT_MODELS` | `GADFLY_MODELS` (csv) | | `GADFLY_DEFAULT_SPECIALISTS` | the lens suite | -| `GADFLY_DEFAULT_PROVIDER_CONCURRENCY` | models-at-once per provider | -| `GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY` | lenses-at-once per provider | +| `GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY` | the per-provider lens budget (lens passes in flight per provider, shared across its models) | | `GADFLY_ENDPOINT_RAGNAROS` | a named endpoint, e.g. `llamaswap\|https://host` | Adding a *new* named endpoint still needs a one-line reusable edit (Gitea can't auto-expose arbitrary @@ -441,10 +434,9 @@ The reviewer binary reads these (the stub/entrypoint set sane defaults): | `GADFLY_SELECTOR_MODEL` | review model | model that picks lenses in `auto` mode | | `GADFLY_WORKER_MODEL` | — | cheap model for `delegate_investigation`; unset = no delegation | | `GADFLY_WORKER_MAX_STEPS` | 8 | tool-step cap for a delegated worker run | -| `GADFLY_CONCURRENCY` | 1 | default max models run at once **per provider** | -| `GADFLY_PROVIDER_CONCURRENCY` | — | per-provider overrides, e.g. `ollama-cloud=3,m1pro=1` | -| `GADFLY_LENS_CONCURRENCY` | 1 | specialist lenses run at once **within a model** (× model cap = total in-flight) | -| `GADFLY_PROVIDER_LENS_CONCURRENCY` | — | per-provider lens overrides, same lanes as `GADFLY_PROVIDER_CONCURRENCY`, e.g. `ollama-cloud=3,m1=1` | +| `GADFLY_LENS_CONCURRENCY` | 1 | **per-provider lens budget** — lens passes in flight per provider, shared across all its models (all a provider's models run at once; this is the only throttle) | +| `GADFLY_PROVIDER_LENS_CONCURRENCY` | — | per-provider lens-budget overrides, a `provider=N` map, e.g. `ollama-cloud=3,m1=1` | +| `GADFLY_CONCURRENCY` / `GADFLY_PROVIDER_CONCURRENCY` | — | **removed** (was the per-provider models-at-once cap; now ignored — the lens budget is the single throttle) | | `GADFLY_MAX_STEPS` | 24 | review-pass tool-step cap | | `GADFLY_TIMEOUT_SECS` | 300 | deadline **per specialist lens** (review+recheck) | | `GADFLY_RECHECK` | on | set `0`/`false` to skip the recheck pass | diff --git a/cmd/gadfly/lens_concurrency_test.go b/cmd/gadfly/lens_concurrency_test.go index 4b67a6a..1ea5e81 100644 --- a/cmd/gadfly/lens_concurrency_test.go +++ b/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 diff --git a/cmd/gadfly/lenssem.go b/cmd/gadfly/lenssem.go new file mode 100644 index 0000000..50fd187 --- /dev/null +++ b/cmd/gadfly/lenssem.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "syscall" + "time" +) + +// lensSem is the PROVIDER-WIDE lens-permit pool. Historically each model's binary +// throttled its own lenses in-process (GADFLY_LENS_CONCURRENCY) while entrypoint.sh +// separately capped how many MODELS ran at once — two multiplicative gates in +// different processes. That let a model hold its whole slot until its last lens +// finished, stalling the next model even with idle lens capacity. +// +// Instead, entrypoint.sh now runs all of a provider's models concurrently and +// seeds ONE directory of N permit files per provider (N = the provider's lens +// budget). Every model process in that lane draws from the same pool: a lens pass +// (review + recheck) acquires a permit before it runs and releases it after, so a +// model winding down immediately yields its freed permits to another model's +// queued lenses. Permits are held with flock, which the kernel drops when the +// holding process exits — so a killed/crashed model frees its permits for free. +type lensSem struct { + dir string + size int +} + +// lensSemPollInterval is how often a blocked acquirer re-sweeps the permit files. +// Lens passes run for many seconds to minutes, so a coarse poll adds negligible +// latency while keeping the mechanism a few lines of stdlib (no IPC primitives). +const lensSemPollInterval = 150 * time.Millisecond + +// activeLensSem returns the shared semaphore configured by entrypoint.sh, or nil +// when it isn't set (standalone/local runs, tests, or an older entrypoint) — in +// which case runSpecialists falls back to the in-process fanout limit alone, i.e. +// the pre-existing per-model behavior. +func activeLensSem() *lensSem { + dir := os.Getenv("GADFLY_LENS_SEM_DIR") + if dir == "" { + return nil + } + 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. It FAILS OPEN: if the pool directory is structurally unusable (missing, +// 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 { + 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 <-timer.C: + } + } +} + +// tryAcquire makes one non-blocking sweep over the permit files. It returns a +// release func for the first one it locks; otherwise it distinguishes a healthy +// FULL pool (some permit was openable but flock-busy → ok=false, err=nil → keep +// polling) from a structurally BROKEN pool (no permit was even acquirable and none +// was merely busy → err set → caller fails open). Permit files are created lazily +// (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 { + 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() + } + } + if sawBusy { + return nil, false, nil // full but healthy: another lens will free a permit + } + return nil, false, structural +} diff --git a/cmd/gadfly/lenssem_test.go b/cmd/gadfly/lenssem_test.go new file mode 100644 index 0000000..52cbc69 --- /dev/null +++ b/cmd/gadfly/lenssem_test.go @@ -0,0 +1,136 @@ +package main + +import ( + "context" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +// flock permits are per open-file-description, so two separate opens of the same +// permit file conflict even within one process — the exhaustion, blocking, and +// max-in-flight tests below therefore exercise the same semantics a real +// multi-process lane would see. + +func TestActiveLensSemUnset(t *testing.T) { + t.Setenv("GADFLY_LENS_SEM_DIR", "") + if s := activeLensSem(); s != nil { + t.Fatalf("expected nil sem when GADFLY_LENS_SEM_DIR unset, got %+v", s) + } + // A dir with a zero/blank size is inert too (must not divide the pool by 0). + t.Setenv("GADFLY_LENS_SEM_DIR", t.TempDir()) + t.Setenv("GADFLY_LENS_SEM_SIZE", "0") + if s := activeLensSem(); s != nil { + t.Fatalf("expected nil sem when size < 1, got %+v", s) + } +} + +func TestLensSemExhaustionAndRelease(t *testing.T) { + s := &lensSem{dir: t.TempDir(), size: 2} + + r1, ok, err := s.tryAcquire() + if !ok || err != nil { + t.Fatalf("first acquire should succeed: ok=%v err=%v", ok, err) + } + r2, ok, err := s.tryAcquire() + if !ok || err != nil { + t.Fatalf("second acquire should succeed: ok=%v err=%v", ok, err) + } + // Pool full but healthy: ok=false with NO error (keep polling), not a + // 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, err := s.tryAcquire() + if !ok || err != nil { + 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} + + // Immediate success while a permit is free. + release, err := s.acquire(context.Background()) + if err != nil { + t.Fatalf("acquire on a free pool: %v", err) + } + + // With the only permit held, acquire must block until ctx expires. + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + start := time.Now() + if _, err := s.acquire(ctx); err == nil { + t.Fatal("acquire on an exhausted pool should return ctx error, not a permit") + } + if waited := time.Since(start); waited < 50*time.Millisecond { + t.Fatalf("acquire returned too fast (%v); it should have blocked on the full pool", waited) + } + release() +} + +func TestLensSemNeverExceedsSize(t *testing.T) { + const size = 3 + s := &lensSem{dir: t.TempDir(), size: size} + + var inFlight, peak int64 + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { + defer wg.Done() + release, err := s.acquire(context.Background()) + if err != nil { + t.Errorf("acquire: %v", err) + return + } + defer release() + n := atomic.AddInt64(&inFlight, 1) + for { + p := atomic.LoadInt64(&peak) + if n <= p || atomic.CompareAndSwapInt64(&peak, p, n) { + break + } + } + time.Sleep(20 * time.Millisecond) + atomic.AddInt64(&inFlight, -1) + }() + } + wg.Wait() + + if peak > size { + t.Fatalf("max in-flight lens permits = %d, exceeds budget %d", peak, size) + } + if peak == 0 { + t.Fatal("no permits were ever acquired") + } +} diff --git a/cmd/gadfly/main.go b/cmd/gadfly/main.go index 59e6bb8..006bba9 100644 --- a/cmd/gadfly/main.go +++ b/cmd/gadfly/main.go @@ -49,15 +49,18 @@ // GADFLY_RECHECK set to 0/false to skip the recheck pass (optional, default on). // GADFLY_RECHECK_MAX_STEPS recheck-pass step cap (optional, default 16). // GADFLY_TIMEOUT_SECS overall deadline in seconds, shared by both passes (optional, default 300). -// GADFLY_LENS_CONCURRENCY how many specialist lenses run concurrently within this -// model (optional, default 1 = sequential). Total in-flight -// model requests ≈ this × entrypoint.sh's per-provider model -// concurrency, so keep the product within the backend's budget. +// GADFLY_LENS_CONCURRENCY how many specialist lenses run concurrently (optional, +// default 1 = sequential). Under entrypoint.sh this is the +// PROVIDER-WIDE lens budget, shared across all of that +// provider's models via a permit pool (GADFLY_LENS_SEM_DIR), +// so it bounds total lens passes in flight per provider. // GADFLY_PROVIDER_LENS_CONCURRENCY per-provider override for the above, as a -// "provider=N,provider=N" map keyed by the SAME provider -// lanes as GADFLY_PROVIDER_CONCURRENCY (e.g. -// "ollama-cloud=3,m1=1"). Wins over GADFLY_LENS_CONCURRENCY -// for the model's provider; falls back to it otherwise. +// "provider=N,provider=N" map keyed by the provider lanes +// (e.g. "ollama-cloud=3,m1=1"). Wins over +// GADFLY_LENS_CONCURRENCY for the model's provider. +// GADFLY_LENS_SEM_DIR / GADFLY_LENS_SEM_SIZE set by entrypoint.sh: the shared +// cross-process lens-permit pool (dir of N flock files) +// and its size. Unset => in-process lensConcurrency only. // GADFLY_MAX_DIFF_CHARS diff chars embedded in the review prompt (optional, default 60000; // the full diff is reachable via the paginated get_diff tool). // @@ -94,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 - // single model. 1 keeps the suite sequential (the historical behavior); - // higher values overlap the independent per-lens passes. See runSpecialists. + // defaultLensConcurrency is the fallback lens budget when neither + // GADFLY_PROVIDER_LENS_CONCURRENCY nor GADFLY_LENS_CONCURRENCY is set. 1 keeps + // 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 ) @@ -242,25 +247,34 @@ 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 -// default of 1 keeps the suite sequential, exactly as before), and fanout.Run -// returns one result per lens in input order. Each lens already runs under its -// own per-lens timeout (reviewWithSpecialist) and the lenses only read the -// immutable repoFS, so concurrency simply overlaps independent passes. +// fanout primitive to overlap independent lens passes (fanout.Run returns one +// result per lens in input order); each lens runs under its own per-lens timeout +// (reviewWithSpecialist) and the lenses only read the immutable repoFS. // -// Caution: this fans out WITHIN one model. It multiplies with entrypoint.sh's -// per-provider model concurrency, so total concurrent backend requests ≈ -// (models at once) × (lenses at once). To fan lenses out without oversubscribing -// the backend, run models one at a time (provider lane cap 1) and raise this. +// Throttling: when entrypoint.sh runs several of a provider's models at once it +// seeds a shared lens-permit pool (activeLensSem) that every model's lenses draw +// from, so the real cap is total lens passes in flight per PROVIDER — not +// (models at once) × (lenses at once). Absent that pool (local runs, tests) the +// 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 // is set, so plain runs are unaffected. sw := newStatusWriter(os.Getenv("GADFLY_MODEL"), modelProvider(), specialists) + // 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 // the shared pool is the real ceiling + } + fanResults := fanout.Run(context.Background(), specialists, fanout.Options[Specialist]{ - MaxConcurrent: lensConcurrency(), - }, func(_ context.Context, sp Specialist) (res specialistResult, _ error) { + MaxConcurrent: maxConcurrent, + }, func(ctx context.Context, sp Specialist) (res specialistResult, _ error) { // A panic in one lens must not crash the whole binary (which would kill // every other lens's output) or leave this lens stuck at "running" on the // status board. fanout does not recover fn panics, so we do it here: @@ -271,6 +285,17 @@ func runSpecialists(eng reviewEngine, base string, specialists []Specialist, tas sw.set(sp.Name, lensFinished, "", true) } }() + // Hold a provider-wide permit for this lens's whole review+recheck pass. + // While waiting the lens stays "queued" on the board; cancellation before + // a permit frees surfaces as a did-not-run lens rather than a leaked slot. + if sem != nil { + release, err := sem.acquire(ctx) + if err != nil { + sw.set(sp.Name, lensFinished, "", true) + return specialistResult{spec: sp, out: fmt.Sprintf("⚠️ This reviewer did not run: %v", err), verdict: verdictUnknown, errored: true}, nil + } + defer release() + } sw.set(sp.Name, lensRunning, "", false) out, errored := reviewWithSpecialist(eng, base, sp, task, diff) v := parseVerdict(out) @@ -293,14 +318,13 @@ func runSpecialists(eng reviewEngine, base string, specialists []Specialist, tas return results } -// lensConcurrency resolves how many specialist lenses run at once for THIS run's -// model. It mirrors entrypoint.sh's per-provider MODEL concurrency: a -// per-provider override in GADFLY_PROVIDER_LENS_CONCURRENCY ("provider=N,...") -// wins for the model's provider, otherwise the GADFLY_LENS_CONCURRENCY scalar -// (default 1). The provider is resolved by modelProvider() — the SAME lane rule -// entrypoint uses for GADFLY_PROVIDER_CONCURRENCY — so e.g. -// "ollama-cloud=3,m1=1" fans cloud lenses out while keeping a slow local box -// serial, exactly the way the model map does for whole models. +// lensConcurrency resolves the lens budget for THIS run's provider: a per-provider +// override in GADFLY_PROVIDER_LENS_CONCURRENCY ("provider=N,...") wins for the +// model's provider (resolved by modelProvider()), otherwise the +// GADFLY_LENS_CONCURRENCY scalar (default 1). Under entrypoint.sh the SAME value +// seeds the shared cross-process permit pool (activeLensSem), so it is the +// provider-wide budget rather than a per-model one; standalone it caps the single +// model's in-process fanout. func lensConcurrency() int { if n, ok := providerOverride("GADFLY_PROVIDER_LENS_CONCURRENCY", modelProvider()); ok { return n @@ -310,7 +334,7 @@ func lensConcurrency() int { // providerOverride parses a "provider=N,provider=N" env map and returns the // value for provider when present and valid (>0). Mirrors entrypoint.sh's -// provider_cap lookup so the two concurrency maps share one syntax. +// provider_lens_cap lookup so the two share one syntax. func providerOverride(envName, provider string) (int, bool) { for _, item := range strings.Split(os.Getenv(envName), ",") { k, v, ok := strings.Cut(item, "=") diff --git a/cmd/gadfly/model.go b/cmd/gadfly/model.go index c969d10..546641c 100644 --- a/cmd/gadfly/model.go +++ b/cmd/gadfly/model.go @@ -162,8 +162,8 @@ func buildSpec(provider, model string) string { // entrypoint.sh's provider_of: the segment before the first "/" in GADFLY_MODEL, // else GADFLY_PROVIDER, else the default (ollama-cloud). The binary reviews one // model per invocation, so this is that model's provider — used to resolve -// per-provider policy (e.g. lens concurrency) against the SAME provider keys -// entrypoint uses for GADFLY_PROVIDER_CONCURRENCY. +// per-provider policy (e.g. the lens budget) against the SAME provider keys +// entrypoint uses for GADFLY_PROVIDER_LENS_CONCURRENCY. func modelProvider() string { model := strings.TrimSpace(os.Getenv("GADFLY_MODEL")) if pfx, _, ok := strings.Cut(model, "/"); ok { diff --git a/entrypoint.sh b/entrypoint.sh index 6f5f81d..43147de 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -198,14 +198,18 @@ export GADFLY_FINDINGS_TOKEN="${GADFLY_FINDINGS_TOKEN:-}" # provider key envs (OPENAI_API_KEY, …) are inherited by run.sh and the binary. # # Concurrency: each PROVIDER is its own lane and lanes run in PARALLEL, so a fast -# cloud provider isn't stuck behind a slow local box. Within a lane, at most -# `cap` models run at once. cap = GADFLY_PROVIDER_CONCURRENCY's "provider=N" -# entry, else GADFLY_CONCURRENCY (default 1). A model's provider is the spec's -# first path segment ("m1pro/qwen3.6:35b-mlx" -> m1pro), or GADFLY_PROVIDER / -# ollama-cloud for a bare id. Default (cap 1) keeps a single-provider pool fully -# sequential, exactly as before. +# cloud provider isn't stuck behind a slow local box. Within a lane ALL of the +# provider's models run at once; the real throttle is a single PROVIDER-WIDE lens +# budget (a shared permit pool, seeded per lane) that every model's lenses draw +# from — so total lens passes in flight per provider is bounded, but a model +# winding down to its last lens immediately yields its freed permits to another +# model's queued lenses instead of holding a whole "model slot" (the old +# GADFLY_PROVIDER_CONCURRENCY model cap, now removed, caused that tail stall). The +# budget = GADFLY_PROVIDER_LENS_CONCURRENCY's "provider=N" entry, else +# GADFLY_LENS_CONCURRENCY (default 1). A model's provider is the spec's first path +# segment ("m1pro/qwen3.6:35b-mlx" -> m1pro), or GADFLY_PROVIDER / ollama-cloud +# for a bare id. MODELS="${GADFLY_MODELS:-${OLLAMA_REVIEW_MODELS:-$DEFAULT_MODELS}}" -DEFAULT_CONC="${GADFLY_CONCURRENCY:-1}" # --- huge-PR downshift ------------------------------------------------------ # A very large diff is what burns the model budget: every review step re-sends @@ -245,24 +249,33 @@ provider_of() { case "$1" in */*) echo "${1%%/*}";; *) echo "${GADFLY_PROVIDER:- STATUS_DIR="${WORKDIR}/status" status_file_for() { echo "${STATUS_DIR}/$(echo "$1" | tr -c '[:alnum:]._-' '_').json"; } -provider_cap() { # provider -> concurrency (override map "p=N,...", else default) +# Root of the per-provider lens permit pools (one subdir per lane, seeded by +# run_lane). Cleared up front so a reused WORKDIR can't leak stale permit files. +LENS_SEM_ROOT="${WORKDIR}/lenssem" +rm -rf "$LENS_SEM_ROOT" 2>/dev/null || true + +provider_lens_cap() { # provider -> provider-wide lens budget (permit-pool size) local p="$1" item k v - IFS=',' read -ra _caps <<< "${GADFLY_PROVIDER_CONCURRENCY:-}" - for item in "${_caps[@]}"; do + IFS=',' read -ra _lcaps <<< "${GADFLY_PROVIDER_LENS_CONCURRENCY:-}" + for item in "${_lcaps[@]}"; do k="$(echo "${item%%=*}" | tr -d '[:space:]')" v="$(echo "${item#*=}" | tr -d '[:space:]')" if [ "$k" = "$p" ] && [ -n "$v" ]; then echo "$v"; return; fi done - echo "$DEFAULT_CONC" + echo "${GADFLY_LENS_CONCURRENCY:-1}" } review_one() { - local sf="" ff="" - [ "${GADFLY_STATUS_BOARD:-1}" != "0" ] && sf="$(status_file_for "$1")" - [ "$CONSOLIDATE" = "1" ] && ff="$(findings_file_for "$1")" - PROVIDER=ollama MODEL="$1" GADFLY_BIN="/usr/local/bin/gadfly" GADFLY_REPO_DIR="$REPO_DIR" \ + local m="$1" sem_dir="${2:-}" sem_size="${3:-}" sf="" ff="" + [ "${GADFLY_STATUS_BOARD:-1}" != "0" ] && sf="$(status_file_for "$m")" + [ "$CONSOLIDATE" = "1" ] && ff="$(findings_file_for "$m")" + # GADFLY_LENS_SEM_DIR/_SIZE point the binary at this provider's shared lens + # permit pool (empty => the binary just uses its in-process lens limit). These + # are inherited by the binary through run.sh's environment. + PROVIDER=ollama MODEL="$m" GADFLY_BIN="/usr/local/bin/gadfly" GADFLY_REPO_DIR="$REPO_DIR" \ GADFLY_STATUS_FILE="$sf" GADFLY_FINDINGS_OUT="$ff" GADFLY_CONSOLIDATE="$CONSOLIDATE" \ - bash "${SCRIPTS_DIR}/run.sh" || log "model $1 failed (continuing)" + GADFLY_LENS_SEM_DIR="$sem_dir" GADFLY_LENS_SEM_SIZE="$sem_size" \ + bash "${SCRIPTS_DIR}/run.sh" || log "model $m failed (continuing)" # If the binary never wrote real status (run.sh skipped it: empty diff, no key, # binary missing), the pre-seed stays {started:0, done:false} and the board # would show this model "waiting to start" forever and never reach N/N. Mark @@ -313,16 +326,19 @@ for m in "${MODEL_LIST[@]}"; do case " $PROVIDERS " in *" $p "*) ;; *) PROVIDERS="${PROVIDERS}${PROVIDERS:+ }$p" ;; esac done -run_lane() { # $1=provider: run its models, at most `cap` at a time - local p="$1" cap inflight=0 m - cap="$(provider_cap "$p")"; [ "$cap" -ge 1 ] 2>/dev/null || cap=1 +run_lane() { # $1=provider: run ALL its models at once, throttled only by a shared + # provider-wide lens permit pool (no per-model cap). + local p="$1" budget sem_dir m + budget="$(provider_lens_cap "$p")"; [ "$budget" -ge 1 ] 2>/dev/null || budget=1 local mine=() for m in "${MODEL_LIST[@]}"; do [ "$(provider_of "$m")" = "$p" ] && mine+=("$m"); done - log "lane ${p}: cap ${cap}; models: ${mine[*]}" + # Seed this provider's lens permit pool: a directory the binary flocks N permit + # files in (created lazily), one shared budget across every model in the lane. + sem_dir="${LENS_SEM_ROOT}/$(echo "$p" | tr -c '[:alnum:]._-' '_')" + mkdir -p "$sem_dir" + log "lane ${p}: lens budget ${budget} shared across ${#mine[@]} model(s): ${mine[*]}" for m in "${mine[@]}"; do - review_one "$m" & - inflight=$((inflight+1)) - if [ "$inflight" -ge "$cap" ]; then wait -n 2>/dev/null || wait; inflight=$((inflight-1)); fi + review_one "$m" "$sem_dir" "$budget" & done wait } @@ -338,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 - # binary overwrites its own file with real per-lens detail once it starts. + # even models whose lenses are still waiting on their provider's lens budget. + # 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:[]}' \ @@ -378,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 -# the lane PIDs so we wait ONLY for the review work — not the status board, -# which intentionally runs until we signal it below. +# Each provider lane runs in parallel; the shared lens budget throttles within +# each lane. Track the lane PIDs so we wait ONLY for the review work — not the +# status board, which intentionally runs until we signal it below. LANE_PIDS=() for p in $PROVIDERS; do run_lane "$p" & diff --git a/examples/adversarial-review.yml b/examples/adversarial-review.yml index 82befb7..37944cd 100644 --- a/examples/adversarial-review.yml +++ b/examples/adversarial-review.yml @@ -55,14 +55,14 @@ jobs: # csv to choose; "all" for everything; or define custom ones via a repo # .gadfly.yml / GADFLY_SPECIALIST_. See README "Specialists". GADFLY_SPECIALISTS: ${{ vars.GADFLY_SPECIALISTS }} - # Lens fan-out (optional; default 1 = lenses run sequentially within a - # model). Raise it to run a model's lenses concurrently so each model - # posts its comment sooner. Total in-flight requests = (models at once) - # × (lenses at once), so to fan out without oversubscribing a backend, - # keep its model cap low and raise its lens cap. Per-provider configurable - # via GADFLY_PROVIDER_LENS_CONCURRENCY (same lanes as the model map): - # GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=1,m1=1" + # Concurrency (optional; default 1 = fully sequential per provider). The + # ONE throttle is a per-provider LENS BUDGET: the max lens passes (a lens = + # one specialist's review+recheck) in flight at once for a provider, shared + # across ALL that provider's models — every model in a lane runs at once and + # its lenses draw from the shared budget. Raise it to overlap lenses; set it + # per provider with GADFLY_PROVIDER_LENS_CONCURRENCY: # GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1=1" + # (The old GADFLY_PROVIDER_CONCURRENCY model cap was removed and is ignored.) # GADFLY_LENS_CONCURRENCY: ${{ vars.GADFLY_LENS_CONCURRENCY }} # GADFLY_PROVIDER_LENS_CONCURRENCY: ${{ vars.GADFLY_PROVIDER_LENS_CONCURRENCY }} # Live status board (optional; ON by default): one consolidated comment