diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 01eeccc05e..195b2dba70 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -261,6 +261,7 @@ import { } from "../settings/agent-execution"; import { ISSUE_WAKE_MAX_PRS, + MERGE_WAKE_MAX_PRS, SWEEP_FANOUT_DEDUP_MS, SWEEP_MAX_PRS, isRegateSweepDraining, @@ -1730,6 +1731,55 @@ async function maybeEnqueueRagReindexForMergedPr( }); } +/** + * Event-driven re-gate trigger on sibling PR merge (#4005): companion to the merge-train gate. When a PR + * MERGES, every OTHER open PR's gate verdict can be invalidated by it (a newly-conflicting base, a duplicate + * cluster now missing its winner, a linked-issue cap that just freed up) with nothing proactively re-checking + * it -- the scheduled sweep is bounded to SWEEP_MAX_PRS per repo per ~2-minute tick and can take several + * cycles to reach a given sibling. Enqueue a bounded, staggered `agent-regate-pr` job per sibling right away + * instead of waiting for the next sweep pass to notice the drift. + * + * Fires ONLY on a genuine merge -- `action === "closed"` AND a `merged_at` timestamp (an ordinary close changed + * nothing on the base branch, so siblings have nothing new to react to; mirrors maybeEnqueueRagReindexForMergedPr's + * own merge check just above). Scoped to the SAME repos the re-gate sweep already covers (self-host convergence- + * allowlisted OR hosted agent-configured) -- this closes the "stale sibling" latency gap for repos already + * getting proactive re-gates, not a scope expansion to repos that never were. `otherOpenPullRequests` is the + * caller's already-fetched, already-bounded (100-row, ascending-by-number) sibling list — reused as-is rather than + * re-querying, so the lowest-numbered open siblings are re-gated first, same tie-break the duplicate-winner + * election uses elsewhere. Best-effort: enqueue failures are logged by the caller, never surfaced to the gate. + */ +async function maybeEnqueueSiblingRegateForMergedPr( + env: Env, + deliveryId: string, + repoFullName: string, + action: string | undefined, + mergedAt: string | null | undefined, + installationId: number, + settings: RepositorySettings, + otherOpenPullRequests: readonly PullRequestRecord[], +): Promise { + // action is only ever undefined before shouldProcessPullRequestPublicSurface's own action-set check has + // already passed at the call site, so a direct comparison (no nullish fallback needed) keeps this line's + // branches exhaustively reachable -- unlike maybeEnqueueRagReindexForMergedPr's `?? ""`, which predates this. + if (action !== "closed" || !mergedAt) return; + if (!(isConvergenceRepoAllowed(env, repoFullName) || isAgentConfigured(settings.autonomy))) return; + const siblings = otherOpenPullRequests.slice(0, MERGE_WAKE_MAX_PRS); + for (const [index, sibling] of siblings.entries()) { + const job: JobMessage = { + type: "agent-regate-pr", + deliveryId, + repoFullName, + prNumber: sibling.number, + installationId, + ...(sibling.createdAt ? { prCreatedAt: sibling.createdAt } : {}), + }; + const delaySeconds = Math.min(index * 10, 600); + await (delaySeconds > 0 + ? env.JOBS.send(job, { delaySeconds }) + : env.JOBS.send(job)); + } +} + // Recompute the DETERMINISTIC gate verdict for a repo's stalest open PRs and record it as an audit event — // ADVISORY ONLY: nothing is published to GitHub (no check, comment, or label) and no PR is mutated. This is // the Phase-0 scheduling rail; the action layer (#778) is what will later turn a flagged verdict into a real @@ -6165,6 +6215,31 @@ async function processGitHubWebhook( }), ); }); + // Event-driven sibling re-gate (#4005): a merge can invalidate every OTHER open PR's gate verdict, and + // otherwise nothing re-checks them until the next bounded sweep tick reaches each one. Enqueued (not run + // inline), same shape as the RAG re-index just above. Best-effort. + await maybeEnqueueSiblingRegateForMergedPr( + env, + deliveryId, + repoFullName, + payload.action, + payload.pull_request.merged_at, + installationId, + settings, + otherOpenPullRequests, + ).catch((error) => { + /* v8 ignore next -- best-effort: a sibling re-gate enqueue failure is logged, never surfaced to the gate. */ + console.error( + JSON.stringify({ + level: "warn", + event: "sibling_regate_enqueue_failed", + deliveryId, + repository: repoFullName, + pullNumber: pr.number, + error: errorMessage(error), + }), + ); + }); } } diff --git a/src/settings/agent-sweep.ts b/src/settings/agent-sweep.ts index 4d2e701330..8979198217 100644 --- a/src/settings/agent-sweep.ts +++ b/src/settings/agent-sweep.ts @@ -26,6 +26,16 @@ export const SWEEP_MAX_PRS = 3; // worst case ~25 x 9 = 225 REST calls, staggered by the same delaySeconds window the caller already uses. export const ISSUE_WAKE_MAX_PRS = 25; +// Sibling-merge wake budget (#4005): companion to the merge-train gate -- when a PR MERGES, every OTHER open PR's +// verdict can be invalidated (a newly-conflicting base, a duplicate cluster missing its winner, a linked-issue cap +// that just freed up) with nothing proactively re-checking it until the next sweep tick. This handler fires ONCE +// per merge, same one-shot shape as ISSUE_WAKE_MAX_PRS, but a merge is a far MORE common trigger than an issue +// label/assignment change -- a busy repo can merge many PRs an hour, each firing this fan-out, so reusing +// ISSUE_WAKE_MAX_PRS's 25 would let repeated merges inside one rate-limit window compound in a way the rarer +// issue-wake trigger never does. 15 keeps each merge's worst case at 15 x 9 ≈ 135 REST calls (same ~9-REST-GET +// per-PR re-review cost as the other agent-regate-pr fan-outs), staggered by the same delaySeconds window. +export const MERGE_WAKE_MAX_PRS = 15; + // Skip-if-fresh window: a PR touched within this span was almost certainly just gated by its webhook, so the // sweep leaves it alone for that brief moment to avoid racing the in-flight webhook review. Kept SHORT (2 min) // because the sweep is now LIGHT (re-gate + act, no AI) and runs every ~2 min — a just-approved PR must be diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 0ea063d51a..e78d446cf9 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -67,7 +67,7 @@ import { fetchPullRequestFreshness, } from "../../src/github/pr-freshness"; import { createTestEnv } from "../helpers/d1"; -import { ISSUE_WAKE_MAX_PRS, SWEEP_MAX_PRS } from "../../src/settings/agent-sweep"; +import { ISSUE_WAKE_MAX_PRS, MERGE_WAKE_MAX_PRS, SWEEP_MAX_PRS } from "../../src/settings/agent-sweep"; import { AGENT_LABEL_PENDING_CLOSURE, DEFAULT_LINKED_ISSUE_HARD_RULES } from "../../src/review/linked-issue-hard-rules"; vi.mock("../../src/github/pr-freshness", async (importOriginal) => { @@ -3287,6 +3287,145 @@ describe("queue processors", () => { ); }); + it("sibling re-gate fan-out (#4005): a merged PR enqueues a bounded agent-regate-pr job for each open sibling PR", async () => { + const sent: Array<{ message: import("../../src/types").JobMessage; options?: QueueSendOptions }> = []; + const env = createTestEnv({ + GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem(), + GITTENSORY_REVIEW_REPOS: "owner/agent-repo", + JOBS: { + async send(message: import("../../src/types").JobMessage, options?: QueueSendOptions) { + sent.push(options ? { message, options } : { message }); + }, + } as unknown as Queue, + }); + await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + for (const number of [10, 11, 12]) { + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number, title: `Sibling PR ${number}`, state: "open", user: { login: "contributor" }, head: { sha: `sib${number}` }, labels: [], body: "No linked issue.", created_at: `2026-07-0${number - 9}T00:00:00.000Z` }); + } + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url === "https://api.gittensor.io/miners") return Response.json([]); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (url.includes("/check-runs")) return Response.json({ id: 900 }, { status: 201 }); + if (url.includes("/files")) return Response.json([]); + return new Response("not found", { status: 404 }); + }); + + await processJob(env, { + type: "github-webhook", + deliveryId: "sibling-merge-fanout", + eventName: "pull_request", + payload: { + action: "closed", + installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" } }, + repository: { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, + pull_request: { number: 999, title: "Merged PR", state: "closed", merged_at: "2026-07-08T00:00:00.000Z", user: { login: "contributor" }, head: { sha: "mergedsha" }, labels: [], body: "No linked issue." }, + } as never, + }); + + // Bounded, staggered fan-out for each OTHER open PR — never for the merged PR's own number. + const regateJobs = sent.filter(({ message }) => message.type === "agent-regate-pr"); + expect(regateJobs.map(({ message }) => message)).toEqual([ + expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 10, installationId: 9001, prCreatedAt: "2026-07-01T00:00:00.000Z" }), + expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 11, installationId: 9001, prCreatedAt: "2026-07-02T00:00:00.000Z" }), + expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 12, installationId: 9001, prCreatedAt: "2026-07-03T00:00:00.000Z" }), + ]); + expect(regateJobs.map(({ options }) => options)).toEqual([undefined, { delaySeconds: 10 }, { delaySeconds: 20 }]); + }); + + it("sibling re-gate fan-out (#4005): closing a PR WITHOUT a merge does not enqueue any sibling re-gate", async () => { + const sent: Array<{ message: import("../../src/types").JobMessage; options?: QueueSendOptions }> = []; + const env = createTestEnv({ + GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem(), + GITTENSORY_REVIEW_REPOS: "owner/agent-repo", + JOBS: { + async send(message: import("../../src/types").JobMessage, options?: QueueSendOptions) { + sent.push(options ? { message, options } : { message }); + }, + } as unknown as Queue, + }); + await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + for (const number of [10, 11, 12]) { + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number, title: `Sibling PR ${number}`, state: "open", user: { login: "contributor" }, head: { sha: `sib${number}` }, labels: [], body: "No linked issue." }); + } + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url === "https://api.gittensor.io/miners") return Response.json([]); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (url.includes("/check-runs")) return Response.json({ id: 900 }, { status: 201 }); + if (url.includes("/files")) return Response.json([]); + return new Response("not found", { status: 404 }); + }); + + await processJob(env, { + type: "github-webhook", + deliveryId: "sibling-close-no-merge", + eventName: "pull_request", + payload: { + action: "closed", + installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" } }, + repository: { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, + pull_request: { number: 999, title: "Closed without merge", state: "closed", merged_at: null, user: { login: "contributor" }, head: { sha: "closedsha" }, labels: [], body: "No linked issue." }, + } as never, + }); + + // An ordinary close changed nothing on the base branch — no sibling has anything new to react to. + const regateJobs = sent.filter(({ message }) => message.type === "agent-regate-pr"); + expect(regateJobs).toEqual([]); + }); + + it("sibling re-gate fan-out (#4005): the fan-out is capped at MERGE_WAKE_MAX_PRS even with more open siblings", async () => { + const sent: Array<{ message: import("../../src/types").JobMessage; options?: QueueSendOptions }> = []; + const env = createTestEnv({ + GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem(), + GITTENSORY_REVIEW_REPOS: "owner/agent-repo", + JOBS: { + async send(message: import("../../src/types").JobMessage, options?: QueueSendOptions) { + sent.push(options ? { message, options } : { message }); + }, + } as unknown as Queue, + }); + await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + for (let number = 1; number <= MERGE_WAKE_MAX_PRS + 2; number += 1) { + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number, title: `Sibling PR ${number}`, state: "open", user: { login: "contributor" }, head: { sha: `sib${number}` }, labels: [], body: "No linked issue." }); + } + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url === "https://api.gittensor.io/miners") return Response.json([]); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (url.includes("/check-runs")) return Response.json({ id: 900 }, { status: 201 }); + if (url.includes("/files")) return Response.json([]); + return new Response("not found", { status: 404 }); + }); + + await processJob(env, { + type: "github-webhook", + deliveryId: "sibling-merge-fanout-bounded", + eventName: "pull_request", + payload: { + action: "closed", + installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" } }, + repository: { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, + pull_request: { number: 999, title: "Merged PR", state: "closed", merged_at: "2026-07-08T00:00:00.000Z", user: { login: "contributor" }, head: { sha: "mergedsha" }, labels: [], body: "No linked issue." }, + } as never, + }); + + // MERGE_WAKE_MAX_PRS + 2 siblings are open, but only the first MERGE_WAKE_MAX_PRS (lowest-numbered, same + // ordering listOtherOpenPullRequests already returns) are enqueued -- a repo with many open PRs must not be + // able to turn one merge into an unbounded burst of ~9-REST-GET re-gates. + const regateJobs = sent.filter(({ message }) => message.type === "agent-regate-pr"); + expect(regateJobs).toHaveLength(MERGE_WAKE_MAX_PRS); + expect(regateJobs.map(({ message }) => (message as { prNumber: number }).prNumber)).toEqual( + Array.from({ length: MERGE_WAKE_MAX_PRS }, (_, index) => index + 1), + ); + }); + it("REGRESSION (#2371): a coalesced issue-side signal schedules a trailing re-review so an add-then-remove sequence is never lost", async () => { // Unlike CI-completion events, same-PR issue-side events are NOT interchangeable within the window: a // label ADD immediately followed by a REMOVE carries genuinely different states. The first event's