diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 58265fc8f4..8a13e89df4 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -762,23 +762,39 @@ async function fanOutAgentRegateSweepJobs( }); return; } - const repositories = await listRepositories(env); + // Sweep every REVIEW-ACTIVE repo (#sweep-all-modes): the convergence allowlist (GITTENSORY_REVIEW_REPOS) UNION the + // webhook-registered repos, deduped case-insensitively. A repo is swept when it is review-active (allowlisted) OR + // has acting autonomy — so ADVISORY repos (autonomy=observe) are re-gated and get fresh reviews too, not only repos + // that can merge/close. The action layer (maybeRunAgentMaintenance) stays autonomy-gated, so an observe repo is + // re-reviewed but never auto-actioned. This is what makes advisory reviews fire on existing open PRs without + // depending on a fresh webhook per PR. + const byKey = new Map(); + for (const repo of await listRepositories(env)) + byKey.set(repo.fullName.toLowerCase(), repo.fullName); + for (const fullName of listConvergenceRepos(env)) + byKey.set(fullName.toLowerCase(), fullName); const configured: string[] = []; let skippedDraining = 0; - for (const repo of repositories) { - const settings = await resolveRepositorySettings(env, repo.fullName); - if (!isAgentConfigured(settings.autonomy)) continue; + for (const repoFullName of byKey.values()) { + const settings = await resolveRepositorySettings(env, repoFullName); + if ( + !( + isConvergenceRepoAllowed(env, repoFullName) || + isAgentConfigured(settings.autonomy) + ) + ) + continue; // In-flight guard (#audit-sweep-fanout): skip a repo whose prior sweep is still draining — its per-PR jobs are // mid-flight and stamping last_regated_at as they run, so the freshest stamp being within the sweep window // means a sweep is active. Re-arming now would enqueue duplicate per-PR jobs for the not-yet-drained // candidates, so this is what finally stops the 2-min cron piling a second full sweep on an unfinished one. if ( - isRegateSweepDraining(await getLatestRegatedAt(env, repo.fullName), now) + isRegateSweepDraining(await getLatestRegatedAt(env, repoFullName), now) ) { skippedDraining += 1; continue; } - configured.push(repo.fullName); + configured.push(repoFullName); } await Promise.all( configured.map((repoFullName, index) => { @@ -925,8 +941,16 @@ async function sweepRepoRegate( ): Promise { if (!repoFullName) return; const settings = await resolveRepositorySettings(env, repoFullName); - // Defensive: a repo can lose its acting autonomy between fan-out and processing. - if (!isAgentConfigured(settings.autonomy)) return; + // Defensive re-check between fan-out and processing (#sweep-all-modes): the repo must still be review-active + // (allowlisted) OR have acting autonomy. Advisory/observe repos pass here and are re-reviewed; the action layer + // stays autonomy-gated, so they are never auto-actioned. + if ( + !( + isConvergenceRepoAllowed(env, repoFullName) || + isAgentConfigured(settings.autonomy) + ) + ) + return; const mode = resolveAgentActionMode({ globalPaused: isGlobalAgentPause(env) || (await isGlobalAgentFrozen(env)), // env brake OR DB kill-switch (#audit-§5.2) agentPaused: settings.agentPaused, @@ -1112,7 +1136,10 @@ async function regatePullRequest( repoFullName, prNumber, undefined, - { skipAiReview: settings.aiReviewMode !== "block" }, + // Run the AI review on the sweep for BOTH advisory and block modes (#sweep-all-modes) — only skip when AI is + // OFF. The #1462 per-(repo,pr,headSha,mode) cache bounds the cost: an unchanged PR re-gates from cache with no + // re-spend, so an advisory PR gets a posted review without burning a token every sweep tick. + { skipAiReview: settings.aiReviewMode === "off" }, ).catch((error) => { console.error( JSON.stringify({ diff --git a/src/selfhost/migrate.ts b/src/selfhost/migrate.ts index b1fc898427..c2b3576c47 100644 --- a/src/selfhost/migrate.ts +++ b/src/selfhost/migrate.ts @@ -4,6 +4,7 @@ // new ones (idempotent), mirroring wrangler's migration ledger. import { readdirSync, readFileSync } from "node:fs"; import { join } from "node:path"; +import { errorMessage } from "../utils/json"; export async function runSelfHostMigrations(db: D1Database, dir: string): Promise { await db.exec("CREATE TABLE IF NOT EXISTS _selfhost_migrations (name TEXT PRIMARY KEY, applied_at TEXT NOT NULL)"); @@ -13,7 +14,16 @@ export async function runSelfHostMigrations(db: D1Database, dir: string): Promis let count = 0; for (const file of files) { if (applied.has(file)) continue; - await db.exec(readFileSync(join(dir, file), "utf8")); + try { + await db.exec(readFileSync(join(dir, file), "utf8")); + } catch (error) { + // Idempotency (#migrate-drift): a renumbered/duplicated migration whose schema change is ALREADY present (e.g. a + // column added under an earlier filename by a prior deploy, then renumbered before merge) must not crash-loop the + // boot. "duplicate column" / "already exists" means the target schema is satisfied — record the file applied and + // continue. Any OTHER error is a real failure and still aborts the boot. + if (!/duplicate column|already exists/i.test(errorMessage(error))) + throw error; + } await db.prepare("INSERT INTO _selfhost_migrations (name, applied_at) VALUES (?, ?)").bind(file, new Date().toISOString()).run(); count += 1; } diff --git a/src/server.ts b/src/server.ts index 189d62c085..a94fe0ec0f 100644 --- a/src/server.ts +++ b/src/server.ts @@ -250,7 +250,25 @@ async function main(): Promise { // arrives, by which point env is set). let env: Env; const consume = async (message: JobMessage): Promise => { - await processJob(env, message); + try { + await processJob(env, message); + } catch (error) { + // Self-host best-effort jobs (#registry-soft-fail): the periodic gittensor-registry refresh re-runs every cron + // tick, so a degraded/unconfigured GITTENSOR_REGISTRY_URL would otherwise retry→dead-letter EVERY cycle and + // flood the dead-letter alert. Swallow its failure here (the next scheduled tick is the retry); keep the last + // snapshot. The Cloudflare Worker path (src/index.ts) is untouched, so its rate-limit-aware retry is preserved. + if (message.type === "refresh-registry") { + console.warn( + JSON.stringify({ + level: "warn", + event: "refresh_registry_soft_fail", + error: error instanceof Error ? error.message : String(error), + }), + ); + return; + } + throw error; + } }; const databaseUrl = process.env.DATABASE_URL; diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 0447fea8e3..f194e140fd 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -536,9 +536,10 @@ describe("queue processors", () => { ]); }); - it("agent re-gate sweep fans out only to repos that opted the agent in (#777)", async () => { + it("agent re-gate sweep fans out to acting-autonomy repos (#777), skipping non-acting ones when not allowlisted", async () => { const sent: import("../../src/types").JobMessage[] = []; const env = createTestEnv({ + GITTENSORY_REVIEW_REPOS: "", // isolate the acting-autonomy gate from the allowlist-sweep path (tested below) JOBS: { async send(message: import("../../src/types").JobMessage) { sent.push(message); @@ -565,6 +566,22 @@ describe("queue processors", () => { expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 2, requestedBy: "schedule" }); }); + it("agent re-gate sweep ALSO fans out to allowlisted repos regardless of autonomy mode (#sweep-all-modes)", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ GITTENSORY_REVIEW_REPOS: "owner/advisory-repo", JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + // advisory-repo is allowlisted but autonomy is observe (NOT acting) — it must STILL be swept so advisory reviews fire. + await upsertRepositoryFromGitHub(env, { name: "advisory-repo", full_name: "owner/advisory-repo", private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/advisory-repo", autonomy: { merge: "observe", close: "observe" } }); + // off-repo is neither allowlisted nor acting → still skipped. + await upsertRepositoryFromGitHub(env, { name: "off-repo", full_name: "owner/off-repo", private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/off-repo", autonomy: { review: "observe" } }); + + await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" }); + + const swept = sent.filter((m): m is Extract => m.type === "agent-regate-sweep").map((m) => m.repoFullName); + expect(swept).toEqual(["owner/advisory-repo"]); // allowlisted observe repo IS swept; off-repo is not + }); + it("agent re-gate sweep recomputes stale open PR verdicts as an advisory audit, never publishing (#777)", async () => { const sent: import("../../src/types").JobMessage[] = []; const env = createTestEnv({ @@ -979,7 +996,7 @@ describe("queue processors", () => { it("INVARIANT (in-flight guard): the fan-out SKIPS a repo whose prior sweep is still draining, enqueues an idle one (#audit-sweep-fanout)", 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 }); + 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: 9101, 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" } }, 9101); diff --git a/test/unit/selfhost-migrate.test.ts b/test/unit/selfhost-migrate.test.ts index bf0215e68d..cd34c5ed20 100644 --- a/test/unit/selfhost-migrate.test.ts +++ b/test/unit/selfhost-migrate.test.ts @@ -19,4 +19,21 @@ describe("runSelfHostMigrations (#980)", () => { writeFileSync(join(dir, "0003_c.sql"), "CREATE TABLE c (id INTEGER);"); expect(await runSelfHostMigrations(db, dir)).toBe(1); // only the new one }); + + it("tolerates a migration whose schema change is already present (column drift), but rethrows real errors (#migrate-drift)", async () => { + // 0001 adds column x; 0002 re-adds the SAME column under a new filename (a renumbered-migration collision, as + // happened with ai_review_all_authors 0071→0075). "duplicate column" must be tolerated — recorded applied, not + // crash-looping the boot. + const dir = mkdtempSync(join(tmpdir(), "gtmig-")); + writeFileSync(join(dir, "0001_add_x.sql"), "CREATE TABLE t (id INTEGER); ALTER TABLE t ADD COLUMN x INTEGER;"); + writeFileSync(join(dir, "0002_readd_x.sql"), "ALTER TABLE t ADD COLUMN x INTEGER;"); + const db = createD1Adapter(nodeSqliteDriver(new DatabaseSync(":memory:") as never)); + expect(await runSelfHostMigrations(db, dir)).toBe(2); // both recorded; the duplicate-column 0002 is tolerated + + // A genuine error (invalid SQL, not a duplicate/exists) still aborts the boot. + const dir2 = mkdtempSync(join(tmpdir(), "gtmig-")); + writeFileSync(join(dir2, "0001_bad.sql"), "THIS IS NOT VALID SQL;"); + const db2 = createD1Adapter(nodeSqliteDriver(new DatabaseSync(":memory:") as never)); + await expect(runSelfHostMigrations(db2, dir2)).rejects.toThrow(); + }); });