diff --git a/src/queue/processors.ts b/src/queue/processors.ts index d1f68dc76d..e2811f0119 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1321,9 +1321,24 @@ async function maybeEnqueueRagReindexForMergedPr( * 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. + * caller's already-fetched, already-bounded (100-row, ascending-by-number) sibling list — reused as-is rather + * than re-querying, so the fallback tier below still returns the lowest-numbered open siblings, same tie-break + * the duplicate-winner election uses elsewhere. Best-effort: enqueue failures are logged by the caller, never + * surfaced to the gate. + * + * #7438-incident: `otherOpenPullRequests`'s ascending-by-number order exists for duplicate-winner election + * (the lowest number IS the winner), not for THIS function's own relevance question -- reused as-is, it meant + * the woken set was always "the MERGE_WAKE_MAX_PRS oldest open PRs in the repo", with zero relation to which + * PRs this specific merge could actually invalidate. On a repo with that many older open PRs already sitting + * around (increasingly likely once one carries a maintainer-only manual-review hold that nothing auto-clears), + * a genuinely-affected, newer duplicate-cluster sibling could be silently excluded from ever being woken -- + * exactly the shape of a real incident (a duplicate-cluster loser left stuck open, its own regate landing on + * the wrong side of the winner's merge by seconds, with no fast follow-up look). `mergedPullRequestLinkedIssues` + * -- the just-merged PR's own linked issues -- lets this function put every sibling that shares one of them + * FIRST, ahead of the generic oldest-first fallback: those are the ones a duplicate-cluster-winner or + * linked-issue-cap change can concretely affect, so they must never lose a slot to an unrelated older PR. The + * fallback tier (any other open PR, oldest first) is preserved unchanged for the "newly-conflicting base" + * case, where no single relevance signal exists and PR age remains as reasonable a heuristic as any. */ async function maybeEnqueueSiblingRegateForMergedPr( env: Env, @@ -1334,13 +1349,18 @@ async function maybeEnqueueSiblingRegateForMergedPr( installationId: number, settings: RepositorySettings, otherOpenPullRequests: readonly PullRequestRecord[], + mergedPullRequestLinkedIssues: readonly number[], ): 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); + const linkedIssueSet = new Set(mergedPullRequestLinkedIssues); + const linkedSiblings = linkedIssueSet.size === 0 ? [] : otherOpenPullRequests.filter((sibling) => sibling.linkedIssues.some((issue) => linkedIssueSet.has(issue))); + const linkedSiblingNumbers = new Set(linkedSiblings.map((sibling) => sibling.number)); + const fallbackSiblings = linkedSiblingNumbers.size === 0 ? otherOpenPullRequests : otherOpenPullRequests.filter((sibling) => !linkedSiblingNumbers.has(sibling.number)); + const siblings = [...linkedSiblings, ...fallbackSiblings].slice(0, MERGE_WAKE_MAX_PRS); for (const [index, sibling] of siblings.entries()) { const job: JobMessage = { type: "agent-regate-pr", @@ -6521,6 +6541,7 @@ async function handlePullRequestWebhookEvent( installationId, settings, otherOpenPullRequests, + pr.linkedIssues, ).catch((error) => { /* v8 ignore next -- best-effort: a sibling re-gate enqueue failure is logged, never surfaced to the gate. */ console.error( diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 9810d93d69..7d823bf2f9 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -4062,6 +4062,110 @@ describe("queue processors", () => { ); }); + it("sibling re-gate fan-out (#4005 / #7438-incident): a sibling sharing the merged PR's linked issue is woken FIRST, even ahead of lower-numbered unrelated siblings", async () => { + const sent: Array<{ message: import("../../src/types").JobMessage; options?: QueueSendOptions }> = []; + const env = createTestEnv({ + GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem(), + LOOPOVER_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" }, gatePack: "oss-anti-slop" }); + await upsertRepoFocusManifest(env, "owner/agent-repo", { settings: { checkRunMode: "off", commentMode: "off", publicSurface: "off", aiReviewMode: "off", reviewCheckMode: "required" } }); + // #10 is lower-numbered but shares no linked issue with the merged PR -- under the OLD ascending-by-number + // order it would win the one available slot. #999 is higher-numbered but closes the SAME issue (#500) the + // merged PR itself closes -- a real duplicate-cluster sibling, exactly the #7438-incident shape. + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 10, title: "Unrelated sibling", state: "open", user: { login: "contributor" }, head: { sha: "sib10" }, labels: [], body: "No linked issue.", created_at: "2026-07-01T00:00:00.000Z" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 999, title: "Duplicate-cluster sibling", state: "open", user: { login: "contributor" }, head: { sha: "sib999" }, labels: [], body: "Closes #500", created_at: "2026-07-02T00: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-linked-priority", + 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: 1000, title: "Merged PR", state: "closed", merged_at: "2026-07-08T00:00:00.000Z", user: { login: "contributor" }, head: { sha: "mergedsha" }, labels: [], body: "Closes #500" }, + } as never, + }); + + const regateJobs = sent.filter(({ message }) => message.type === "agent-regate-pr"); + // #999 (linked) is dispatched immediately (index 0, no delay); #10 (unrelated fallback) is pushed to index 1. + expect(regateJobs.map(({ message }) => (message as { prNumber: number }).prNumber)).toEqual([999, 10]); + expect(regateJobs.map(({ options }) => options)).toEqual([undefined, { delaySeconds: 10 }]); + }); + + it("sibling re-gate fan-out (#4005 / #7438-incident): every linked-issue sibling survives the cap, displacing the highest-numbered fallback siblings first", async () => { + const sent: Array<{ message: import("../../src/types").JobMessage; options?: QueueSendOptions }> = []; + const env = createTestEnv({ + GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem(), + LOOPOVER_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" }, gatePack: "oss-anti-slop" }); + await upsertRepoFocusManifest(env, "owner/agent-repo", { settings: { checkRunMode: "off", commentMode: "off", publicSurface: "off", aiReviewMode: "off", reviewCheckMode: "required" } }); + // MERGE_WAKE_MAX_PRS lower-numbered, unrelated open siblings -- on their own, this is exactly the pre-existing + // "capped" scenario above and would fill every available slot. + for (let number = 1; number <= MERGE_WAKE_MAX_PRS; number += 1) { + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number, title: `Unrelated sibling ${number}`, state: "open", user: { login: "contributor" }, head: { sha: `sib${number}` }, labels: [], body: "No linked issue." }); + } + // Two HIGHER-numbered siblings that share the merged PR's linked issue -- under the old order these would + // never be reached at all (positions 100/101 of a list capped at MERGE_WAKE_MAX_PRS). + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 100, title: "Duplicate-cluster sibling A", state: "open", user: { login: "contributor" }, head: { sha: "sib100" }, labels: [], body: "Closes #500" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 101, title: "Duplicate-cluster sibling B", state: "open", user: { login: "contributor" }, head: { sha: "sib101" }, labels: [], body: "Closes #500" }); + 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-linked-priority-capped", + 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: 1000, title: "Merged PR", state: "closed", merged_at: "2026-07-08T00:00:00.000Z", user: { login: "contributor" }, head: { sha: "mergedsha" }, labels: [], body: "Closes #500" }, + } as never, + }); + + const regateJobs = sent.filter(({ message }) => message.type === "agent-regate-pr"); + expect(regateJobs).toHaveLength(MERGE_WAKE_MAX_PRS); + // Both linked siblings (100, 101) come first, regardless of number; the fallback tier fills the remaining + // MERGE_WAKE_MAX_PRS - 2 slots with the LOWEST-numbered unrelated siblings (1..MERGE_WAKE_MAX_PRS - 2) -- + // #14 and #15 are the ones displaced out of the cap. + expect(regateJobs.map(({ message }) => (message as { prNumber: number }).prNumber)).toEqual([ + 100, + 101, + ...Array.from({ length: MERGE_WAKE_MAX_PRS - 2 }, (_, 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