fix(selfhost): sweep reviews regardless of autonomy mode + registry soft-fail + idempotent migrations

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<string, string>();
+  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<void> {
   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({

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<number> {
   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;
   }

src/server.ts

@@ -250,7 +250,25 @@ async function main(): Promise<void> {
   // arrives, by which point env is set).
   let env: Env;
   const consume = async (message: JobMessage): Promise<void> => {
-    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;

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<import("../../src/types").JobMessage, { type: "agent-regate-sweep" }> => 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);

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();
+  });
 });