From 74831368ab804f63520f94f79844cc2f1decc400 Mon Sep 17 00:00:00 2001 From: Steve Dudenhoeffer Date: Sat, 18 Jul 2026 12:22:11 -0400 Subject: [PATCH] feat(concurrency): provider-wide lens budget, drop the model cap Concurrency was two multiplicative gates in two processes: entrypoint.sh capped MODELS-at-once per provider (GADFLY_PROVIDER_CONCURRENCY) while each model's binary separately capped its own lenses (GADFLY_LENS_CONCURRENCY). A model therefore held its whole model-slot until its LAST lens finished, stalling the next model even with idle lens capacity. Collapse to one throttle: a provider-wide lens budget shared across all of that provider's models. entrypoint now runs every model in a lane at once and seeds a single cross-process permit pool per lane (a dir of N flock files, sized by GADFLY_PROVIDER_LENS_CONCURRENCY -> GADFLY_LENS_CONCURRENCY). Each 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. flock auto-releases on process death, so a killed/crashed model can't leak budget. - cmd/gadfly/lenssem.go: the flock permit pool (+ lenssem_test.go). - main.go: runSpecialists holds a shared permit per lens; fanout sized to the budget so a lone model can use all of it. Falls back to the in-process limit when no pool is set (local runs, tests). - entrypoint.sh: drop provider_cap/DEFAULT_CONC; run_lane runs all models and seeds the per-lane pool. - GADFLY_PROVIDER_CONCURRENCY / GADFLY_CONCURRENCY are now ignored; the reusable workflow marks provider_concurrency deprecated and stops forwarding it. Docs (README, CLAUDE.md, examples) updated per the maintenance rule. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitea/workflows/review-reusable.yml | 10 ++- CLAUDE.md | 19 +++-- README.md | 54 ++++++------- cmd/gadfly/lenssem.go | 86 +++++++++++++++++++++ cmd/gadfly/lenssem_test.go | 110 +++++++++++++++++++++++++++ cmd/gadfly/main.go | 69 +++++++++++------ cmd/gadfly/model.go | 4 +- entrypoint.sh | 62 +++++++++------ examples/adversarial-review.yml | 14 ++-- 9 files changed, 333 insertions(+), 95 deletions(-) create mode 100644 cmd/gadfly/lenssem.go create mode 100644 cmd/gadfly/lenssem_test.go 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/lenssem.go b/cmd/gadfly/lenssem.go new file mode 100644 index 0000000..ad806a4 --- /dev/null +++ b/cmd/gadfly/lenssem.go @@ -0,0 +1,86 @@ +package main + +import ( + "context" + "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 { + 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 +// surface the lens as "did not run" rather than leaking a permit. +func (s *lensSem) acquire(ctx context.Context) (func(), error) { + for { + if release, ok := s.tryAcquire(); ok { + return release, nil + } + select { + case <-ctx.Done(): + return func() {}, ctx.Err() + case <-time.After(lensSemPollInterval): + } + } +} + +// tryAcquire makes one non-blocking sweep over the permit files, returning a +// release func for the first one it locks. The permit files are created lazily +// (entrypoint.sh only guarantees the directory exists), so a fresh pool needs no +// seeding step beyond mkdir. 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) { + 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 { + continue + } + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + f.Close() + continue + } + return func() { f.Close() }, true + } + return nil, false +} diff --git a/cmd/gadfly/lenssem_test.go b/cmd/gadfly/lenssem_test.go new file mode 100644 index 0000000..f42472f --- /dev/null +++ b/cmd/gadfly/lenssem_test.go @@ -0,0 +1,110 @@ +package main + +import ( + "context" + "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 := s.tryAcquire() + if !ok { + t.Fatal("first acquire should succeed") + } + r2, ok := s.tryAcquire() + if !ok { + t.Fatal("second acquire should succeed") + } + if _, ok := s.tryAcquire(); ok { + t.Fatal("third acquire should fail: pool of 2 is exhausted") + } + + r1() // free one permit + r3, ok := s.tryAcquire() + if !ok { + t.Fatal("acquire should succeed after a release") + } + r2() + r3() +} + +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..9aa19fb 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). // @@ -248,19 +251,29 @@ func run() error { // own per-lens timeout (reviewWithSpecialist) and the lenses only read the // immutable repoFS, so concurrency simply overlaps independent passes. // -// 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. +// Provider-wide 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) this falls back to the in-process lensConcurrency() limit alone. 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. + sem := activeLensSem() + maxConcurrent := lensConcurrency() + if sem != nil { + maxConcurrent = sem.size + } + 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 +284,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 +317,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 +333,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..f6dedf4 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 } 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