diff --git a/migrations/0134_pr_last_backlog_convergence_regated_at.sql b/migrations/0134_pr_last_backlog_convergence_regated_at.sql new file mode 100644 index 0000000000..8292247f7c --- /dev/null +++ b/migrations/0134_pr_last_backlog_convergence_regated_at.sql @@ -0,0 +1,16 @@ +-- Per-repo draining guard for backlog-convergence-sweep (#4502), mirroring last_regated_at (0062) but scoped +-- to THIS sweep specifically, not shared with agent-regate-sweep's own marker. +-- +-- BEFORE: fanOutBacklogConvergenceSweepJobs has no in-flight guard at all -- while a prior cycle's fanned-out +-- agent-regate-pr jobs (deliveryId "backlog-convergence:...") are still mid-flight, the next 30-min cron tick +-- re-selects the same candidates and re-enqueues duplicate jobs. A crashed/restarted worker leaves its claimed +-- row status:"processing" until queueProcessingTimeoutMs() reclaims it (default 30 min) -- almost exactly this +-- sweep's own cadence -- so a stuck row from one cycle does not stop the next cycle from firing. +-- +-- AFTER: sweepRepoBacklogConvergence stamps this marker for every candidate AT DISPATCH time (mirroring +-- #audit-sweep-dispatch-stamp), and fanOutBacklogConvergenceSweepJobs's resolution loop skips a repo whose +-- freshest stamp is within the sweep's own draining window (isRegateSweepDraining, BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS). +-- +-- gittensory-computed (sweep-written), keyed to the PR, omitted from upsertPullRequestFromGitHub's SET clause so +-- a later GitHub sync cannot clobber it. Nullable / no default -> backward-compatible. +ALTER TABLE pull_requests ADD COLUMN last_backlog_convergence_regated_at TEXT; diff --git a/migrations/0135_backlog_convergence_fanout_marker.sql b/migrations/0135_backlog_convergence_fanout_marker.sql new file mode 100644 index 0000000000..fdfb8697a5 --- /dev/null +++ b/migrations/0135_backlog_convergence_fanout_marker.sql @@ -0,0 +1,14 @@ +-- Fan-out dedup marker for backlog-convergence-sweep (#4502), mirroring last_regate_fanout_at (0063) with a +-- DISTINCT column so the two sweeps' dedup windows never interfere with each other. +-- +-- BEFORE: fanOutBacklogConvergenceSweepJobs has no atomic dedup -- a burst of fan-out trigger jobs (a +-- deploy-restart cron catch-up, or triggers queued behind a backlogged consumer and drained together) each +-- re-enumerate eligible repos and re-enqueue duplicate per-repo sweeps before the (0134) dispatch-stamp +-- in-flight guard can engage. +-- +-- AFTER: claimBacklogConvergenceFanoutSlot performs an atomic conditional UPDATE on this singleton column, +-- mirroring claimRegateFanoutSlot (0063) -- D1 serializes writes, so only ONE concurrent fan-out's UPDATE +-- matches the "unset or older than the dedup window" predicate and proceeds. +-- +-- Reuses the global_agent_controls singleton (0059); nullable / no default -> backward-compatible. +ALTER TABLE global_agent_controls ADD COLUMN last_backlog_convergence_fanout_at TEXT; diff --git a/src/db/repositories.ts b/src/db/repositories.ts index ca50c258b7..e49ef9b902 100644 --- a/src/db/repositories.ts +++ b/src/db/repositories.ts @@ -2542,6 +2542,23 @@ export async function claimRegateFanoutSlot(env: Env, now: string, windowMs: num } } +/** Atomic backlog-convergence-sweep fan-out dedup (#4502), mirroring {@link claimRegateFanoutSlot} exactly but on + * a DISTINCT singleton column so the two differently-cadenced sweeps' dedup windows never interfere. */ +export async function claimBacklogConvergenceFanoutSlot(env: Env, now: string, windowMs: number): Promise { + const threshold = new Date(Date.parse(now) - windowMs).toISOString(); + try { + const result = await env.DB.prepare( + "UPDATE global_agent_controls SET last_backlog_convergence_fanout_at = ?1 WHERE id = 'singleton' AND (last_backlog_convergence_fanout_at IS NULL OR last_backlog_convergence_fanout_at < ?2)", + ) + .bind(now, threshold) + .run(); + /* v8 ignore next -- D1 update metadata normally includes changes; the ?? 0 fallback protects driver anomalies. */ + return Number(result.meta.changes ?? 0) === 1; + } catch { + return true; + } +} + /** Atomic per-period dedup for the cross-repo maintainer recap digest (#2249): claim `periodKey` (the current * UTC date, "YYYY-MM-DD") as the singleton's last-sent period. Mirrors {@link claimRegateFanoutSlot}: the * conditional UPDATE matches only when the stored period is unset or DIFFERENT from `periodKey`, so a retried @@ -3779,6 +3796,20 @@ export async function markPullRequestsRegated(env: Env, fullName: string, number .where(and(eq(pullRequests.repoFullName, fullName), inArray(pullRequests.number, numbers))); } +/** Batch variant of {@link markPullRequestsRegated} for backlog-convergence-sweep (#4502): stamps the SEPARATE + * last_backlog_convergence_regated_at marker at sweep DISPATCH time, mirroring the same "stamp immediately, not + * in the downstream per-PR job" shape so getLatestBacklogConvergenceRegatedAt reflects this sweep before its + * staggered per-PR jobs complete. */ +export async function markPullRequestsBacklogConvergenceRegated(env: Env, fullName: string, numbers: number[]): Promise { + if (numbers.length === 0) return; + const db = getDb(env.DB); + const now = nowIso(); + await db + .update(pullRequests) + .set({ lastBacklogConvergenceRegatedAt: now, updatedAt: now }) + .where(and(eq(pullRequests.repoFullName, fullName), inArray(pullRequests.number, numbers))); +} + /** In-flight guard input for the re-gate sweep fan-out (#audit-sweep-fanout): the MOST RECENT last_regated_at * across a repo's OPEN PRs (the freshest sweep stamp), or null if none has been swept. fanOutAgentRegateSweepJobs * passes this to isRegateSweepDraining to skip re-arming a repo whose prior sweep is still draining. */ @@ -3793,6 +3824,19 @@ export async function getLatestRegatedAt(env: Env, fullName: string): Promise { + const db = getDb(env.DB); + const [row] = await db + .select({ latest: sql`max(${pullRequests.lastBacklogConvergenceRegatedAt})` }) + .from(pullRequests) + .where(and(eq(pullRequests.repoFullName, fullName), eq(pullRequests.state, "open"))); + /* v8 ignore next -- max() always returns exactly one row; the empty-array guard only satisfies the destructure type. */ + if (!row) return null; + return row.latest; +} + export async function getIssue(env: Env, fullName: string, number: number): Promise { const db = getDb(env.DB); const [row] = await db.select().from(issues).where(and(eq(issues.repoFullName, fullName), eq(issues.number, number))).limit(1); diff --git a/src/db/schema.ts b/src/db/schema.ts index 4ed14a1a0f..76d448f004 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -470,6 +470,11 @@ export const pullRequests = sqliteTable( // review WRITE that would bump updated_at is suppressed (dry-run / paused). gittensory-computed (sweep-written), // omitted from the GitHub-sync SET clause so a later sync cannot clobber it. (Mirrors approved_head_sha.) lastRegatedAt: text("last_regated_at"), + // Draining guard for backlog-convergence-sweep (#4502), mirroring lastRegatedAt but scoped to THIS sweep -- + // stamped at dispatch by sweepRepoBacklogConvergence, read by fanOutBacklogConvergenceSweepJobs to skip a + // repo whose prior fan-out is still draining. Kept separate from lastRegatedAt so the two differently-cadenced + // sweeps' in-flight signals never conflate. gittensory-computed, omitted from the GitHub-sync SET clause. + lastBacklogConvergenceRegatedAt: text("last_backlog_convergence_regated_at"), // Public-surface marker: the head SHA at which the public surface (comment/label/check-run) was LAST published. // Used for reporting and stale-surface diagnostics, not as a hard sweep skip; GitHub comments/checks can still // be stale or partial while this marker matches headSha. gittensory-computed (publish-written), omitted from diff --git a/src/index.ts b/src/index.ts index 09a112b46e..e2457f5c22 100644 --- a/src/index.ts +++ b/src/index.ts @@ -28,6 +28,12 @@ const app = createApp(); // per-repo drain guard (getLatestRegatedAt / isRegateSweepDraining) already protects individual repos once that // single fan-out runs, so this only needs to stop a SECOND trigger from queuing up behind the first. const REGATE_SWEEP_TRIGGER_TYPES = ["agent-regate-sweep"] as const; +// Same shape as REGATE_SWEEP_TRIGGER_TYPES, scoped to backlog-convergence-sweep's own top-level trigger (#4502): +// its per-repo draining guard (getLatestBacklogConvergenceRegatedAt / isRegateSweepDraining) already protects +// individual repos once a fan-out runs, so this only needs to stop a SECOND trigger queuing up behind the first +// — the gap that let a crashed/restarted worker's stuck "processing" trigger row (reclaimed only after +// queueProcessingTimeoutMs(), which defaults to this sweep's own 30-min cadence) go unnoticed by the next tick. +const BACKLOG_CONVERGENCE_SWEEP_TRIGGER_TYPES = ["backlog-convergence-sweep"] as const; export { RateLimiter }; @@ -175,7 +181,17 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): // jobs above: it is a backstop for a rare stranding, not the primary convergence path, so it does not need the // sweep's ~2-min cadence. Self-host only (mirrors "agent-regate-sweep") — the trigger job itself is maintenance- // classified (MAINTENANCE_JOB_TYPES) so it defers under live-work pressure like every other periodic sweep here. - if (selfHostedReviews) jobs.push({ type: "backlog-convergence-sweep", requestedBy: "schedule" }); + if (selfHostedReviews) { + const backlogConvergenceTriggerBacklog = queueSnapshotBacklog(queueSnapshot, BACKLOG_CONVERGENCE_SWEEP_TRIGGER_TYPES); + if (backlogConvergenceTriggerBacklog > 0) { + // A fan-out trigger is already pending/processing (#4502) — skip re-arming so a crashed/restarted worker's + // stuck row (reclaimed only after queueProcessingTimeoutMs(), which coincides with this sweep's own 30-min + // cadence) cannot go unnoticed by the next tick and duplicate per-repo/per-PR work underneath it. + console.log(JSON.stringify({ event: "backlog_convergence_sweep_trigger_backlog_deferred", backlog: backlogConvergenceTriggerBacklog })); + } else { + jobs.push({ type: "backlog-convergence-sweep", requestedBy: "schedule" }); + } + } } // Self-heal (flag GITTENSORY_PR_RECONCILIATION). Every 10 minutes — see isReconciliationWindow above. // Enqueued ONLY when the flag is ON — flag-OFF (default) this job is never created, so the cron tick does diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 0579504d04..a4e82c9512 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -54,11 +54,14 @@ import { getCachedLinkedIssueSatisfaction, putCachedLinkedIssueSatisfaction, markPullRequestsRegated, + markPullRequestsBacklogConvergenceRegated, markPullRequestReviewsInvalidated, markPullRequestSurfacePublished, markPullRequestVisualCaptureSatisfied, getLatestRegatedAt, + getLatestBacklogConvergenceRegatedAt, claimRegateFanoutSlot, + claimBacklogConvergenceFanoutSlot, recordAgentCommandFeedback, recordAuditEvent, countRecentAuditEventsForActorAndTarget, @@ -265,6 +268,7 @@ import { ISSUE_WAKE_MAX_PRS, MERGE_WAKE_MAX_PRS, SWEEP_FANOUT_DEDUP_MS, + BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS, SWEEP_MAX_PRS, isRegateSweepDraining, selectRegateCandidates, @@ -2076,15 +2080,26 @@ async function sweepRepoRegate( // #selfhost-backlog-convergence: the cron (index.ts) enqueues one fan-out trigger periodically; this enqueues a // per-repo sweep job for every repo eligible for convergence (the SAME repo selection as the re-gate sweep, so -// a repo that opted the agent in — or is explicitly convergence-allowlisted — gets both). Deliberately has no -// fan-out dedup CAS (contrast fanOutAgentRegateSweepJobs): unlike that sweep, this one stamps nothing -// optimistically, so a second overlapping trigger just re-reads current state and re-enqueues, which coalesces -// harmlessly into the same pending agent-regate-pr rows (queue-common.ts's job_key coalescing) rather than -// duplicating work. +// a repo that opted the agent in — or is explicitly convergence-allowlisted — gets both). #4502: now mirrors +// fanOutAgentRegateSweepJobs's three-layer anti-duplication shape exactly — an atomic fan-out-slot claim +// (claimBacklogConvergenceFanoutSlot) collapses a BURST of this trigger, and the per-repo resolution below skips +// any repo whose prior fan-out is still draining (getLatestBacklogConvergenceRegatedAt / isRegateSweepDraining) — +// closing the gap where a crashed/restarted worker's stuck "processing" trigger row went unnoticed by the next +// 30-min tick and re-enqueued duplicate per-repo (and per-PR) jobs underneath the still-in-flight one. async function fanOutBacklogConvergenceSweepJobs( env: Env, requestedBy: "schedule" | "api" | "test", ): Promise { + const now = nowIso(); + if (!(await claimBacklogConvergenceFanoutSlot(env, now, SWEEP_FANOUT_DEDUP_MS))) { + await recordAuditEvent(env, { + eventType: "agent.sweep.backlog_convergence.fanout", + outcome: "denied", + detail: "backlog-convergence fan-out deduped: another fan-out already claimed this window", + metadata: { requestedBy, deduped: true }, + }); + return; + } const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo])); const byKey = new Map(); for (const repo of repositoriesByKey.values()) @@ -2096,12 +2111,39 @@ async function fanOutBacklogConvergenceSweepJobs( ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}), }); } + // #4502 (ports #3899): resolve every repo's settings + drain-state CONCURRENTLY (bounded), not one at a time — + // mirrors fanOutAgentRegateSweepJobs's own port of this fix, the same "many small per-repo D1/KV reads" shape. + const outcomes = await mapWithConcurrencyLimit( + [...byKey.values()], + SWEEP_FANOUT_RESOLUTION_CONCURRENCY, + async (repo): Promise => { + const repoFullName = repo.fullName; + try { + const settings = await resolveRepositorySettings(env, repoFullName); + if (!(isConvergenceRepoAllowed(env, repoFullName) || isAgentConfigured(settings.autonomy))) return { kind: "ineligible" }; + if (isRegateSweepDraining(await getLatestBacklogConvergenceRegatedAt(env, repoFullName), now, BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS)) + return { kind: "draining" }; + return { kind: "configured", repo }; + } catch (error) { + console.error( + JSON.stringify({ + level: "error", + event: "backlog_convergence_fanout_repo_check_failed", + repository: repoFullName, + error: errorMessage(error), + }), + ); + return { kind: "errored" }; + } + }, + ); const configured: Array<{ fullName: string; installationId?: number }> = []; - for (const repo of byKey.values()) { - const settings = await resolveRepositorySettings(env, repo.fullName); - if (isConvergenceRepoAllowed(env, repo.fullName) || isAgentConfigured(settings.autonomy)) { - configured.push(repo); - } + let skippedDraining = 0; + let skippedErrored = 0; + for (const outcome of outcomes) { + if (outcome.kind === "configured") configured.push(outcome.repo); + else if (outcome.kind === "draining") skippedDraining += 1; + else if (outcome.kind === "errored") skippedErrored += 1; } await Promise.all( configured.map((repo, index) => { @@ -2112,15 +2154,25 @@ async function fanOutBacklogConvergenceSweepJobs( ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}), }; const delaySeconds = Math.min(index * 10, 600); - return delaySeconds > 0 - ? env.JOBS.send(message, { delaySeconds }) - : env.JOBS.send(message); + const send = delaySeconds > 0 ? env.JOBS.send(message, { delaySeconds }) : env.JOBS.send(message); + // #audit-sweep-fanout-isolation (mirrors fanOutAgentRegateSweepJobs): one repo's dispatch failure must not + // reject this Promise.all and abort every OTHER repo's already-in-flight send. + return send.catch((error) => { + console.error( + JSON.stringify({ + level: "error", + event: "backlog_convergence_fanout_dispatch_failed", + repository: repo.fullName, + error: errorMessage(error), + }), + ); + }); }), ); await recordAuditEvent(env, { eventType: "agent.sweep.backlog_convergence.fanout", outcome: "queued", - metadata: { repoCount: configured.length, requestedBy }, + metadata: { repoCount: configured.length, skippedDraining, skippedErrored, requestedBy }, }); } @@ -2215,6 +2267,24 @@ async function sweepRepoBacklogConvergence( const openPullRequests = await listOpenPullRequests(env, repoFullName); const candidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests }); if (candidates.length === 0) return; + // Stamp the backlog-convergence draining marker for EVERY candidate NOW, at dispatch — not in the downstream + // per-PR job (#4502, mirrors #audit-sweep-dispatch-stamp). This makes getLatestBacklogConvergenceRegatedAt + // reflect this sweep immediately, so fanOutBacklogConvergenceSweepJobs's in-flight guard skips re-arming this + // repo on the next cron tick BEFORE the staggered per-PR re-reviews finish. A plain D1 write → dry-run stays inert. + await markPullRequestsBacklogConvergenceRegated( + env, + repoFullName, + candidates.map((pr) => pr.number), + ).catch((error) => { + console.error( + JSON.stringify({ + level: "warn", + event: "backlog_convergence_mark_regated_failed", + repository: repoFullName, + error: errorMessage(error), + }), + ); + }); await Promise.all( candidates.map((pr, index) => { const job: JobMessage = { diff --git a/src/settings/agent-sweep.ts b/src/settings/agent-sweep.ts index e4f1195677..9144ce9631 100644 --- a/src/settings/agent-sweep.ts +++ b/src/settings/agent-sweep.ts @@ -49,6 +49,12 @@ export const SWEEP_FRESHNESS_MS = 2 * 60 * 1000; // behind a per-PR backlog and drained together). export const SWEEP_FANOUT_DEDUP_MS = 90 * 1000; +// Draining window for backlog-convergence-sweep (#4502), the isRegateSweepDraining windowMs for THIS sweep +// specifically -- distinct from SWEEP_FRESHNESS_MS because this sweep runs every ~30 min (not ~2 min), so its +// own per-PR dispatch batch can legitimately still be draining minutes after fan-out. Sized to roughly match its +// own cron cadence, mirroring how SWEEP_FRESHNESS_MS is sized to the ~2-min regate-sweep cadence. +export const BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS = 30 * 60 * 1000; + // Candidate ordering mode (#3815, RepositorySettings["regateSweepOrderMode"]). "staleness" (default) is // selectRegateCandidates' original ordering; "oldest-first" is opt-in per repo. See the function doc comment // for the convergence-guarantee rationale each preserves. diff --git a/test/unit/db-parsers.test.ts b/test/unit/db-parsers.test.ts index 52072d9356..694f4336b5 100644 --- a/test/unit/db-parsers.test.ts +++ b/test/unit/db-parsers.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { claimMaintainerRecapPeriod, claimRegateFanoutSlot, + claimBacklogConvergenceFanoutSlot, countRecentDeadLetters, countRecentDeadLettersByType, countRecentAuditEventsForActorAndTarget, @@ -21,6 +22,7 @@ import { listRepoSyncStates, markPullRequestRegated, markPullRequestsRegated, + markPullRequestsBacklogConvergenceRegated, markPullRequestSurfacePublished, recordAuditEvent, recordWebhookEvent, @@ -350,6 +352,25 @@ describe("database row parser hardening", () => { expect(rows.find((p) => p.number === 6)?.lastRegatedAt ?? null).toBeNull(); // #6 not in the batch → untouched }); + it("markPullRequestsBacklogConvergenceRegated batch-stamps every candidate at dispatch and no-ops on an empty list (#4502)", async () => { + const env = createTestEnv(); + for (const number of [5, 6, 7]) { + await upsertPullRequestFromGitHub(env, "owner/repo", { number, title: `PR${number}`, state: "open", user: { login: "alice" }, labels: [] }); + } + const stampedAt = async (number: number) => + (await env.DB.prepare("select last_backlog_convergence_regated_at as v from pull_requests where repo_full_name = ? and number = ?").bind("owner/repo", number).first<{ v: string | null }>())?.v ?? null; + + await markPullRequestsBacklogConvergenceRegated(env, "owner/repo", []); // empty → no-op (early return) + expect(await stampedAt(5)).toBeNull(); + expect(await stampedAt(6)).toBeNull(); + expect(await stampedAt(7)).toBeNull(); + + await markPullRequestsBacklogConvergenceRegated(env, "owner/repo", [5, 7]); // batch stamps only 5 and 7 + expect(typeof (await stampedAt(5))).toBe("string"); + expect(typeof (await stampedAt(7))).toBe("string"); + expect(await stampedAt(6)).toBeNull(); // #6 not in the batch → untouched + }); + it("claimRegateFanoutSlot collapses a burst to one winner per window (#audit-fanout-dedup)", async () => { const env = createTestEnv(); const W = 90 * 1000; @@ -394,6 +415,12 @@ describe("database row parser hardening", () => { expect(await claimRegateFanoutSlot(broken, "2026-06-25T01:00:00.000Z", 90 * 1000)).toBe(true); }); + it("claimBacklogConvergenceFanoutSlot fails open (returns true) on a DB error so the fleet never stalls (#4502)", async () => { + const env = createTestEnv(); + const broken = { ...env, DB: null } as unknown as typeof env; + expect(await claimBacklogConvergenceFanoutSlot(broken, "2026-06-25T01:00:00.000Z", 90 * 1000)).toBe(true); + }); + it("claimMaintainerRecapPeriod: first claim for a period wins, a retry for the SAME period loses, a DIFFERENT period wins again (#2249)", async () => { const env = createTestEnv(); expect(await claimMaintainerRecapPeriod(env, "2026-07-09")).toBe(true); // first claim (marker NULL) diff --git a/test/unit/index.test.ts b/test/unit/index.test.ts index c4913bba0b..397a845e1f 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -340,6 +340,34 @@ describe("worker entrypoint", () => { ]); }); + it("INVARIANT (#4502): defers a new backlog-convergence-sweep trigger while a prior one is still pending or processing", async () => { + const sent: Array = []; + const env = createTestEnv({ + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + snapshot: async () => ({ + totals: { pending: 0, processing: 1, dead: 0, due: 0 }, + byType: [{ type: "backlog-convergence-sweep", status: "processing", count: 1, due: 0 }], + }), + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + + await worker.scheduled(controllerFor("2026-05-25T05:30:00.000Z"), env, executionContext(waitUntil)); + await Promise.all(waitUntil); + + // No SECOND "backlog-convergence-sweep" trigger is enqueued behind the one already in flight; the other :30 + // jobs (including agent-regate-sweep, whose OWN backlog is unaffected) still fire normally. + expect(sent).toEqual([ + { type: "agent-regate-sweep", requestedBy: "schedule" }, + { type: "backfill-registered-repos", requestedBy: "schedule", mode: "light" }, + { type: "repair-data-fidelity", requestedBy: "schedule" }, + { type: "refresh-installation-health", requestedBy: "schedule" }, + ]); + }); + it("does not require queue introspection for regular review sweep scheduling", async () => { const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); const sent: Array = []; diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index ce399fe848..ffbd3f6aee 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -31104,6 +31104,185 @@ describe("backlog-convergence sweep (#selfhost-backlog-convergence)", () => { expect(meta).toMatchObject({ repoFullName: "owner/agent-repo", openCount: 4, examined: 3 }); expect(meta.candidatePulls.sort((a: number, b: number) => a - b)).toEqual([7, 8, 10]); }); + + it("REGRESSION (#4502, #audit-sweep-dispatch-stamp): ONE sweep stamps ALL candidates AT DISPATCH, so the next fan-out skips the repo as draining — no overlapping sweeps", 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: 9510, 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" } }, 9510); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" } }); + for (const number of [7, 8, 9]) { + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number, title: `PR${number}`, state: "open", user: { login: "c" }, head: { sha: `a${number}` }, labels: [], body: "" }); + } + + // Run ONE per-repo sweep — do NOT drain the per-PR jobs (simulate the staggered re-reviews not having run yet). + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo" }); + + // The marker is stamped for EVERY candidate immediately at dispatch — NOT waiting on the per-PR jobs. + const stamped = await env.DB.prepare("select count(*) as n from pull_requests where repo_full_name = ? and last_backlog_convergence_regated_at is not null").bind("owner/agent-repo").first<{ n: number }>(); + expect(stamped?.n).toBe(3); + + // So the very next cron fan-out sees the fresh stamp and SKIPS this repo as draining — the overlap that would + // duplicate per-PR jobs is gone. + sent.length = 0; + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); + expect(sent.some((m) => m.type === "backlog-convergence-sweep" && m.repoFullName === "owner/agent-repo")).toBe(false); + const fanout = await env.DB.prepare("select metadata_json from audit_events where event_type = ? order by created_at desc limit 1").bind("agent.sweep.backlog_convergence.fanout").first<{ metadata_json: string }>(); + expect(JSON.parse(fanout?.metadata_json ?? "{}").skippedDraining).toBeGreaterThanOrEqual(1); + }); + + it("INVARIANT (#4502, in-flight guard): the fan-out SKIPS a repo whose prior sweep is still draining, enqueues an idle one", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ GITTENSORY_REVIEW_REPOS: "", JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + await upsertInstallation(env, { action: "created", installation: { id: 9511, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + for (const name of ["draining", "idle"]) { + await upsertRepositoryFromGitHub(env, { name, full_name: `owner/${name}`, private: false, owner: { login: "owner" } }, 9511); + await upsertRepositorySettings(env, { repoFullName: `owner/${name}`, autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, `owner/${name}`, { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "h1" }, labels: [], body: "" }); + } + // owner/draining was just backlog-convergence-regated (a sweep is mid-drain); owner/idle has never been swept. + await repositoriesModule.markPullRequestsBacklogConvergenceRegated(env, "owner/draining", [1]); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); // no repoFullName → fan-out path + + const sweepRepos = sent.filter((m): m is Extract => m.type === "backlog-convergence-sweep").map((m) => m.repoFullName); + expect(sweepRepos).toEqual(["owner/idle"]); // the draining repo is skipped, the idle one enqueued + const fanout = await env.DB.prepare("select metadata_json from audit_events where event_type = ?").bind("agent.sweep.backlog_convergence.fanout").first<{ metadata_json: string }>(); + expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 1, skippedDraining: 1 }); + }); + + it("INVARIANT (#4502, #audit-fanout-dedup): a BURST of fan-outs collapses to ONE — the second claims nothing and audits denied", 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: 9512, 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" } }, 9512); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "PR7", state: "open", user: { login: "c" }, head: { sha: "a7" }, labels: [], body: "" }); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); // first fan-out claims the window + expect(sent.some((m) => m.type === "backlog-convergence-sweep" && m.repoFullName === "owner/agent-repo")).toBe(true); + + sent.length = 0; + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); // burst sibling in the same window → deduped + expect(sent.filter((m) => m.type === "backlog-convergence-sweep")).toEqual([]); // enqueues no redundant sweep + const denied = await env.DB.prepare("select count(*) as n from audit_events where event_type='agent.sweep.backlog_convergence.fanout' and outcome='denied'").first<{ n: number }>(); + expect(denied?.n).toBe(1); + }); + + it("REGRESSION (#4502, #audit-sweep-fanout-isolation): one repo's settings-check failure does not abort the fan-out for every other repo", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ + GITTENSORY_REVIEW_REPOS: "", + JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue, + }); + await upsertRepositoryFromGitHub(env, { name: "agent-a", full_name: "owner/agent-a", private: false, owner: { login: "owner" } }); + await upsertRepositoryFromGitHub(env, { name: "agent-b", full_name: "owner/agent-b", private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-a", autonomy: { label: "auto" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-b", autonomy: { label: "auto" } }); + const realResolve = repositorySettingsModule.resolveRepositorySettings; + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + const resolveSpy = vi.spyOn(repositorySettingsModule, "resolveRepositorySettings").mockImplementation(async (e, repoFullName) => { + if (repoFullName === "owner/agent-a") throw new Error("D1 read error"); + return realResolve(e, repoFullName); + }); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); + + expect(sent).toEqual([expect.objectContaining({ type: "backlog-convergence-sweep", repoFullName: "owner/agent-b" })]); // agent-a's failure did not block agent-b + expect(errors.mock.calls.some((call) => String(call[0]).includes("backlog_convergence_fanout_repo_check_failed") && String(call[0]).includes("owner/agent-a"))).toBe(true); + const fanout = await env.DB.prepare("select outcome, metadata_json from audit_events where event_type = ?").bind("agent.sweep.backlog_convergence.fanout").first<{ outcome: string; metadata_json: string }>(); + expect(fanout?.outcome).toBe("queued"); // the fan-out still completes and records its own outcome + expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 1, skippedErrored: 1 }); + errors.mockRestore(); + resolveSpy.mockRestore(); + }); + + it("REGRESSION (#4502, #audit-sweep-fanout-isolation): one repo's dispatch failure does not abort dispatch for every other repo, and the fan-out audit event still records", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ + GITTENSORY_REVIEW_REPOS: "", + JOBS: { + async send(m: import("../../src/types").JobMessage) { + if (m.type === "backlog-convergence-sweep" && m.repoFullName === "owner/agent-a") throw new Error("queue send error"); + sent.push(m); + }, + } as unknown as Queue, + }); + await upsertRepositoryFromGitHub(env, { name: "agent-a", full_name: "owner/agent-a", private: false, owner: { login: "owner" } }); + await upsertRepositoryFromGitHub(env, { name: "agent-b", full_name: "owner/agent-b", private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-a", autonomy: { label: "auto" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-b", autonomy: { label: "auto" } }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); + + expect(sent).toEqual([expect.objectContaining({ type: "backlog-convergence-sweep", repoFullName: "owner/agent-b" })]); // agent-a's failed send did not block agent-b's + expect(errors.mock.calls.some((call) => String(call[0]).includes("backlog_convergence_fanout_dispatch_failed") && String(call[0]).includes("owner/agent-a"))).toBe(true); + const fanout = await env.DB.prepare("select outcome, metadata_json from audit_events where event_type = ?").bind("agent.sweep.backlog_convergence.fanout").first<{ outcome: string; metadata_json: string }>(); + expect(fanout?.outcome).toBe("queued"); // reached — the dispatch failure did not throw the fan-out itself + expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 2 }); // both PASSED their settings/draining checks regardless of dispatch outcome + errors.mockRestore(); + }); + + it("agent re-gate sweep swallows a failing last_backlog_convergence_regated_at stamp and still completes (#4502, #audit-sweep-converge)", 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: 9513, 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" } }, 9513); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Stale surface", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "" }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + const stamp = vi.spyOn(repositoriesModule, "markPullRequestsBacklogConvergenceRegated").mockRejectedValueOnce(new Error("D1 write error")); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo" }); + + const audit = await env.DB.prepare("select outcome from audit_events where event_type = ?").bind("agent.sweep.backlog_convergence").first<{ outcome: string }>(); + expect(audit?.outcome).toBe("completed"); // the sweep still completes; the dispatch-time stamp failure is swallowed + expect(sent.some((m) => m.type === "agent-regate-pr" && m.prNumber === 7)).toBe(true); // the per-PR fan-out still happens + expect(errors.mock.calls.some((call) => String(call[0]).includes("backlog_convergence_mark_regated_failed"))).toBe(true); + stamp.mockRestore(); + errors.mockRestore(); + }); + + it("REGRESSION (#4502, #3899-style port): resolves multiple repos' settings/drain-state CONCURRENTLY, bounded by SWEEP_FANOUT_RESOLUTION_CONCURRENCY, and drops no repo", async () => { + vi.useRealTimers(); + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ + GITTENSORY_REVIEW_REPOS: "", + JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue, + }); + const repoNames = ["r1", "r2", "r3", "r4", "r5", "r6"]; + for (const name of repoNames) { + await upsertRepositoryFromGitHub(env, { name, full_name: `owner/${name}`, private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: `owner/${name}`, autonomy: { merge: "auto" } }); + } + const { mapWithConcurrencyLimit: realMapWithConcurrencyLimit } = + await vi.importActual("../../src/signals/focus-manifest-loader"); + let inFlight = 0; + let maxInFlight = 0; + const mapSpy = vi.spyOn(focusManifestLoaderModule, "mapWithConcurrencyLimit").mockImplementation( + async (items, limit, mapper) => { + expect(limit).toBe(SWEEP_FANOUT_RESOLUTION_CONCURRENCY); + return realMapWithConcurrencyLimit(items, limit, async (item) => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + try { + await new Promise((resolve) => setTimeout(resolve, 5)); // hold the window open long enough for others to overlap + return await mapper(item); + } finally { + inFlight -= 1; + } + }); + }, + ); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule" }); + + expect(mapSpy).toHaveBeenCalled(); + expect(maxInFlight).toBeGreaterThan(1); // proves real overlap — not the old strictly-sequential loop + expect(maxInFlight).toBeLessThanOrEqual(SWEEP_FANOUT_RESOLUTION_CONCURRENCY); // proves BOUNDED, not unlimited fan-out + expect(sent.filter((m) => m.type === "backlog-convergence-sweep").length).toBe(repoNames.length); // every repo still dispatched, none silently dropped + }); }); // #selfhost-auto-action-convergence: end-to-end regression coverage for the GENERAL heuristic plan+execute path