diff --git a/src/queue/processors.ts b/src/queue/processors.ts index df986f7359..23031ca808 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1080,13 +1080,13 @@ export async function processJob(env: Env, message: JobMessage): Promise { // One bounded re-gate unit fanned out by the sweep (#audit-sweep-fanout): re-review + stamp a single PR. await regatePullRequest( env, - message.repairHeadSha, message.repoFullName, message.prNumber, message.installationId, message.deliveryId, message.force, message.prCreatedAt, + message.repairHeadSha, ); return; case "run-agent": @@ -1462,13 +1462,17 @@ async function refreshOpenPullRequestsForScheduledSweep( // surfaceRepairPriorityPullNumbers has no memory of prior attempts -- if the repair keeps failing for the SAME // head SHA (e.g. every AI-provider attempt times out), it would otherwise re-select that PR forever, burning a // fresh review attempt every cycle for zero output. These two constants cap that: once a SHA has already had -const REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA = 5; +// REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA dispatches recorded, it drops back to ordinary staleness-gated candidacy // (still eventually re-checked, just not on every tick) and a single REGATE_REPAIR_EXHAUSTED_EVENT_TYPE audit // event is recorded so the stuck PR is visible instead of silently retried forever. A new commit changes the // head SHA, which resets the count naturally (the target key is scoped to repo+PR+SHA). const REGATE_REPAIR_ATTEMPT_EVENT_TYPE = "agent.sweep.regate.repair_attempt"; const REGATE_REPAIR_EXHAUSTED_EVENT_TYPE = "agent.sweep.regate.repair_exhausted"; -const REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA = 2; +// Bumped 2 -> 5 (#3998): the dispatch-time recording below double-counted rate-limit-deferred attempts as +// real ones, exhausting the budget before the repair ever actually ran. Recording moved to execution-time +// (regatePullRequest) makes each count a genuine attempt, so a higher cap gives real transient failures +// (e.g. a single AI-provider timeout) more room without letting a truly broken PR retry indefinitely. +const REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA = 5; const REGATE_REPAIR_ATTEMPT_LOOKBACK_MS = 24 * 60 * 60 * 1000; function regateRepairTargetKey(repoFullName: string, prNumber: number, headSha: string): string { @@ -1871,21 +1875,21 @@ async function sweepRepoRegate( const job: JobMessage = { type: "agent-regate-pr", deliveryId: isPriorityRepair - // #orb-retry-storm: pass the repair SHA so regatePullRequest can record the attempt at - // execution time (after rate-limit admission), not here at dispatch time. Jobs that are - // deferred or dropped before they run no longer count against the per-SHA cap. - ...(isPriorityRepair && pr.headSha ? { repairHeadSha: pr.headSha } : {}), ? `regate-repair:${repoFullName}#${pr.number}` : `regate-sweep:${repoFullName}#${pr.number}`, repoFullName, prNumber: pr.number, installationId: sweepInstallationId, - targetKey: regateRepairTargetKey(repoFullName, pr.number, pr.headSha), - outcome: "queued", - detail: `outage-repair re-review dispatched for ${repoFullName}#${pr.number}`, - metadata: { repoFullName, prNumber: pr.number, headSha: pr.headSha }, - }); - } + ...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}), + // #orb-retry-storm: pass the repair SHA so regatePullRequest can record the attempt at + // execution time (after rate-limit admission), not here at dispatch time. Jobs that are + // deferred or dropped before they run no longer count against the per-SHA cap. + ...(isPriorityRepair && pr.headSha ? { repairHeadSha: pr.headSha } : {}), + }; + const delaySeconds = Math.min(index * 10, 600); + await (delaySeconds > 0 + ? env.JOBS.send(job, { delaySeconds }) + : env.JOBS.send(job)); } } await recordAuditEvent(env, { @@ -2089,13 +2093,13 @@ async function sweepRepoBacklogConvergence( // per-PR job always re-evaluates the head. async function regatePullRequest( env: Env, - repairHeadSha?: string, repoFullName: string, prNumber: number, installationId: number, deliveryId: string, force?: boolean, prCreatedAt?: string | null, + repairHeadSha?: string, ): Promise { // Reserve installation rate-limit headroom (#audit-rate-headroom): all repos share ONE GitHub App installation // = ONE REST bucket, so when the shared budget is low, DEFER this re-review until the reset instead of @@ -2117,11 +2121,17 @@ async function regatePullRequest( await env.JOBS.send( { type: "agent-regate-pr", - ...(repairHeadSha ? { repairHeadSha } : {}), deliveryId, repoFullName, prNumber, installationId, + ...(prCreatedAt ? { prCreatedAt } : {}), + ...(force ? { force: true } : {}), + ...(repairHeadSha ? { repairHeadSha } : {}), + }, + { delaySeconds: delayUntil(rateResetAt) }, + ); + return; } // #orb-retry-storm: record the repair attempt NOW — after rate-limit admission — so the cap in // surfaceRepairPriorityPullNumbers counts actual executions, not queued dispatches that may have @@ -2132,16 +2142,10 @@ async function regatePullRequest( eventType: REGATE_REPAIR_ATTEMPT_EVENT_TYPE, actor: "gittensory", targetKey: regateRepairTargetKey(repoFullName, prNumber, repairHeadSha), - outcome: "started", + outcome: "queued", detail: `outage-repair re-review executing for ${repoFullName}#${prNumber}`, metadata: { repoFullName, prNumber, headSha: repairHeadSha }, }); - ...(prCreatedAt ? { prCreatedAt } : {}), - ...(force ? { force: true } : {}), - }, - { delaySeconds: delayUntil(rateResetAt) }, - ); - return; } const settings = await resolveRepositorySettings(env, repoFullName); await reReviewStoredPullRequest( diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index a9cf622176..c72403418a 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -6861,12 +6861,13 @@ describe("queue processors", () => { await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9407); await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); // PR 1: missing its current Gate check for its current head -- would ordinarily be flagged outage-repair - // priority on every tick. Pre-seed 3 prior repair-attempt audit events for this EXACT head SHA to simulate a - // review that keeps failing (e.g. a timeout) and never publishes a completed gate check. + // priority on every tick. Pre-seed REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA (5) prior repair-attempt audit + // events for this EXACT head SHA to simulate a review that keeps failing (e.g. a timeout) and never + // publishes a completed gate check. await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 1, title: "Stuck repair", state: "open", user: { login: "c" }, head: { sha: "stuck-sha" }, labels: [], body: "" }); await repositoriesModule.markPullRequestSurfacePublished(env, "owner/agent-repo", 1, "stuck-sha"); const targetKey = "owner/agent-repo#1#stuck-sha"; - for (let i = 0; i < 3; i += 1) { + for (let i = 0; i < 5; i += 1) { await repositoriesModule.recordAuditEvent(env, { eventType: "agent.sweep.regate.repair_attempt", actor: "gittensory", @@ -6894,18 +6895,18 @@ describe("queue processors", () => { const attempts = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") .bind("agent.sweep.regate.repair_attempt", targetKey) .first<{ n: number }>(); - expect(attempts?.n).toBe(3); + expect(attempts?.n).toBe(5); // Sentry-visible signal (via the structured-log forwarder) fires exactly once alongside the audit event. const exhaustedLogs = errors.mock.calls.filter(([line]) => typeof line === "string" && line.includes("regate_repair_exhausted")); expect(exhaustedLogs).toHaveLength(1); const logged = JSON.parse(exhaustedLogs[0]![0] as string) as Record; - expect(logged).toMatchObject({ level: "error", event: "regate_repair_exhausted", repo: "owner/agent-repo", pullNumber: 1, headSha: "stuck-sha", attempts: 3 }); + expect(logged).toMatchObject({ level: "error", event: "regate_repair_exhausted", repo: "owner/agent-repo", pullNumber: 1, headSha: "stuck-sha", attempts: 5 }); } finally { errors.mockRestore(); } }); - it("REGRESSION (#orb-retry-storm): a repair dispatch under the attempt cap records a repair_attempt audit event, and a second sweep tick does not duplicate the repair_exhausted event once already flagged", async () => { + it("REGRESSION (#orb-retry-storm, #3998): a repair dispatch alone does not record a repair_attempt audit event (only execution does), and a second sweep tick does not duplicate the repair_exhausted event once already flagged", async () => { const sent: import("../../src/types").JobMessage[] = []; const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); await upsertInstallation(env, { action: "created", installation: { id: 9408, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); @@ -6920,14 +6921,18 @@ describe("queue processors", () => { const fanned = sent.filter((job): job is Extract => job.type === "agent-regate-pr"); expect(fanned.map((job) => job.deliveryId)).toContain("regate-repair:owner/agent-repo#2"); - const attemptsAfterFirst = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") + // #orb-retry-storm (#3998): dispatching the repair job no longer records an attempt by itself -- only + // actually EXECUTING it does (inside regatePullRequest, after rate-limit admission), so a deferred or + // dropped dispatch that never runs no longer exhausts the per-SHA cap. + const attemptsAfterDispatch = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") .bind("agent.sweep.regate.repair_attempt", targetKey) .first<{ n: number }>(); - expect(attemptsAfterFirst?.n).toBe(1); + expect(attemptsAfterDispatch?.n).toBe(0); - // Manually push this SHA over the cap, then run the sweep twice more -- the exhausted event must be recorded - // only once even though the PR is (re-)evaluated on every tick. - for (let i = 0; i < 2; i += 1) { + // Manually seed REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA (5) executed attempts to push this SHA over the cap, + // then run the sweep twice more -- the exhausted event must be recorded only once even though the PR is + // (re-)evaluated on every tick. + for (let i = 0; i < 5; i += 1) { await repositoriesModule.recordAuditEvent(env, { eventType: "agent.sweep.regate.repair_attempt", actor: "gittensory", @@ -27223,7 +27228,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri // #terminal-outcome-audit: the disposition counter's "close" action_class, with the actual gate-blocker // code (missing_linked_issue, from the default linkedIssueGateMode:block + no-linked-issue body) as the // bounded blocker_class -- proof this reaches the real gate.blockers, not just a hardcoded label. - expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="close",autonomy_level="auto",blocker_class="missing_linked_issue"} 1'); + expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="close",autonomy_level="auto",blocker_class="missing_linked_issue",repo="redacted-1"} 1'); const nativeDecision = await env.DB.prepare("select decision, summary, source from review_audit where event_type = 'gate_decision' and target_id = ?").bind(`${REPO}#60`).first<{ decision: string; summary: string; source: string }>(); expect(nativeDecision).toMatchObject({ decision: "close", summary: "missing_linked_issue", source: "gittensory-native" }); }); @@ -27266,7 +27271,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri expect(seen.merged).toBe(false); expect(seen.closed).toBe(false); - expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="auto",blocker_class="guardrail_hold"} 1'); + expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="auto",blocker_class="guardrail_hold",repo="redacted-1"} 1'); const holdAudit = await env.DB.prepare("select metadata_json from audit_events where event_type = 'agent.action.hold' order by created_at desc limit 1").first<{ metadata_json: string }>(); expect(JSON.parse(holdAudit?.metadata_json ?? "{}")).toMatchObject({ repoFullName: REPO, @@ -27394,7 +27399,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri const mergeAudit = await env.DB.prepare("select count(*) as n from audit_events where event_type = 'agent.action.merge'").first<{ n: number }>(); expect(mergeAudit?.n).toBeGreaterThanOrEqual(1); // #terminal-outcome-audit: the disposition counter's "merge" action_class, on the actual live call site. - expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="merge",autonomy_level="auto",blocker_class="none"} 1'); + expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="merge",autonomy_level="auto",blocker_class="none",repo="redacted-1"} 1'); }); // #terminal-outcome-audit: end-to-end proof that the LIVE runAgentMaintenancePlanAndExecute call site (not just @@ -27468,7 +27473,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri // early return -- this is the exact "hold, but no audit_events row at all" shape (the breaker downgrade // leaves no merge/close action) that previously had zero aggregate signal. close autonomy is unset in this // repo's settings (only merge/approve are configured), so it resolves to the default "observe". - expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="observe",blocker_class="none"} 1'); + expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="observe",blocker_class="none",repo="redacted-1"} 1'); const holdAudit = await env.DB.prepare("select detail, metadata_json from audit_events where event_type = 'agent.action.hold' order by created_at desc limit 1").first<{ detail: string; metadata_json: string }>(); expect(holdAudit?.detail).toBe("auto-action held by precision circuit breaker"); expect(JSON.parse(holdAudit?.metadata_json ?? "{}")).toMatchObject({