diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a742414..dbe8489 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -57,6 +57,9 @@ A cron job runs the sweep every minute (`*/1 * * * *`); the sweep self-locks so `workspace-write` sandbox restricted to `/tmp` with **network off and no `gh`**. Codex reads the rubric + corpus + the inlined diff and writes the review to a temp file ending in the marker; the *runner*, not Codex, posts it with `gh pr comment`. + Candidates are collected serially (all the cheap gh gates), then reviewed by a pool of up to `CODEX_JOBS` + (default 3) concurrent codex runs — a busy sweep's wall-clock is the slowest review, not the sum of them. A + quota wall from any run stops new launches while in-flight runs drain. 6. **Cap.** `MAX_PRS` limits PRs *actually reviewed* per sweep, counted only after the cheap dedup skips, so a backlog of already-reviewed PRs at the front of the list can't starve later ones. diff --git a/src/review-sweep.test.ts b/src/review-sweep.test.ts index 2f77a0e..92e26ac 100644 --- a/src/review-sweep.test.ts +++ b/src/review-sweep.test.ts @@ -33,6 +33,7 @@ const cfg = (): Config => ({ statusAppKeyPath: '', gatewayPool: '', rotateCooldownMs: 600_000, + codexJobs: 3, }) const pr = (number: number, sha: string): Pr => ({ diff --git a/src/review-sweep.ts b/src/review-sweep.ts index 412029b..5fcd9d7 100755 --- a/src/review-sweep.ts +++ b/src/review-sweep.ts @@ -51,6 +51,7 @@ export interface Config { githubStatusContext: string gatewayPool: string // CODEX_GATEWAY_POOL: ordered comma-separated gateway hostnames codex may rotate through; empty = rotation off rotateCooldownMs: number // min gap between gateway rotations, so a fully-drained pool cycles calmly instead of thrashing + codexJobs: number // concurrent codex reviews per sweep; the gh I/O around them stays serial } function loadConfig(): Config { @@ -115,6 +116,7 @@ function loadConfig(): Config { statusAppKeyPath: pick('GITHUB_STATUS_APP_KEY', '').trim(), gatewayPool: pick('CODEX_GATEWAY_POOL', ''), rotateCooldownMs: int('CODEX_ROTATE_COOLDOWN_MIN', 10, 0) * 60_000, + codexJobs: int('CODEX_JOBS', 3, 1), // a single codex run takes minutes; a small pool keeps a busy sweep from serializing them } } @@ -949,10 +951,26 @@ export type ReviewOutcome = | { kind: 'fixed'; tokens: number | null } // codex emitted the fixed token → prior findings resolved | { kind: 'review'; text: string; tokens: number | null } // a real review, sign-off already stripped (no marker yet) +// Async twin of the kit's `exec`, same result shape — ONLY for the codex child, so several multi-minute reviews +// can run at once while every gh call around them stays the kit's sync exec. +async function execAsync(cmd: string, args: string[], opts: { cwd: string; input: string; timeoutMs: number }): Promise<{ ok: boolean; stdout: string; combined: string }> { + try { + const child = Bun.spawn([cmd, ...args], { cwd: opts.cwd, stdin: new TextEncoder().encode(opts.input), stdout: 'pipe', stderr: 'pipe' }) + const timer = setTimeout(() => child.kill(), opts.timeoutMs) + const [stdout, stderr, code] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited]) + clearTimeout(timer) + let combined = stdout + stderr + if (child.signalCode) combined += `\n${cmd}: process killed by ${child.signalCode} (timeout ${opts.timeoutMs}ms)` + return { ok: code === 0 && child.signalCode === null, stdout, combined } + } catch (e) { + return { ok: false, stdout: '', combined: `${cmd}: ${e instanceof Error ? e.message : String(e)}` } // spawn failure (ENOENT etc.) + } +} + /** Run codex over one PR's diff and classify the result. Does NO gh I/O and NO posting — codex runs sandboxed with * no network of its own and /tmp-only writes, so a prompt-injected diff can at worst make it write a junk review * file: it cannot exfiltrate, touch the gh token, or run commands. Callers decide what to do with the outcome. */ -export function runReview(cfg: Config, pr: Pr, priorThread: string, diff: string, dismissed: string[] = []): ReviewOutcome { +export async function runReview(cfg: Config, pr: Pr, priorThread: string, diff: string, dismissed: string[] = []): Promise { const outPath = reviewOutPath(cfg, pr) rmSync(outPath, { force: true }) // clear any stale file so we never read a previous run's review const codexArgs = [ @@ -972,7 +990,7 @@ export function runReview(cfg: Config, pr: Pr, priorThread: string, diff: string if (cfg.codexModel) codexArgs.push('-c', `model=${cfg.codexModel}`) codexArgs.push('-') // read the prompt from STDIN, not argv — the inlined corpus + diff would blow ARG_MAX (E2BIG) - const cx = exec('codex', codexArgs, { cwd: cfg.repoDir, timeoutMs: 1_200_000, input: reviewPrompt(cfg, pr, priorThread, diff, dismissed) }) + const cx = await execAsync('codex', codexArgs, { cwd: cfg.repoDir, timeoutMs: 1_200_000, input: reviewPrompt(cfg, pr, priorThread, diff, dismissed) }) appendFileSync(LOG, `${cx.combined}\n`) let review = cx.ok && existsSync(outPath) ? readFileSync(outPath, 'utf8').trim() : '' // Flake recovery: codex declared convergence in its final message but skipped the file write. Accept a spoken @@ -1008,9 +1026,9 @@ export function commitStatusForSweepResult(result: number | 'clean' | 'fixed' | * 'fixed' when it resolved prior findings, 'limit' on exhaustion, or null on a failure the caller throttles. * Every ✅ that posts is honest: it only fires when no stupify finding is open — "nothing new while findings * still stand" stays silent (those threads remain open); a fix resolves the threads and posts a visible note. */ -function reviewPr(cfg: Config, pr: Pr, priorThread: string, diff: string, firstReview: boolean, openThreadIds: string[], dismissed: string[]): SweepReviewResult { +async function reviewPr(cfg: Config, pr: Pr, priorThread: string, diff: string, firstReview: boolean, openThreadIds: string[], dismissed: string[]): Promise { log(`reviewing PR #${pr.number} @ ${pr.headRefOid.slice(0, 8)}`) - const r = runReview(cfg, pr, priorThread, diff, dismissed) + const r = await runReview(cfg, pr, priorThread, diff, dismissed) if (r.kind === 'limit' || r.kind === 'fail') { log(` review FAILED for #${pr.number} — ${r.reason}`) if (r.kind === 'limit') { @@ -1099,7 +1117,7 @@ function reviewPr(cfg: Config, pr: Pr, priorThread: string, diff: string, firstR /** `stupify review ` — review ONE pull request on demand (no cron, no checkout) and print it, or `--post` it. * Reviews from the inlined diff with a FRESH perspective (no prior-review memory), so you always get the full take. * Accepts a PR URL or `owner/repo#123` (the CLI resolves a bare `#123` against the cwd repo before calling here). */ -function reviewOne(cfg: Config, ref: string, post: boolean): void { +async function reviewOne(cfg: Config, ref: string, post: boolean): Promise { const url = ref.match(/github\.com\/([^/\s]+\/[^/\s]+)\/(?:pull|issues)\/(\d+)/i) const short = ref.match(/^([A-Za-z0-9._-]+\/[A-Za-z0-9._-]+)[#/](\d+)$/) const slug = url?.[1] ?? short?.[1] ?? '' @@ -1137,7 +1155,7 @@ function reviewOne(cfg: Config, ref: string, post: boolean): void { process.exit(1) } console.error(`reviewing ${slug}#${number} …`) // progress on stderr; stdout stays just the review - const r = runReview(cfg, pr, '', diff) // no memory: a manual review is always a fresh, full take + const r = await runReview(cfg, pr, '', diff) // no memory: a manual review is always a fresh, full take if (r.kind === 'limit' || r.kind === 'fail') { console.error(`stupify review: ${r.kind === 'limit' ? 'codex is out of credits / rate-limited' : "codex couldn't produce a review"} — ${r.reason}`) process.exit(1) @@ -1185,7 +1203,7 @@ function failureReason(out: string): string { return cleaned || 'codex run failed (no output captured — check the sweep log)' } -function main(): void { +async function main(): Promise { const cfg = loadConfig() // also mkdirs stateDir and sets LOG, so config warnings are already captured const ref = process.env.REVIEW_PR if (ref) return reviewOne(cfg, ref, process.env.REVIEW_POST === '1') // `stupify review ` — one-shot, no sweep/lock/checkout @@ -1258,12 +1276,17 @@ function main(): void { let reviewed = 0 let tokens = 0 // Count PRs we do real (costly) work on, and cap THAT at MAX_PRS — so a backlog of already-reviewed PRs at - // the front of the list can't consume the budget and starve later ones. + // the front of the list can't consume the budget and starve later ones. Candidates are collected here (all the + // cheap serial gates) and reviewed below by a pool of CODEX_JOBS concurrent codex runs — the sweep's wall-clock + // was dominated by running those multi-minute reviews strictly one after another. let handled = 0 + // Each candidate is one codex run, so the daily ceiling gates collection up front. + const dailyBudget = cfg.maxReviewsPerDay > 0 && !cfg.dryRun ? cfg.maxReviewsPerDay - daily.count : Number.POSITIVE_INFINITY + const candidates: { pr: Pr; prior: PriorState; diff: string; lines: number; firstReview: boolean }[] = [] for (let i = 0; i < queue.length; i++) { const pr = queue[i] if (pr === undefined) continue - if (cfg.maxReviewsPerDay > 0 && !cfg.dryRun && daily.count >= cfg.maxReviewsPerDay) { + if (handled >= dailyBudget) { log(`daily cap hit (MAX_REVIEWS_PER_DAY=${cfg.maxReviewsPerDay}) — no more reviews today; resumes tomorrow`) deferQueuedStatusPrs(cfg, status, queue, i, `daily cap hit (MAX_REVIEWS_PER_DAY=${cfg.maxReviewsPerDay}); resumes tomorrow`) break @@ -1327,45 +1350,65 @@ function main(): void { setStatusPr(cfg, status, pr, 'dry_run', `would review ${lines} diff lines`, lines) continue } + candidates.push({ pr, prior, diff, lines, firstReview }) + } - setStatusPr(cfg, status, pr, 'reviewing', `running codex over ${lines} diff lines`, lines) - setCommitStatus(cfg, commitStatuses, pr, 'pending', `stupify is reviewing ${lines} diff lines`) - const used = reviewPr(cfg, pr, prior.memory, diff, firstReview, prior.openThreadIds, prior.dismissed) - if (used === 'limit') { - log('codex plan is rate-limited — ending this sweep early (the rest would fail the same way); retries next sweep') - setStatusPr(cfg, status, pr, 'failed', 'codex plan is rate-limited; ending sweep early', lines) - setStatusStage(cfg, status, 'blocked', 'codex plan is rate-limited') - setCommitStatus(cfg, commitStatuses, pr, 'error', 'codex plan is rate-limited; retrying later') - deferQueuedStatusPrs(cfg, status, queue, i + 1, 'codex plan is rate-limited; deferred to next sweep') - recordHeadAttempt(failuresPath(cfg), failures, String(pr.number), pr.headRefOid) // throttle this head too so the next sweep doesn't immediately re-hit the wall - break - } - if (used === null) { - recordHeadAttempt(failuresPath(cfg), failures, String(pr.number), pr.headRefOid) // logged, not posted — throttle re-attempt until the window lapses or the head moves - setStatusPr(cfg, status, pr, 'failed', 'review failed; retry will wait for the failure window', lines) - setCommitStatus(cfg, commitStatuses, pr, 'error', 'stupify review failed; retrying later') - continue + // The pool: up to CODEX_JOBS candidates in flight at once. Workers share a cursor; a quota `limit` from any + // worker stops NEW launches (the rest would fail the same way) while in-flight runs drain. All the shared-state + // mutation (counters, status, throttle files) happens between awaits on the one JS thread, so it needs no locks. + let next = 0 + let limitHit = false + const worker = async (): Promise => { + while (!limitHit) { + const c = candidates[next++] + if (c === undefined) return + const { pr, prior, diff, lines } = c + setStatusPr(cfg, status, pr, 'reviewing', `running codex over ${lines} diff lines`, lines) + setCommitStatus(cfg, commitStatuses, pr, 'pending', `stupify is reviewing ${lines} diff lines`) + const used = await reviewPr(cfg, pr, prior.memory, diff, c.firstReview, prior.openThreadIds, prior.dismissed) + if (used === 'limit') { + limitHit = true + log('codex plan is rate-limited — no new reviews this sweep (the rest would fail the same way); retries next sweep') + setStatusPr(cfg, status, pr, 'failed', 'codex plan is rate-limited; ending sweep early', lines) + setStatusStage(cfg, status, 'blocked', 'codex plan is rate-limited') + setCommitStatus(cfg, commitStatuses, pr, 'error', 'codex plan is rate-limited; retrying later') + recordHeadAttempt(failuresPath(cfg), failures, String(pr.number), pr.headRefOid) // throttle this head too so the next sweep doesn't immediately re-hit the wall + continue + } + if (used === null) { + recordHeadAttempt(failuresPath(cfg), failures, String(pr.number), pr.headRefOid) // logged, not posted — throttle re-attempt until the window lapses or the head moves + setStatusPr(cfg, status, pr, 'failed', 'review failed; retry will wait for the failure window', lines) + setCommitStatus(cfg, commitStatuses, pr, 'error', 'stupify review failed; retrying later') + continue + } + // codex ran and reached a verdict (findings posted, or a no-op). Record this head so the next sweep doesn't + // re-run codex on it — without this a SUPPRESSED no-op (no thread marker) would re-run every minute and drain + // the plan. Count the run toward the daily spend ceiling either way: a no-op still spent the tokens. + recordReviewedHead(reviewedPath(cfg), reviewedLocal, String(pr.number), pr.headRefOid) + bumpDailyCounter(dailyPath(cfg), daily) + if (typeof used === 'number') { + reviewed += 1 + tokens += used + setStatusPr(cfg, status, pr, 'posted', `posted review (${used} tokens)`, lines) + } else if (used === 'open') { + setStatusPr(cfg, status, pr, 'skipped', 'prior findings still open; no new review posted', lines) + } else if (used === 'fixed') { + setStatusPr(cfg, status, pr, 'clean', 'prior findings resolved', lines) + } else { + setStatusPr(cfg, status, pr, 'clean', 'no new review needed', lines) + } + const finalStatus = commitStatusForSweepResult(used) + setCommitStatus(cfg, commitStatuses, pr, finalStatus.state, finalStatus.description) + status.totals.reviewed = reviewed + status.totals.tokens = tokens } - // codex ran and reached a verdict (findings posted, or a no-op). Record this head so the next sweep doesn't - // re-run codex on it — without this a SUPPRESSED no-op (no thread marker) would re-run every minute and drain - // the plan. Count the run toward the daily spend ceiling either way: a no-op still spent the tokens. - recordReviewedHead(reviewedPath(cfg), reviewedLocal, String(pr.number), pr.headRefOid) - bumpDailyCounter(dailyPath(cfg), daily) - if (typeof used === 'number') { - reviewed += 1 - tokens += used - setStatusPr(cfg, status, pr, 'posted', `posted review (${used} tokens)`, lines) - } else if (used === 'open') { - setStatusPr(cfg, status, pr, 'skipped', 'prior findings still open; no new review posted', lines) - } else if (used === 'fixed') { - setStatusPr(cfg, status, pr, 'clean', 'prior findings resolved', lines) - } else { - setStatusPr(cfg, status, pr, 'clean', 'no new review needed', lines) + } + await Promise.all(Array.from({ length: Math.min(cfg.codexJobs, candidates.length) }, () => worker())) + if (limitHit) { + for (const c of candidates.slice(next)) { + skipStatusPr(cfg, status, c.pr, 'deferred', 'codex plan is rate-limited; deferred to next sweep') + setCommitStatus(cfg, commitStatuses, c.pr, 'error', 'codex plan is rate-limited; retrying later') } - const finalStatus = commitStatusForSweepResult(used) - setCommitStatus(cfg, commitStatuses, pr, finalStatus.state, finalStatus.description) - status.totals.reviewed = reviewed - status.totals.tokens = tokens } log(`sweep done — scope=${cfg.scope} reviewed=${reviewed} tokens~${tokens}`) @@ -1380,4 +1423,4 @@ function main(): void { writeStatus(cfg, status) } -if (import.meta.main) main() // run only when invoked directly (cron / `stupify run`); stays importable for tests +if (import.meta.main) await main() // run only when invoked directly (cron / `stupify run`); stays importable for tests