diff --git a/internal/crq/config.go b/internal/crq/config.go index 8c237fe..7bb4b62 100644 --- a/internal/crq/config.go +++ b/internal/crq/config.go @@ -69,6 +69,10 @@ type Config struct { // CodeRabbit review following its comment shells) is caught by crq instead // of by a human re-checking the PR. 0 disables. SettleWindow time.Duration + // RateLimitCodexDegrade degrades an account-blocked round to Codex-only + // (return Codex findings promptly, keep CodeRabbit queued for the window) + // instead of waiting the block out. CRQ_RL_CODEX_DEGRADE, default on. + RateLimitCodexDegrade bool } func LoadConfig() (Config, error) { @@ -135,6 +139,8 @@ func LoadConfig() (Config, error) { DryRun: env["CRQ_DRY_RUN"] == "1", FeedbackWaitTimeout: durationEnv(env, "CRQ_FEEDBACK_WAIT_TIMEOUT", 20*time.Minute), SettleWindow: durationEnv(env, "CRQ_SETTLE", 90*time.Second), + + RateLimitCodexDegrade: env["CRQ_RL_CODEX_DEGRADE"] != "0", } if len(cfg.Scope) == 0 && cfg.GateRepo != "" { cfg.Scope = []string{ownerOf(cfg.GateRepo)} diff --git a/internal/crq/feedback.go b/internal/crq/feedback.go index 66d06a2..bad4608 100644 --- a/internal/crq/feedback.go +++ b/internal/crq/feedback.go @@ -26,6 +26,12 @@ type FeedbackReport struct { ReviewedBy map[string]bool `json:"reviewed_by"` Findings []dialect.Finding `json:"findings"` CheckedAt time.Time `json:"checked_at"` + // CodeRabbitDeferred marks a round degraded to Codex-only while the + // CodeRabbit account is rate-limited: Codex feedback is authoritative for + // this round, the CodeRabbit review stays queued and fires after + // DeferredUntil. Converged stays false until it does. + CodeRabbitDeferred bool `json:"coderabbit_deferred,omitempty"` + DeferredUntil *time.Time `json:"coderabbit_deferred_until,omitempty"` } func (s *Service) Feedback(ctx context.Context, repo string, pr int) (FeedbackReport, error) { @@ -238,9 +244,29 @@ func (s *Service) Feedback(ctx context.Context, repo string, pr int) (FeedbackRe return report.Findings[i].Line < report.Findings[j].Line }) report.Converged = engine.Converged(report.Findings, completion) - if report.Converged { + // Degrade detection: a live rate-limit window plus observed Codex + // responsiveness means this round runs Codex-only for now. Converged is + // structurally false here (CodeRabbit has no review evidence), so a + // deferred round can never masquerade as converged. Only the ACCOUNT-wide + // quota block qualifies — a round's own awaiting_retry cooldown also + // covers non-quota retries (post failures, timeouts) that must keep their + // normal retry handling. + if s.cfg.RateLimitCodexDegrade && !report.Converged && st.Account.BlockedUntil != nil { + until := st.Account.BlockedUntil.UTC() + if until.After(now) && engine.CodexOnlyEligible(completionRound, obs.eng, &until, now) { + report.CodeRabbitDeferred = true + report.DeferredUntil = &until + } + } + switch { + case report.Converged: report.Status = "converged" - } else if len(report.Findings) == 0 { + case report.CodeRabbitDeferred && len(report.Findings) == 0 && + engine.DoneExceptWithEvidence(report.ReviewedBy, s.cfg.Bot, dialect.CodexBotLogin): + report.Status = "deferred" + report.Reason = "codex reviewed clean; coderabbit review deferred until " + + report.DeferredUntil.UTC().Format(time.RFC3339) + " (account rate-limited)" + case len(report.Findings) == 0: report.Status = "waiting" } return report, nil @@ -325,7 +351,7 @@ func (s *Service) Loop(ctx context.Context, repo string, pr int) (FeedbackReport return FeedbackReport{}, 1, err } var lastLog time.Time - var convergedAt time.Time + var settledAt time.Time // Pump keeps the queue moving while we wait, but once a minute is plenty (the // autoreview daemon pumps too); pumping on every tick just burns REST quota. var lastPump time.Time @@ -362,6 +388,14 @@ func (s *Service) Loop(ctx context.Context, repo string, pr int) (FeedbackReport if allReviewed(report.ReviewedBy) { report.Reason = "all required reviewers finished; address findings, push once, and resolve threads" s.completeWaitRound(ctx, repo, pr, head) + } else if report.CodeRabbitDeferred && engine.DoneExceptWithEvidence(report.ReviewedBy, s.cfg.Bot, dialect.CodexBotLogin) { + // Degraded round: every required bot except the rate-limited + // CodeRabbit has finished. These findings are this round's work — + // fixing and pushing is exactly right; the CodeRabbit review stays + // queued and fires against the newest head once the window opens. + // With ANOTHER required bot still pending, the hold-head branch + // below applies instead: pushing would restart its checks. + report.Reason = "codex findings during a coderabbit rate-limit window; fix, push, and loop again — the coderabbit review stays queued and fires when the window opens" } else { // A required reviewer is still pending (e.g. Codex posted a finding // before CodeRabbit reviewed). Return the findings to work on, but leave @@ -372,21 +406,25 @@ func (s *Service) Loop(ctx context.Context, repo string, pr int) (FeedbackReport } return report, 10, nil } - if report.Converged { + if report.Converged || report.Status == "deferred" { // Don't trust the first converged observation: bots deliver in waves // (Codex auto-reviews a pushed head minutes later; CodeRabbit's real // review can trail its comment shells). Hold the verdict for the settle // window and only exit 0 if nothing new lands; any finding or pending - // reviewer resets the normal flow above. - if convergedAt.IsZero() { - convergedAt = s.clock() + // reviewer resets the normal flow above. A deferred (Codex-only) clean + // verdict settles the same way but must NOT complete the round — the + // queued CodeRabbit review is still owed for this head. + if settledAt.IsZero() { + settledAt = s.clock() } - if s.cfg.SettleWindow <= 0 || s.clock().Sub(convergedAt) >= s.cfg.SettleWindow { - s.completeWaitRound(ctx, repo, pr, head) + if s.cfg.SettleWindow <= 0 || s.clock().Sub(settledAt) >= s.cfg.SettleWindow { + if report.Converged { + s.completeWaitRound(ctx, repo, pr, head) + } return report, 0, nil } } else { - convergedAt = time.Time{} + settledAt = time.Time{} } // Keep the queue moving (re-fire once an account-block window clears) and pick up // the Blocked state it leaves behind. Pumping every poll tick is redundant — @@ -411,7 +449,13 @@ func (s *Service) Loop(ctx context.Context, repo string, pr int) (FeedbackReport blockedUntil = &until } } - if blockedUntil != nil { + // A degraded round waits for Codex, not for the window: keep the normal + // poll cadence against the un-extended deadline. (The rate-limit reply + // usually lands before Codex answers, so an iteration or two may extend + // the deadline first — harmless: once Codex evidence arrives the loop + // exits promptly, and if Codex never answers the round degrades + // gracefully back to riding out the window.) + if blockedUntil != nil && !report.CodeRabbitDeferred { extended := extendDeadlineForBlock(deadline, blockedUntil, now, s.cfg.FeedbackWaitTimeout) if extended.After(deadline) { deadline = extended @@ -419,8 +463,12 @@ func (s *Service) Loop(ctx context.Context, repo string, pr int) (FeedbackReport } poll = blockedPollInterval(*blockedUntil, now, s.cfg.PollInterval) } - if now.After(deadline) && convergedAt.IsZero() { - s.completeWaitRound(ctx, repo, pr, head) + if now.After(deadline) && settledAt.IsZero() { + // A degraded round must not be completed on timeout: marking the head + // reviewed would silently cancel the still-owed CodeRabbit review. + if !report.CodeRabbitDeferred { + s.completeWaitRound(ctx, repo, pr, head) + } if len(report.Findings) > 0 { report.Status = "feedback" report.Reason = "review wait timed out; actionable findings must be addressed before retrying" @@ -431,7 +479,9 @@ func (s *Service) Loop(ctx context.Context, repo string, pr int) (FeedbackReport } if s.log != nil && time.Since(lastLog) >= 30*time.Second { activeElapsed := feedbackWaitElapsed(deadline, s.cfg.FeedbackWaitTimeout, now) - if blockedUntil != nil { + if blockedUntil != nil && report.CodeRabbitDeferred { + s.log.Printf("%s#%d degraded to codex-only — coderabbit rate-limited until %s; waiting for codex on %s (%s / %s)", repo, pr, blockedUntil.UTC().Format(time.RFC3339), report.Head, activeElapsed.Round(time.Second), s.cfg.FeedbackWaitTimeout) + } else if blockedUntil != nil { s.log.Printf("%s#%d queued — account blocked until %s; waiting, not counting it against the %s review wait (%s active)", repo, pr, blockedUntil.UTC().Format(time.RFC3339), s.cfg.FeedbackWaitTimeout, activeElapsed.Round(time.Second)) } else { s.log.Printf("%s#%d waiting for review feedback on %s — reviewed %s (%s / %s)", repo, pr, report.Head, reviewedSummary(report.ReviewedBy), activeElapsed.Round(time.Second), s.cfg.FeedbackWaitTimeout) diff --git a/internal/crq/service.go b/internal/crq/service.go index 7e425f3..270a3e6 100644 --- a/internal/crq/service.go +++ b/internal/crq/service.go @@ -261,7 +261,39 @@ func (s *Service) Pump(ctx context.Context) (PumpResult, error) { return PumpResult{}, err } decision := engine.DecideFire(s.global(st, now), *next, obs.eng, now, s.policy()) - return s.applyFire(ctx, *next, obs.eng, decision, now) + result, err := s.applyFire(ctx, *next, obs.eng, decision, now) + if err != nil { + return result, err + } + // A blocked or slot-busy front of the queue whose Codex command is + // already posted must not starve later PRs of THEIR early Codex round — + // scan a bounded number of following queued rounds for a postable + // Codex defer (each costs one observation; ETag caching keeps it cheap). + if s.cfg.RateLimitCodexDegrade && decision.Verdict == engine.FireNo && + (strings.Contains(decision.Reason, "account blocked") || + strings.Contains(decision.Reason, "fire slot busy")) { + scanned := 0 + for _, r := range st.QueuedRounds(now) { + if r.Repo == next.Repo && r.PR == next.PR { + continue + } + if r.CodexCommandID != 0 || scanned >= 3 { + continue + } + scanned++ + round := r + robs, oerr := s.observe(ctx, round.Repo, round.PR, &round, now) + if oerr != nil { + continue + } + d := engine.DecideFire(s.global(st, now), round, robs.eng, now, s.policy()) + if d.Verdict != engine.FireCodexDeferred { + continue + } + return s.applyFire(ctx, round, robs.eng, d, now) + } + } + return result, nil } func (s *Service) global(st State, now time.Time) engine.Global { @@ -463,6 +495,8 @@ func (s *Service) applyFire(ctx context.Context, round Round, obs engine.Observa return s.dedupeRound(ctx, round, now, d.Reason) case engine.FireCodexOnly: return s.fireCodexOnly(ctx, round, d.Reason, now) + case engine.FireCodexDeferred: + return s.fireCodexDeferred(ctx, round, d.AdoptCommandID, d.AdoptAt, d.Reason, now) case engine.FireCoReviewWait: return s.fireCoReviewWait(ctx, round, obs, d.Reason, now) case engine.FireSupersede: @@ -622,8 +656,16 @@ func (s *Service) fireRound(ctx context.Context, round Round, obs engine.Observa // Not posting because a live `@codex review` command already answers // this head — record its id now. The self-heal scan anchors on FiredAt // and would miss a command posted before the adopted CodeRabbit command, - // posting a duplicate; recording it here keeps the round "asked". + // posting a duplicate; recording it here keeps the round "asked". Its + // timestamp anchors the codex cutoff too, or a SHA-less answer that + // landed before this adopted fire would never bind to the round. r.CodexCommandID = id + for _, c := range obs.CodexCommands { + if c.ID == id && !c.CreatedAt.IsZero() { + at := c.CreatedAt.UTC() + r.CodexCommandedAt = &at + } + } } st.PutRound(*r) recorded = true @@ -700,7 +742,7 @@ func (s *Service) fireRound(ctx context.Context, round Round, obs engine.Observa // write. A failed post returns 0 (logged) and the self-heal path retries. var codexID int64 if postCodex { - codexID = s.postCodexReviewComment(ctx, round) + codexID, _ = s.postCodexReviewComment(ctx, round) } updated, err := s.recordFire(ctx, round, token, comment.ID, codexID, firedAt, now) if err != nil { @@ -846,6 +888,12 @@ func (s *Service) fireCoReviewWait(ctx context.Context, round Round, obs engine. } if r.CodexCommandID == 0 && codexID != 0 { r.CodexCommandID = codexID + for _, c := range obs.CodexCommands { + if c.ID == codexID && !c.CreatedAt.IsZero() { + at := c.CreatedAt.UTC() + r.CodexCommandedAt = &at + } + } } st.PutRound(*r) changed = true @@ -910,18 +958,25 @@ func (s *Service) recordFire(ctx context.Context, round Round, token string, com // id, or 0 on failure. A failed post is non-fatal: it logs and leaves // CodexCommandID unset so a later pump's self-heal retries. The fresh-fire path // folds the returned id into recordFire's write. -func (s *Service) postCodexReviewComment(ctx context.Context, round Round) int64 { +func (s *Service) postCodexReviewComment(ctx context.Context, round Round) (int64, time.Time) { comment, err := s.gh.PostIssueComment(ctx, round.Repo, round.PR, s.cfg.CodexCommand) if err != nil { if s.log != nil { s.log.Printf("warning: Codex review command post failed for %s@%s: %v (will retry on a later pump)", QueueKey(round.Repo, round.PR), round.Head, err) } - return 0 + return 0, time.Time{} } if s.log != nil { s.log.Printf("fire %s@%s (posted %s)", QueueKey(round.Repo, round.PR), round.Head, strings.TrimSpace(s.cfg.CodexCommand)) } - return comment.ID + // The command's GitHub timestamp is the evidence anchor: a fast Codex + // reply can otherwise land before a local post-return clock reading and + // fall outside the stored cutoff forever (with no repost possible). + at := comment.CreatedAt.UTC() + if at.IsZero() { + at = s.clock().UTC() + } + return comment.ID, at } // fireCodexReview posts the Codex review command for an already-fired round and @@ -929,7 +984,7 @@ func (s *Service) postCodexReviewComment(ctx context.Context, round Round) int64 // retry (the fresh-post path records the id inside recordFire instead). The CAS // guard (same head, CodexCommandID still unset) makes a concurrent post benign. func (s *Service) fireCodexReview(ctx context.Context, round Round) { - codexID := s.postCodexReviewComment(ctx, round) + codexID, codexAt := s.postCodexReviewComment(ctx, round) if codexID == 0 { // Failed post: KEEP the claim — its TTL is the retry backoff. Clearing it // here would let the very next pump repost, bypassing codexClaimTTL. @@ -945,6 +1000,7 @@ func (s *Service) fireCodexReview(ctx context.Context, round Round) { r.CodexClaimedAt = nil if r.CodexCommandID == 0 { r.CodexCommandID = codexID + r.CodexCommandedAt = &codexAt } st.PutRound(*r) return nil @@ -958,6 +1014,88 @@ func (s *Service) fireCodexReview(ctx context.Context, round Round) { s.sync(ctx, updated) } +// fireCodexDeferred posts ONLY the Codex review command for a round the +// CodeRabbit account block is holding back. Unlike fireCodexOnly (CodeRabbit +// already reviewed the head) the round stays queued/awaiting-retry: the +// CodeRabbit review is still owed and fires normally the moment the window +// opens, at which point PostCodex stays false because CodexCommandID is set. +// The claim-then-post shape mirrors selfHealCodex — this path is not +// serialized by the fire slot either. +func (s *Service) fireCodexDeferred(ctx context.Context, round Round, adoptID int64, adoptAt time.Time, reason string, now time.Time) (PumpResult, error) { + result := PumpResult{Action: "codex_fired", Repo: round.Repo, PR: round.PR, Head: round.Head, Reason: reason} + if s.cfg.DryRun { + return result, nil + } + if adoptID != 0 { + adopted := false + updated, err := s.store.Update(ctx, func(st *State) error { + r := st.Round(round.Repo, round.PR) + if !sameRound(r, round) || r.CodexCommandID != 0 || + (r.Phase != PhaseQueued && r.Phase != PhaseAwaitingRetry) { + return ErrNoChange + } + // A concurrent worker may already be posting the command represented by + // this observation. Let its fresh claim finish instead of adopting and + // racing its state write; a stale claim no longer blocks recovery. + if r.CodexClaimedAt != nil && now.Sub(r.CodexClaimedAt.UTC()) < codexClaimTTL { + return ErrNoChange + } + at := adoptAt.UTC() + if at.IsZero() { + at = now.UTC() + } + r.CodexCommandID = adoptID + r.CodexCommandedAt = &at + r.CodexClaimedAt = nil + r.Note = "existing codex review command adopted; coderabbit deferred (account rate-limited)" + st.PutRound(*r) + adopted = true + return nil + }) + if err != nil && !errors.Is(err, ErrNoChange) { + return result, err + } + if !adopted { + result.Action = "deduped" + result.Reason = "codex command already adopted or posted" + return result, nil + } + s.sync(ctx, updated) + result.Action = "codex_adopted" + return result, nil + } + claimed := false + updated, err := s.store.Update(ctx, func(st *State) error { + r := st.Round(round.Repo, round.PR) + if !sameRound(r, round) || r.CodexCommandID != 0 { + return ErrNoChange + } + if r.Phase != PhaseQueued && r.Phase != PhaseAwaitingRetry { + return ErrNoChange + } + if r.CodexClaimedAt != nil && now.Sub(r.CodexClaimedAt.UTC()) < codexClaimTTL { + return ErrNoChange + } + t := now.UTC() + r.CodexClaimedAt = &t + r.Note = "codex review requested; coderabbit deferred (account rate-limited)" + st.PutRound(*r) + claimed = true + return nil + }) + if err != nil && !errors.Is(err, ErrNoChange) { + return result, err + } + if !claimed { + result.Action = "deduped" + result.Reason = "codex command already claimed or posted" + return result, nil + } + s.sync(ctx, updated) + s.fireCodexReview(ctx, round) + return result, nil +} + // selfHealCodex re-posts the Codex review command for a fired/reviewing round // whose initial Codex post failed (CodexCommandID still 0). It runs on the // daemon's progress/sweep paths; idempotence comes from the observation — Codex @@ -1441,6 +1579,15 @@ func (s *Service) Wait(ctx context.Context, repo string, pr int) (PumpResult, in } return PumpResult{Action: "deduped", Repo: repo, PR: pr, Head: report.Head, Reason: "feedback already available"}, 3, nil } + // A clean Codex answer on a degraded (rate-limited) round is the + // round's verdict for now — hand off to the feedback poll loop + // instead of spinning here until the CodeRabbit window opens. + if report.Status == "deferred" { + if s.log != nil { + s.log.Printf("%s#%d codex answered on %s; coderabbit deferred — leaving review slot wait", repo, pr, report.Head) + } + return PumpResult{Action: "deduped", Repo: repo, PR: pr, Head: report.Head, Reason: "codex answered; coderabbit deferred"}, 3, nil + } } result, err := s.Pump(ctx) if err != nil { diff --git a/internal/crq/service_test.go b/internal/crq/service_test.go index 40a6784..0371553 100644 --- a/internal/crq/service_test.go +++ b/internal/crq/service_test.go @@ -1894,3 +1894,338 @@ func TestRefreshQuotaPreservesBlockOnInconclusiveProbe(t *testing.T) { t.Fatalf("an inconclusive probe must preserve the live block %v, got %v", block, updated.Account.BlockedUntil) } } + +// TestLoopDegradesToCodexOnlyOnRateLimit: a rate-limited round with Codex +// activity must return Codex findings promptly — marked deferred — instead of +// waiting out the CodeRabbit window. +func TestLoopDegradesToCodexOnlyOnRateLimit(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RateLimitCodexDegrade = true + cfg.FeedbackBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + gh := newFakeGitHub() + head := "a0646f010" + pull := ghapi.Pull{State: "open"} + pull.Head.SHA = head + "abcdef0" + gh.pulls[fakeKey("o/carrier", 90)] = pull + store := NewMemoryStore(cfg) + svc := NewService(cfg, gh, store, nil) + + if _, err := svc.Enqueue(ctx, "o/carrier", 90); err != nil { + t.Fatal(err) + } + if res, err := svc.Pump(ctx); err != nil || res.Action != "fired" { + t.Fatalf("expected fire, got %#v err=%v", res, err) + } + answer := time.Now().UTC().Add(time.Second) + rl := ghapi.IssueComment{ID: 501, Body: "\n> ## Review limit reached\n> **Next review available in:** **40 minutes**", CreatedAt: answer, UpdatedAt: answer} + rl.User.Login = "coderabbitai[bot]" + finding := ghapi.IssueComment{ID: 502, Body: "Actionable Codex finding on the current head", CreatedAt: answer.Add(time.Second), UpdatedAt: answer.Add(time.Second)} + finding.User.Login = "chatgpt-codex-connector[bot]" + gh.comments[fakeKey("o/carrier", 90)] = []ghapi.IssueComment{rl, finding} + if res, err := svc.Pump(ctx); err != nil || res.Action != "requeued" { + t.Fatalf("expected rate-limited requeue, got %#v err=%v", res, err) + } + + begin := time.Now() + report, code, err := svc.Loop(ctx, "o/carrier", 90) + if err != nil { + t.Fatal(err) + } + if code != 10 || len(report.Findings) == 0 { + t.Fatalf("degraded loop must return the codex findings, code=%d report=%#v", code, report) + } + if !report.CodeRabbitDeferred || report.DeferredUntil == nil { + t.Fatalf("the report must mark coderabbit deferred with a window, got %#v", report) + } + if report.ReviewedBy["coderabbitai[bot]"] { + t.Fatalf("coderabbit must still read as unreviewed: %#v", report.ReviewedBy) + } + if report.Converged { + t.Fatal("a deferred round must never read as converged") + } + if elapsed := time.Since(begin); elapsed > 5*time.Second { + t.Fatalf("degraded loop must not wait out the rate-limit window, took %s", elapsed) + } +} + +// TestFeedbackDefersCleanAutoCodexWithDefaultRequiredBots covers the default +// feedback-only Codex setup: its SHA-bound clean verdict is sufficient evidence +// even though Completion.ReviewedBy contains only the pending CodeRabbit key. +func TestFeedbackDefersCleanAutoCodexWithDefaultRequiredBots(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RateLimitCodexDegrade = true + cfg.FeedbackBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + gh := newFakeGitHub() + pull := ghapi.Pull{State: "open"} + pull.Head.SHA = "abcdef1234567890" + gh.pulls[fakeKey("o/carrier", 93)] = pull + cleanAt := time.Now().UTC().Add(-time.Minute) + clean := ghapi.IssueComment{ID: 701, + Body: "Codex Review: Didn't find any major issues. :tada:\n\n**Reviewed commit:** `abcdef1234`", + CreatedAt: cleanAt, UpdatedAt: cleanAt} + clean.User.Login = "chatgpt-codex-connector[bot]" + gh.comments[fakeKey("o/carrier", 93)] = []ghapi.IssueComment{clean} + store := NewMemoryStore(cfg) + seedRound(t, store, cfg, "o/carrier", 93, "abcdef123", PhaseQueued, cleanAt.Add(-time.Minute), 0) + blockedUntil := time.Now().UTC().Add(30 * time.Minute) + if _, err := store.Update(ctx, func(st *State) error { + st.Account.BlockedUntil = &blockedUntil + return nil + }); err != nil { + t.Fatal(err) + } + + svc := NewService(cfg, gh, store, nil) + report, err := svc.Feedback(ctx, "o/carrier", 93) + if err != nil { + t.Fatal(err) + } + if report.Status != "deferred" || !report.CodeRabbitDeferred || report.Converged { + t.Fatalf("clean feedback-only codex must produce a non-converged deferred report, got %#v", report) + } + if len(report.ReviewedBy) != 1 || report.ReviewedBy["coderabbitai[bot]"] { + t.Fatalf("default required-bot state must remain coderabbit-only and pending, got %#v", report.ReviewedBy) + } +} + +// TestLoopDeferredCodexCleanExitsZeroNotConverged: a clean Codex verdict on a +// degraded round exits 0 as "deferred" without completing the round — the +// CodeRabbit review stays owed. +func TestLoopDeferredCodexCleanExitsZeroNotConverged(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RateLimitCodexDegrade = true + cfg.RequiredBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + cfg.FeedbackBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + cfg.CodexCommand = "@codex review" + cfg.SettleWindow = 0 + gh := newFakeGitHub() + pull := ghapi.Pull{State: "open"} + pull.Head.SHA = "abcdef1234567890" + gh.pulls[fakeKey("o/carrier", 91)] = pull + store := NewMemoryStore(cfg) + started := time.Now().UTC().Add(-time.Minute) + seedRound(t, store, cfg, "o/carrier", 91, "abcdef123", PhaseReviewing, started, 1001) + blockedUntil := time.Now().UTC().Add(30 * time.Minute) + if _, err := store.Update(ctx, func(st *State) error { + st.Account.BlockedUntil = &blockedUntil + return nil + }); err != nil { + t.Fatal(err) + } + clean := ghapi.IssueComment{ID: 700, Body: "## Codex Review\n\nDidn't find any major issues. Keep them coming!", CreatedAt: started.Add(30 * time.Second), UpdatedAt: started.Add(30 * time.Second)} + clean.User.Login = "chatgpt-codex-connector[bot]" + gh.comments[fakeKey("o/carrier", 91)] = []ghapi.IssueComment{clean} + + svc := NewService(cfg, gh, store, nil) + report, code, err := svc.Loop(ctx, "o/carrier", 91) + if err != nil { + t.Fatal(err) + } + if code != 0 || report.Status != "deferred" { + t.Fatalf("clean codex on a degraded round must exit 0 as deferred, code=%d report=%#v", code, report) + } + if report.Converged { + t.Fatal("deferred must not be converged") + } + st, _, _ := store.Load(ctx) + if r := st.Round("o/carrier", 91); r == nil || r.Phase == PhaseCompleted { + t.Fatalf("a deferred round must not be completed, got %#v", r) + } + if !st.ContainsActive("o/carrier", 91) { + t.Fatal("the coderabbit review must stay owed (round active)") + } +} + +// TestPumpPostsCodexDeferredDuringBlockThenFiresCodeRabbit: during a block the +// pump posts only the Codex command and keeps the round queued; once the +// window passes it fires the CodeRabbit review without re-posting Codex. +func TestPumpPostsCodexDeferredDuringBlockThenFiresCodeRabbit(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RateLimitCodexDegrade = true + cfg.RequiredBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + cfg.CodexCommand = "@codex review" + gh := newFakeGitHub() + pull := ghapi.Pull{State: "open"} + pull.Head.SHA = "abcdef1234567890" + gh.pulls[fakeKey("o/carrier", 92)] = pull + store := NewMemoryStore(cfg) + svc := NewService(cfg, gh, store, nil) + now := time.Now().UTC() + svc.now = func() time.Time { return now } + + if _, err := svc.Enqueue(ctx, "o/carrier", 92); err != nil { + t.Fatal(err) + } + blockedUntil := now.Add(30 * time.Minute) + if _, err := store.Update(ctx, func(st *State) error { + st.Account.BlockedUntil = &blockedUntil + return nil + }); err != nil { + t.Fatal(err) + } + res, err := svc.Pump(ctx) + if err != nil { + t.Fatal(err) + } + if res.Action != "codex_fired" { + t.Fatalf("expected the codex-deferred fire, got %#v", res) + } + st, _, _ := store.Load(ctx) + r := st.Round("o/carrier", 92) + if r == nil || r.Phase != PhaseQueued || r.CodexCommandID == 0 { + t.Fatalf("the round must stay queued with the codex command recorded, got %#v", r) + } + if _, err := svc.Pump(ctx); err != nil { + t.Fatal(err) + } + countPosts := func(suffix string) int { + n := 0 + for _, p := range gh.posted { + if strings.HasSuffix(p, suffix) { + n++ + } + } + return n + } + if countPosts(":@codex review") != 1 { + t.Fatalf("exactly one codex command must be posted during the block, got %v", gh.posted) + } + if countPosts(":@coderabbitai review") != 0 { + t.Fatalf("no coderabbit review may fire during the block, got %v", gh.posted) + } + + // Window opens: the queued round fires CodeRabbit, without re-posting Codex. + now = blockedUntil.Add(time.Minute) + res, err = svc.Pump(ctx) + if err != nil { + t.Fatal(err) + } + if res.Action != "fired" { + t.Fatalf("expected the coderabbit fire after the window, got %#v", res) + } + if countPosts(":@coderabbitai review") != 1 || countPosts(":@codex review") != 1 { + t.Fatalf("after the window: one coderabbit fire, still one codex command, got %v", gh.posted) + } +} + +func TestPumpAdoptsExistingCodexCommandDuringBlock(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RateLimitCodexDegrade = true + cfg.RequiredBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + cfg.CodexCommand = "@codex review" + gh := newFakeGitHub() + headTime := time.Now().UTC().Add(-2 * time.Minute) + pull := ghapi.Pull{State: "open"} + pull.Head.SHA = "abcdef1234567890" + gh.pulls[fakeKey("o/carrier", 94)] = pull + commit := ghapi.Commit{SHA: pull.Head.SHA} + commit.Committer.Date = headTime + gh.commits[pull.Head.SHA] = commit + commandAt := headTime.Add(time.Minute) + command := ghapi.IssueComment{ID: 702, Body: cfg.CodexCommand, CreatedAt: commandAt, UpdatedAt: commandAt} + command.User.Login = "kristofferR" + gh.comments[fakeKey("o/carrier", 94)] = []ghapi.IssueComment{command} + gh.graphQL = noForcePush + store := NewMemoryStore(cfg) + svc := NewService(cfg, gh, store, nil) + if _, err := svc.Enqueue(ctx, "o/carrier", 94); err != nil { + t.Fatal(err) + } + blockedUntil := time.Now().UTC().Add(30 * time.Minute) + if _, err := store.Update(ctx, func(st *State) error { + st.Account.BlockedUntil = &blockedUntil + return nil + }); err != nil { + t.Fatal(err) + } + + res, err := svc.Pump(ctx) + if err != nil { + t.Fatal(err) + } + if res.Action != "codex_adopted" { + t.Fatalf("expected the existing codex command to be adopted, got %#v", res) + } + if len(gh.posted) != 0 { + t.Fatalf("adopting the codex command must not post another command, got %v", gh.posted) + } + st, _, err := store.Load(ctx) + if err != nil { + t.Fatal(err) + } + r := st.Round("o/carrier", 94) + if r == nil || r.Phase != PhaseQueued || r.CodexCommandID != command.ID || + r.CodexCommandedAt == nil || !r.CodexCommandedAt.Equal(commandAt) { + t.Fatalf("the queued round must persist the adopted codex anchor, got %#v", r) + } +} + +func TestFireCodexDeferredAdoptionHonorsDryRunAndActiveClaim(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RateLimitCodexDegrade = true + cfg.RequiredBots = []string{"coderabbitai[bot]", "chatgpt-codex-connector[bot]"} + cfg.CodexCommand = "@codex review" + gh := newFakeGitHub() + store := NewMemoryStore(cfg) + now := time.Now().UTC() + seedRound(t, store, cfg, "o/carrier", 95, "abcdef123", PhaseQueued, now.Add(-time.Minute), 0) + svc := NewService(cfg, gh, store, nil) + st, _, err := store.Load(ctx) + if err != nil { + t.Fatal(err) + } + round := *st.Round("o/carrier", 95) + + svc.cfg.DryRun = true + res, err := svc.fireCodexDeferred(ctx, round, 703, now.Add(-time.Minute), "dry run adopt", now) + if err != nil { + t.Fatal(err) + } + if res.Action != "codex_fired" || len(gh.posted) != 0 { + t.Fatalf("dry run must report the existing simulated action without posting, got %#v posts=%v", res, gh.posted) + } + st, _, _ = store.Load(ctx) + if got := st.Round("o/carrier", 95); got == nil || got.CodexCommandID != 0 { + t.Fatalf("dry run must not persist the adopted command, got %#v", got) + } + + svc.cfg.DryRun = false + if _, err := store.Update(ctx, func(st *State) error { + r := st.Round("o/carrier", 95) + r.CodexClaimedAt = &now + st.PutRound(*r) + return nil + }); err != nil { + t.Fatal(err) + } + res, err = svc.fireCodexDeferred(ctx, round, 703, now.Add(-time.Minute), "claimed adopt", now) + if err != nil { + t.Fatal(err) + } + if res.Action != "deduped" { + t.Fatalf("a fresh in-flight claim must defer adoption, got %#v", res) + } + st, _, _ = store.Load(ctx) + if got := st.Round("o/carrier", 95); got == nil || got.CodexCommandID != 0 || got.CodexClaimedAt == nil { + t.Fatalf("the active claim must remain untouched, got %#v", got) + } + + staleNow := now.Add(codexClaimTTL + time.Second) + res, err = svc.fireCodexDeferred(ctx, round, 703, now.Add(-time.Minute), "stale claim adopt", staleNow) + if err != nil { + t.Fatal(err) + } + if res.Action != "codex_adopted" { + t.Fatalf("an expired claim must allow adoption recovery, got %#v", res) + } + st, _, _ = store.Load(ctx) + if got := st.Round("o/carrier", 95); got == nil || got.CodexCommandID != 703 || got.CodexClaimedAt != nil { + t.Fatalf("the recovered adoption must replace the stale claim, got %#v", got) + } +} diff --git a/internal/crq/state.go b/internal/crq/state.go index b3364fd..d4e36aa 100644 --- a/internal/crq/state.go +++ b/internal/crq/state.go @@ -81,12 +81,13 @@ func issueBody(st State, cfg Config) (string, error) { // policy assembles the engine Policy from config. func (s *Service) policy() engine.Policy { return engine.Policy{ - Bot: s.cfg.Bot, - RequiredBots: s.cfg.RequiredBots, - CodexCommand: s.cfg.CodexCommand, - MinInterval: s.cfg.MinInterval, - InflightTimeout: s.cfg.InflightTimeout, - RateLimitFallback: s.cfg.RateLimitFallback, + Bot: s.cfg.Bot, + RequiredBots: s.cfg.RequiredBots, + CodexCommand: s.cfg.CodexCommand, + MinInterval: s.cfg.MinInterval, + InflightTimeout: s.cfg.InflightTimeout, + RateLimitFallback: s.cfg.RateLimitFallback, + RateLimitCodexDegrade: s.cfg.RateLimitCodexDegrade, } } diff --git a/internal/engine/codex.go b/internal/engine/codex.go index 35288b6..44f12b1 100644 --- a/internal/engine/codex.go +++ b/internal/engine/codex.go @@ -22,6 +22,21 @@ func roundCutoff(r state.Round) time.Time { return time.Time{} } +// codexCutoff is the evidence floor for Codex specifically: evidence produced +// in response to crq's own Codex command binds from the command time, which +// can precede a deferred CodeRabbit fire (the command posts while the round +// is still queued behind a rate-limit window or busy slot). +func codexCutoff(r state.Round) time.Time { + cut := roundCutoff(r) + if r.CodexCommandedAt != nil { + at := r.CodexCommandedAt.UTC() + if cut.IsZero() || at.Before(cut) { + return at + } + } + return cut +} + // codexReviewedRound reports whether a submitted Codex review binds to this // round: one whose commit prefixes the head, or — SHA-less — one submitted // at/after the fire. @@ -55,15 +70,35 @@ func codexCommentedRound(obs Observation, cutoff time.Time) bool { return false } -// codexReviewedHead reports whether Codex has a submitted review whose commit -// prefixes the observed head — the "Codex already reviewed this head" fire guard. -func codexReviewedHead(obs Observation) bool { +// codexReviewedHeadAt reports the newest Codex verdict explicitly bound to the +// observed head: either a submitted review or a clean summary naming that SHA. +// The timestamp is the evidence floor used to ignore older usage-limit notices. +func codexReviewedHeadAt(obs Observation) (time.Time, bool) { + var latest time.Time + matched := false for _, review := range obs.Reviews { if dialect.IsCodexBot(review.Bot) && obs.Head != "" && review.Commit != "" && strings.HasPrefix(review.Commit, obs.Head) { - return true + matched = true + if review.SubmittedAt.After(latest) { + latest = review.SubmittedAt + } } } - return false + for _, ev := range obs.Events { + if ev.Kind == dialect.EvCodexClean && obs.Head != "" && dialect.SHAPrefixMatch(ev.SHA, obs.Head) { + matched = true + if at := ev.ObservedTime(); at.After(latest) { + latest = at + } + } + } + return latest, matched +} + +// codexReviewedHead is the "Codex already reviewed this head" fire guard. +func codexReviewedHead(obs Observation) bool { + _, matched := codexReviewedHeadAt(obs) + return matched } // CodexActiveThisRound reports whether Codex shows any activity bound to this @@ -71,7 +106,7 @@ func codexReviewedHead(obs Observation) bool { // thumbs-up. observe() stores it on the Observation so the dynamic completion // gate requires Codex when it participates without being configured-required. func CodexActiveThisRound(r state.Round, obs Observation) bool { - cutoff := roundCutoff(r) + cutoff := codexCutoff(r) return codexReviewedRound(r, obs, cutoff) || codexCommentedRound(obs, cutoff) || obs.CodexThumbsUp } @@ -161,6 +196,35 @@ func CodexCommandSince(obs Observation, since time.Time) bool { return false } +// CodexOnlyEligible reports whether an account-blocked round may degrade to a +// Codex-only round: the block is live AND Codex has evidence bound to THIS +// work — a review of the current head, or round-window activity anchored by +// the fire or by crq's own (possibly pre-fire) Codex command — AND no +// usage-limit exhaustion notice inside that same window. Auto-activity on +// older heads, configuration, or a live unanswered command merely predict +// evidence; degradation waits for the evidence itself, since before Codex +// responds there is nothing to return early anyway, and marking a round +// deferred stops the loop from extending its deadline over the block. +func CodexOnlyEligible(r state.Round, obs Observation, blockedUntil *time.Time, now time.Time) bool { + if blockedUntil == nil || !blockedUntil.After(now) { + return false + } + headEvidenceAt, headReviewed := codexReviewedHeadAt(obs) + anchored := r.FiredAt != nil || r.CodexCommandedAt != nil + if !headReviewed && !(anchored && obs.CodexActiveThisRound) { + return false + } + // The usage-limit floor is the evidence window. For an unfired, + // uncommanded round the cutoff is zero — floor it at the head review that + // qualified the round instead, or any old exhaustion notice still on the + // PR would suppress the degrade until the window expires. + floor := codexCutoff(r) + if floor.IsZero() { + floor = headEvidenceAt + } + return !codexUsageLimitedSince(obs, floor) +} + // codexUsageLimitedSince reports whether Codex posted its usage-limit // exhaustion notice at/after since — the round window it can no longer finish. func codexUsageLimitedSince(obs Observation, since time.Time) bool { diff --git a/internal/engine/completion.go b/internal/engine/completion.go index f8cfe64..d95a65c 100644 --- a/internal/engine/completion.go +++ b/internal/engine/completion.go @@ -84,7 +84,10 @@ func Completion(r state.Round, obs Observation, p Policy) CompletionStatus { } continue } - if r.FiredAt != nil && notBefore(ev.ObservedTime(), cutoff) { + // SHA-less summaries bind from the Codex command time when crq posted + // it before the (deferred) CodeRabbit fire — see codexCutoff. + if (r.FiredAt != nil || r.CodexCommandedAt != nil) && + notBefore(ev.ObservedTime(), codexCutoff(r)) { markReviewed(reviewedBy, ev.Bot) } } @@ -277,6 +280,46 @@ func commandReplies(obs Observation, p Policy) []commandReply { return out } +// DoneExcept reports whether every gating bot EXCEPT the named one has review +// evidence — and at least one other bot gates at all. The vacuous case +// matters: with only the excluded bot required, a degraded round must never +// read as done, or a CodeRabbit rate-limit window with no Codex configured +// would let rounds complete with no review evidence whatsoever. +func DoneExcept(reviewedBy map[string]bool, except string) bool { + norm := dialect.NormalizeBotName(except) + others := 0 + for bot, reviewed := range reviewedBy { + if bot == except || dialect.NormalizeBotName(bot) == norm { + continue + } + if !reviewed { + return false + } + others++ + } + return others > 0 +} + +// DoneExceptWithEvidence is DoneExcept after applying independently established +// review evidence for one bot. It preserves every other gating bot, while also +// handling configured login spellings that differ only by the "[bot]" suffix. +func DoneExceptWithEvidence(reviewedBy map[string]bool, except, evidenceBot string) bool { + withEvidence := make(map[string]bool, len(reviewedBy)+1) + evidenceNorm := dialect.NormalizeBotName(evidenceBot) + found := false + for bot, reviewed := range reviewedBy { + if dialect.NormalizeBotName(bot) == evidenceNorm { + reviewed = true + found = true + } + withEvidence[bot] = reviewed + } + if !found { + withEvidence[evidenceBot] = true + } + return DoneExcept(withEvidence, except) +} + // needsBotReview reports whether login gates completion (has a ReviewedBy // key) and its review hasn't been seen yet. func needsBotReview(reviewedBy map[string]bool, login string) bool { diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 287a293..2123522 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -22,6 +22,11 @@ type Policy struct { InflightTimeout time.Duration // fired round with no bot response at all RateLimitFallback time.Duration // block window when "available in" is unparseable RetryBackoff time.Duration // cooldown after a non-rate-limit retry (timeout, failure) + + // RateLimitCodexDegrade lets an account-blocked round degrade to a + // Codex-only round (post the Codex command now, keep CodeRabbit queued + // for the window) instead of waiting the block out. CRQ_RL_CODEX_DEGRADE. + RateLimitCodexDegrade bool } func (p Policy) rateLimitFallback() time.Duration { diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index 98b7ff3..13f2835 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -636,3 +636,211 @@ func TestFindingsForActiveRoundIncludesOnlyCurrentOrNewlyArrivedFeedback(t *test t.Fatalf("unexpected active-round findings: %#v", got) } } + +// TestCodexOnlyEligible covers the degrade predicate: a live block plus +// OBSERVED Codex responsiveness — configuration or a live command alone never +// qualifies, and a usage-limit notice disengages it. +func TestCodexOnlyEligible(t *testing.T) { + r := firedRound(t, "abcdef123") + now := t0.Add(5 * time.Minute) + blocked := now.Add(25 * time.Minute) + expired := now.Add(-time.Minute) + usageLimit := dialect.BotEvent{Kind: dialect.EvCodexUsageLimit, Bot: dialect.CodexBotLogin, CommentID: 700, + CreatedAt: r.FiredAt.Add(30 * time.Second), UpdatedAt: r.FiredAt.Add(30 * time.Second)} + + active := Observation{Head: "abcdef123", Open: true, CodexActiveThisRound: true} + if CodexOnlyEligible(r, active, nil, now) { + t.Fatal("no block must not degrade") + } + if CodexOnlyEligible(r, active, &expired, now) { + t.Fatal("an expired block must not degrade") + } + if !CodexOnlyEligible(r, active, &blocked, now) { + t.Fatal("a live block with round-bound codex activity must degrade") + } + // Auto-activity alone predicts evidence; it does not qualify until Codex + // actually responds to this head/round. + auto := Observation{Head: "abcdef123", Open: true, CodexAutoActive: true} + if CodexOnlyEligible(r, auto, &blocked, now) { + t.Fatal("auto-activity without current evidence must not degrade") + } + configOnly := Observation{Head: "abcdef123", Open: true, CodexCommands: []CommandSeen{{ID: 55, CreatedAt: now}}} + if CodexOnlyEligible(r, configOnly, &blocked, now) { + t.Fatal("a live command without observed codex evidence must not degrade") + } + limited := Observation{Head: "abcdef123", Open: true, CodexActiveThisRound: true, Events: []dialect.BotEvent{usageLimit}} + if CodexOnlyEligible(r, limited, &blocked, now) { + t.Fatal("a codex usage limit since the fire must disengage the degrade") + } +} + +// TestDoneExcept covers the Codex-side completeness check for degraded rounds, +// including the vacuous guard: an excluded bot alone never reads as done. +func TestDoneExcept(t *testing.T) { + cr := "coderabbitai[bot]" + if !DoneExcept(map[string]bool{cr: false, dialect.CodexBotLogin: true}, cr) { + t.Fatal("codex reviewed, coderabbit excluded: must be done-except") + } + if DoneExcept(map[string]bool{cr: false}, cr) { + t.Fatal("only the excluded bot gates: must NOT be vacuously done") + } + if DoneExcept(map[string]bool{cr: false, dialect.CodexBotLogin: false}, cr) { + t.Fatal("codex still pending: must not be done-except") + } + if !DoneExcept(map[string]bool{cr: true, dialect.CodexBotLogin: true}, cr) { + t.Fatal("everything reviewed must be done-except too") + } + // Bot-name normalization: the suffixless form excludes the bracketed key. + if DoneExcept(map[string]bool{"coderabbitai": false}, cr) { + t.Fatal("normalized excluded bot alone must not be vacuously done") + } +} + +func TestDoneExceptWithEvidence(t *testing.T) { + cr := "coderabbitai[bot]" + tests := []struct { + name string + reviewedBy map[string]bool + want bool + }{ + {name: "feedback-only evidence supplies non-vacuous gate", reviewedBy: map[string]bool{cr: false}, want: true}, + {name: "configured evidence bot is marked reviewed", reviewedBy: map[string]bool{cr: false, dialect.CodexBotLogin: false}, want: true}, + {name: "suffixless evidence bot is normalized", reviewedBy: map[string]bool{cr: false, "chatgpt-codex-connector": false}, want: true}, + {name: "another pending reviewer still gates", reviewedBy: map[string]bool{cr: false, dialect.CodexBotLogin: false, "other[bot]": false}, want: false}, + {name: "another completed reviewer permits", reviewedBy: map[string]bool{cr: false, "other[bot]": true}, want: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := DoneExceptWithEvidence(tc.reviewedBy, cr, dialect.CodexBotLogin); got != tc.want { + t.Fatalf("DoneExceptWithEvidence() = %v, want %v", got, tc.want) + } + }) + } +} + +// TestDecideFireBlockedCodexDeferred: an account block with the degrade flag +// posts only the Codex command and keeps the round queued for CodeRabbit. +func TestDecideFireBlockedCodexDeferred(t *testing.T) { + now := t0.Add(10 * time.Minute) + blocked := now.Add(25 * time.Minute) + g := Global{SlotFree: true, BlockedUntil: &blocked} + head := "abcdef123" + queued := state.Round{Repo: "owner/repo", PR: 448, Head: head, Phase: state.PhaseQueued, Seq: 1} + open := Observation{Head: head, Open: true} + + degrade := policy + degrade.RequiredBots = []string{"coderabbitai[bot]", dialect.CodexBotLogin} + degrade.CodexCommand = "@codex review" + degrade.RateLimitCodexDegrade = true + + if d := DecideFire(g, queued, open, now, degrade); d.Verdict != FireCodexDeferred { + t.Fatalf("blocked + degrade + postable codex must defer to codex, got %+v", d) + } + // Flag off → today's behavior. + off := degrade + off.RateLimitCodexDegrade = false + if d := DecideFire(g, queued, open, now, off); d.Verdict != FireNo { + t.Fatalf("degrade off must stay FireNo blocked, got %+v", d) + } + // Codex command already posted for this round → plain blocked FireNo. + posted := queued + posted.CodexCommandID = 99 + if d := DecideFire(g, posted, open, now, degrade); d.Verdict != FireNo { + t.Fatalf("an already-posted codex command must not re-defer, got %+v", d) + } + // A live command on the PR is adopted as this round's Codex anchor rather + // than re-posted or left unrecorded. + cmdObs := Observation{Head: head, Open: true, CodexCommands: []CommandSeen{{ID: 55, CreatedAt: now}}} + if d := DecideFire(g, queued, cmdObs, now, degrade); d.Verdict != FireCodexDeferred || + d.AdoptCommandID != 55 || !d.AdoptAt.Equal(now) || d.PostCodex { + t.Fatalf("a live codex command must be adopted without re-posting, got %+v", d) + } + // Auto-active Codex reviews unprompted → nothing to post; blocked FireNo. + autoObs := Observation{Head: head, Open: true, CodexAutoActive: true} + if d := DecideFire(g, queued, autoObs, now, degrade); d.Verdict != FireNo { + t.Fatalf("auto-active codex must not be commanded, got %+v", d) + } + // Unblocked → the normal fire path is untouched. + if d := DecideFire(Global{SlotFree: true}, queued, open, now, degrade); d.Verdict != FirePost || !d.PostCodex { + t.Fatalf("unblocked fire must stay FirePost with codex, got %+v", d) + } + // A busy fire slot defers to Codex the same way — Codex needs no slot. + if d := DecideFire(Global{SlotFree: false}, queued, open, now, degrade); d.Verdict != FireCodexDeferred { + t.Fatalf("slot-busy + degrade + postable codex must defer to codex, got %+v", d) + } + if d := DecideFire(Global{SlotFree: false}, queued, open, now, off); d.Verdict != FireNo { + t.Fatalf("slot-busy with degrade off must stay FireNo, got %+v", d) + } +} + +// TestCodexOnlyEligibleUnfiredRound: stale PR-level Codex evidence must not +// defer an unfired round — only a head-bound review (or auto-activity) may. +func TestCodexOnlyEligibleUnfiredRound(t *testing.T) { + now := t0.Add(10 * time.Minute) + blocked := now.Add(25 * time.Minute) + queued := state.Round{Repo: "owner/repo", PR: 448, Head: "abcdef123", Phase: state.PhaseQueued, Seq: 1} + // A round-window flag computed from a zero cutoff (an old SHA-less review) + // must not qualify an unfired, uncommanded round. + stale := Observation{Head: "abcdef123", Open: true, CodexActiveThisRound: true} + if CodexOnlyEligible(queued, stale, &blocked, now) { + t.Fatal("stale round-window evidence must not defer an unfired round") + } + headReviewed := Observation{Head: "abcdef123", Open: true, + Reviews: []ReviewSeen{{Bot: dialect.CodexBotLogin, Commit: "abcdef1234567890", SubmittedAt: now}}} + if !CodexOnlyEligible(queued, headReviewed, &blocked, now) { + t.Fatal("a codex review of the current head must defer an unfired round") + } + auto := Observation{Head: "abcdef123", Open: true, CodexAutoActive: true} + if CodexOnlyEligible(queued, auto, &blocked, now) { + t.Fatal("auto-activity without current evidence must not defer an unfired round") + } + cleanAt := now.Add(-time.Minute) + cleanHead := Observation{Head: "abcdef123", Open: true, CodexAutoActive: true, + Events: []dialect.BotEvent{{Kind: dialect.EvCodexClean, Bot: dialect.CodexBotLogin, + SHA: "abcdef1234567890", CommentID: 601, CreatedAt: cleanAt, UpdatedAt: cleanAt}}} + if !CodexOnlyEligible(queued, cleanHead, &blocked, now) { + t.Fatal("a clean codex summary naming the current head must defer an unfired round") + } + // A deferred command anchors the window: command-bound activity (e.g. the + // usual SHA-less clean comment) qualifies, and an old usage-limit notice + // from before the command no longer disqualifies. + commandedAt := now.Add(-5 * time.Minute) + commanded := queued + commanded.CodexCommandID = 77 + commanded.CodexCommandedAt = &commandedAt + answered := Observation{Head: "abcdef123", Open: true, CodexActiveThisRound: true, + Events: []dialect.BotEvent{{Kind: dialect.EvCodexUsageLimit, Bot: dialect.CodexBotLogin, + CommentID: 600, CreatedAt: commandedAt.Add(-time.Hour), UpdatedAt: commandedAt.Add(-time.Hour)}}} + if !CodexOnlyEligible(commanded, answered, &blocked, now) { + t.Fatal("command-bound activity must defer a commanded unfired round despite an old usage-limit notice") + } +} + +// TestCompletionBindsPreFireCodexAnswer: a SHA-less Codex clean summary +// delivered after the deferred command but before the delayed CodeRabbit fire +// must still count for the round — the command time is the codex cutoff. +func TestCompletionBindsPreFireCodexAnswer(t *testing.T) { + r := firedRound(t, "abcdef123") + commandedAt := r.FiredAt.Add(-10 * time.Minute) + r.CodexCommandedAt = &commandedAt + r.CodexCommandID = 77 + gated := policy + gated.RequiredBots = []string{"coderabbitai[bot]", dialect.CodexBotLogin} + crReview := ReviewSeen{Bot: "coderabbitai[bot]", Commit: "abcdef1234567890", SubmittedAt: r.FiredAt.Add(time.Minute)} + cleanBeforeFire := dialect.BotEvent{Kind: dialect.EvCodexClean, Bot: dialect.CodexBotLogin, CommentID: 900, + CreatedAt: commandedAt.Add(2 * time.Minute), UpdatedAt: commandedAt.Add(2 * time.Minute)} + + obs := Observation{Head: "abcdef123", Open: true, Reviews: []ReviewSeen{crReview}, + Events: []dialect.BotEvent{cleanBeforeFire}} + if got := Completion(r, obs, gated); !got.Done { + t.Fatalf("a codex answer after its command but before the deferred fire must count: %+v", got) + } + // Evidence from before the command stays excluded. + early := cleanBeforeFire + early.CreatedAt = commandedAt.Add(-time.Minute) + early.UpdatedAt = early.CreatedAt + obs.Events = []dialect.BotEvent{early} + if got := Completion(r, obs, gated); got.Done { + t.Fatalf("evidence older than the codex command must not count: %+v", got) + } +} diff --git a/internal/engine/fire.go b/internal/engine/fire.go index c8cc93c..a74a0ca 100644 --- a/internal/engine/fire.go +++ b/internal/engine/fire.go @@ -14,14 +14,15 @@ import ( type FireVerdict int const ( - FireNo FireVerdict = iota // skip this pass (Reason says why) - FirePost // reserve the slot and post the command - FireAdopt // a command is already on the PR — adopt it - FireDedupe // bot already reviewed this head — complete without firing - FireCodexOnly // CodeRabbit reviewed the head but a required Codex still must — post only the Codex command - FireCoReviewWait // CodeRabbit reviewed the head; a gating co-bot has not — wait for it, bounded, without posting or holding the slot - FireSupersede // observed head differs — supersede the round first - FireDrop // PR closed/merged — abandon the round + FireNo FireVerdict = iota // skip this pass (Reason says why) + FirePost // reserve the slot and post the command + FireAdopt // a command is already on the PR — adopt it + FireDedupe // bot already reviewed this head — complete without firing + FireCodexOnly // CodeRabbit reviewed the head but a required Codex still must — post only the Codex command + FireCoReviewWait // CodeRabbit reviewed the head; a gating co-bot has not — wait for it, bounded, without posting or holding the slot + FireCodexDeferred // account blocked — post only the Codex command now; the round stays queued so CodeRabbit fires when the window opens + FireSupersede // observed head differs — supersede the round first + FireDrop // PR closed/merged — abandon the round ) type FireDecision struct { @@ -64,6 +65,25 @@ func DecideFire(g Global, r state.Round, obs Observation, now time.Time, p Polic return FireDecision{Verdict: FireNo, Reason: reason} } if !g.SlotFree { + // Codex needs no fire slot: a round parked behind another PR's + // in-flight review can start its Codex round immediately. The round + // stays queued and CodeRabbit fires once the slot frees, with + // CodexCommandID preventing a duplicate Codex post. NOT for a head + // CodeRabbit already reviewed — that round belongs to the dedupe + // resolution below once the slot frees (a queued round Codex answers + // clean cannot complete, so deferring it here could wedge the wait). + reviewedHead := false + for _, review := range obs.Reviews { + if sameBot(review.Bot, p.Bot) && review.Commit != "" && strings.HasPrefix(review.Commit, obs.Head) { + reviewedHead = true + break + } + } + if !reviewedHead { + if d, ok := decideCodexDeferred(r, obs, p, "fire slot busy"); ok { + return d + } + } return FireDecision{Verdict: FireNo, Reason: "fire slot busy"} } // Belt-and-braces live check: even with a fresh round, never fire at a @@ -80,6 +100,14 @@ func DecideFire(g Global, r state.Round, obs Observation, now time.Time, p Polic } } if g.BlockedUntil != nil && g.BlockedUntil.After(now) { + // Degrade instead of stalling: the block only gates CodeRabbit quota, + // so ask Codex now and leave the round queued — CodeRabbit still + // fires the moment the window opens. DecideCodexPost's guards + // (command configured, codex required, not auto-active, no live or + // already-posted command) make this idempotent per round. + if d, ok := decideCodexDeferred(r, obs, p, "account blocked"); ok { + return d + } return FireDecision{Verdict: FireNo, Reason: "account blocked until " + g.BlockedUntil.UTC().Format(time.RFC3339)} } if g.LastFired != nil && now.Sub(*g.LastFired) < p.MinInterval { @@ -108,6 +136,43 @@ func DecideFire(g Global, r state.Round, obs Observation, now time.Time, p Polic return FireDecision{Verdict: FirePost, PostCodex: postCodex} } +// decideCodexDeferred starts or adopts the Codex half of a round while +// CodeRabbit cannot fire. CodexCommands is cutoff-filtered by observe(), so an +// existing command here is safe to bind to this head and must be recorded as +// the round anchor rather than merely suppressing a duplicate post. +func decideCodexDeferred(r state.Round, obs Observation, p Policy, reason string) (FireDecision, bool) { + if !p.RateLimitCodexDegrade || r.CodexCommandID != 0 { + return FireDecision{}, false + } + if DecideCodexPost(r, obs, p, len(obs.CodexCommands) > 0) { + return FireDecision{ + Verdict: FireCodexDeferred, + Reason: reason + "; requesting codex review now, coderabbit deferred", + PostCodex: true, + }, true + } + var newest *CommandSeen + for i := range obs.CodexCommands { + cmd := &obs.CodexCommands[i] + if newest == nil || cmd.CreatedAt.After(newest.CreatedAt) { + newest = cmd + } + } + if newest == nil { + return FireDecision{}, false + } + at := newest.CreatedAt + if at.IsZero() { + at = newest.UpdatedAt + } + return FireDecision{ + Verdict: FireCodexDeferred, + Reason: reason + "; adopting existing codex review command, coderabbit deferred", + AdoptCommandID: newest.ID, + AdoptAt: at, + }, true +} + // codexAwareDedupe resolves what to do when CodeRabbit already reviewed the head. // If no gating Codex is still outstanding, the round is genuinely done (FireDedupe). // If a required-or-auto-active Codex has no review of this head yet, the round is diff --git a/internal/state/state.go b/internal/state/state.go index 821a6ac..7bfa97f 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -67,6 +67,12 @@ type Round struct { // for the same round. A stale claim (the poster died mid-flight) expires and // may be re-claimed. CodexClaimedAt *time.Time `json:"codex_claimed_at,omitempty"` + // CodexCommandedAt records when crq posted this round's Codex command. It + // can precede FiredAt (a deferred post while queued behind a rate-limit + // window or busy slot), and Codex evidence binds from it — otherwise a + // SHA-less Codex answer delivered before the delayed CodeRabbit fire + // would be ignored by the completion cutoff and never re-requested. + CodexCommandedAt *time.Time `json:"codex_commanded_at,omitempty"` // LastAttemptAt is the adoption cutoff: command comments older than the // most recent failed/abandoned attempt must not be adopted as this round's diff --git a/llms.txt b/llms.txt index 5d2fb62..e684910 100644 --- a/llms.txt +++ b/llms.txt @@ -48,6 +48,9 @@ The loop also returns as soon as any feedback bot reports a finding; it does not remaining required bots or a shared review slot. Fix those findings locally, but HOLD THE HEAD: resolve each addressed thread immediately, but do not commit or push while any `.reviewed_by` value is false, because changing the head restarts the pending checks while thread resolution does not. +Exception: when `.coderabbit_deferred` is true and every non-CodeRabbit required reviewer is true, +the round is Codex-only (CodeRabbit rate-limited) — fix, push, and loop again without holding the +head; the queued CodeRabbit review is untouched. Keep the queued review alive and poll `crq feedback` with the same `CRQ_REQUIRED_BOTS`. When every required bot is true, fix and resolve the rest, then commit all fixes once and push once. @@ -67,17 +70,30 @@ never when Codex auto-reviews the PR on its own. A Codex that joins a round unin round's convergence dynamically (its usage-limit notice releases the dynamic gate). If a bot reports actionable findings first, it emits them immediately while leaving the queued round alive. Fix and validate locally and resolve or decline each addressed thread immediately. While any required bot's -`.reviewed_by` value is false, do not commit or push. Only after every `CRQ_REQUIRED_BOTS` entry is true -for the current head should you fix and resolve the rest, then commit and push the combined fixes once. -Convergence is reported only after that same condition is met. +`.reviewed_by` value is false, do not commit or push — unless `.coderabbit_deferred` is true and all +non-CodeRabbit required reviewers are true, in which case push freely: the round is Codex-only while +CodeRabbit is rate-limited. Only after every +`CRQ_REQUIRED_BOTS` entry is true for the current head should you fix and resolve the rest, then +commit and push the combined fixes once. Convergence is reported only after that same condition is met. Exit codes: -- `0`: done; converged or no actionable findings. +- `0`: done; converged or no actionable findings. Check `.coderabbit_deferred`: when true (with + `.status == "deferred"`), Codex reviewed clean but the CodeRabbit review is still owed — this is + progress, NOT convergence (`.converged` stays false). Re-run `crq loop` after + `.coderabbit_deferred_until` or on the next push; the queued CodeRabbit review fires by itself + once the window opens. - `10`: read `.findings[]`, fix valid issues, and validate locally. If any `.reviewed_by` value is - false, resolve addressed threads but hold the head. After all are true, fix/resolve the rest, - commit/push once, then repeat. + false, resolve addressed threads but hold the head — EXCEPT when `.coderabbit_deferred` is true + and all non-CodeRabbit required reviewers are true: those are Codex-only findings during a + CodeRabbit rate-limit window, so fix, push, and loop again (each Codex round takes minutes; + several fit inside one CodeRabbit window). - `2`: timed out waiting for feedback; do not push a stale-feedback round. +Rate-limit degrade: when CodeRabbit is rate-limited ("Review limit reached … Next review available +in: NN minutes") and Codex demonstrably reviews the PR, the loop degrades to Codex-only rounds +instead of waiting the window out, and the pump posts `CRQ_CODEX_CMD` for blocked rounds while +keeping the CodeRabbit review queued. Disable with `CRQ_RL_CODEX_DEGRADE=0` (default on). + Resolve addressed threads on GitHub (crq keys off GitHub resolution state; a finding keeps reappearing until its thread is resolved there): diff --git a/skills/coderabbit-queue/SKILL.md b/skills/coderabbit-queue/SKILL.md index 95347b1..0b97cbd 100644 --- a/skills/coderabbit-queue/SKILL.md +++ b/skills/coderabbit-queue/SKILL.md @@ -40,10 +40,21 @@ and `CODERABBIT_API_KEY` presence for headless local review. Exit codes: -- `0`: converged or no actionable findings -- `10`: actionable findings were written to JSON +- `0`: converged or no actionable findings. Check `.coderabbit_deferred` first: when true (with + `.status == "deferred"`), Codex reviewed clean while CodeRabbit is rate-limited — the CodeRabbit + review is still owed and `.converged` is false. Treat it as progress, not convergence: re-run + `crq loop` after `.coderabbit_deferred_until` or on the next push. +- `10`: actionable findings were written to JSON. When `.coderabbit_deferred` is true and all + non-CodeRabbit required reviewers are true, these are Codex-only findings during a CodeRabbit + rate-limit window: fix, push, and loop again immediately instead of holding the head — the queued + CodeRabbit review fires by itself once the window opens. - `2`: timed out waiting for feedback +Rate-limit degrade (default on, `CRQ_RL_CODEX_DEGRADE=0` disables): when CodeRabbit is +rate-limited and Codex demonstrably reviews the PR, the loop returns Codex feedback promptly +instead of waiting out the window, and the pump posts the Codex command for blocked rounds while +keeping the CodeRabbit review queued. + ## Drain Findings Before Waiting An autonomous review loop is a work loop, not a review-status waiter. Before starting or