JSONbored · JSONbored · Jul 11, 2026 · Jul 11, 2026
@@ -273,6 +273,20 @@ export async function markRepositoriesRemovedFromInstallation(env: Env, installa
     .where(and(eq(repositories.installationId, installationId), inArray(repositories.fullName, names)));
 }
 
+/** Every repo full name currently marked `isInstalled` under this installation, so a caller can diff it against
+ *  a freshly-fetched live list and hand the leftovers to {@link markRepositoriesRemovedFromInstallation} (#5028).
+ *  Unlike listRepoFullNamesForInstallation (used for cross-repo aggregation with a truncation-audit concern),
+ *  this is a plain currently-installed set for a single maintainer's own installation, which is never large
+ *  enough to need a cap. */
+export async function listInstalledRepoFullNamesForInstallation(env: Env, installationId: number): Promise<string[]> {
+  const db = getDb(env.DB);
+  const rows = await db
+    .select({ fullName: repositories.fullName })
+    .from(repositories)
+    .where(and(eq(repositories.installationId, installationId), eq(repositories.isInstalled, true)));
+  return rows.map((row) => row.fullName);
+}
+
 export async function getInstallation(env: Env, installationId: number): Promise<InstallationRecord | null> {
   const db = getDb(env.DB);
   const [row] = await db.select().from(installations).where(eq(installations.id, installationId)).limit(1);

@@ -4,6 +4,7 @@ import { delayUntil, shouldWaitForGitHubRateLimit, LOW_REST_RATE_LIMIT_REMAINING
 import { processDlqBatch } from "./queue/dlq";
 import { processJob } from "./queue/processors";
 import { isOrbBrokerEnabled } from "./orb/broker";
+import { isOrbBrokerMode } from "./orb/broker-client";
 import { isOpsEnabled } from "./review/ops-wire";
 import { isRecapEnabled, resolveMaintainerRecapManifestOverride, shouldFireMaintainerRecap } from "./review/maintainer-recap-wire";
 import { isSweepWatchdogEnabled } from "./review/sweep-watchdog";
@@ -202,6 +203,11 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
   if (selfHostedReviews && isReconciliationWindow && isPrReconciliationEnabled(env)) jobs.push({ type: "reconcile-open-prs", requestedBy: "schedule" });
   if (isHourly) {
     jobs.push({ type: "refresh-registry", requestedBy: "schedule" });
+    // Brokered self-host installed-repo sync (#5028): the central Orb relay deliberately does not forward
+    // installation/installation_repositories events to brokered containers, so a brokered self-host has no
+    // other way to learn its own repo list beyond the first forwarded PR/issue event per repo. Self-host +
+    // broker-mode only (isOrbBrokerMode reads ORB_ENROLLMENT_SECRET) — a no-op everywhere else, byte-identical.
+    if (selfHostedReviews && isOrbBrokerMode(env)) jobs.push({ type: "sync-brokered-installed-repos", requestedBy: "schedule" });
     jobs.push({ type: "refresh-scoring-model", requestedBy: "schedule" });
     jobs.push({ type: "refresh-upstream-drift", requestedBy: "schedule" });
     jobs.push({ type: "rollup-product-usage", requestedBy: "schedule", days: 7 });

@@ -0,0 +1,80 @@
+// Brokered self-host installed-repo sync (#5028, part of the isRegistered/isInstalled untangling epic #5016). A
+// brokered self-host learns its own repo list from GitHub directly, via its broker token -- the same list a
+// non-brokered self-host or cloud installation already gets eagerly through real `installation`/
+// `installation_repositories` webhooks. The central Orb relay deliberately does NOT forward those two events to
+// brokered containers (src/orb/relay.ts's RELAY_FORWARD_EVENTS -- the container runs under the CENTRAL Orb App,
+// not its own, so it must not treat those as its own installation state). Without this sync, a brokered
+// self-host only learns about a repo the FIRST time a forwarded PR/issue event arrives for it: a freshly
+// enrolled, quiet repo has no local `repositories` row at all, and every core feature gated on `isInstalled`
+// silently skips it.
+
+import { listInstalledRepoFullNamesForInstallation, markRepositoriesRemovedFromInstallation, upsertRepositoryFromGitHub } from "../db/repositories";
+import { githubHeaders } from "../github/client";
+import type { GitHubRepositoryPayload } from "../types";
+import { fetchBrokeredInstallationToken, isOrbBrokerMode } from "./broker-client";
+
+const GITHUB_INSTALLATION_REPOS_PAGE_SIZE = 100;
+// Bounds worst-case pagination for a single sync tick. A real maintainer's installation has a handful of repos;
+// this caps runaway pagination against a misbehaving response to a sane worst case (5,000 repos) rather than
+// looping unbounded.
+const MAX_INSTALLATION_REPOS_PAGES = 50;
+
+export type InstalledReposSyncResult =
+  | { status: "skipped" }
+  | { status: "synced"; installationId: number; repoCount: number; removedCount: number }
+  | { status: "failed"; reason: string };
+
+/** Fetch every repo currently accessible to this brokered installation via GitHub's own
+ *  `GET /installation/repositories`, paginated. Uses the broker token directly -- its response already carries
+ *  the bound installationId, so no separate token mint is needed for this call. */
+async function fetchAllInstallationRepos(token: string, fetchImpl: typeof fetch): Promise<GitHubRepositoryPayload[]> {
+  const repos: GitHubRepositoryPayload[] = [];
+  for (let page = 1; page <= MAX_INSTALLATION_REPOS_PAGES; page += 1) {
+    const res = await fetchImpl(`https://api.github.com/installation/repositories?per_page=${GITHUB_INSTALLATION_REPOS_PAGE_SIZE}&page=${page}`, {
+      headers: githubHeaders({ token }),
+      signal: AbortSignal.timeout(20_000),
+    });
+    if (!res.ok) throw new Error(`installation_repositories_http_${res.status}`);
+    const body = (await res.json()) as { repositories?: GitHubRepositoryPayload[] };
+    const batch = body.repositories ?? [];
+    repos.push(...batch);
+    if (batch.length < GITHUB_INSTALLATION_REPOS_PAGE_SIZE) break;
+  }
+  return repos;
+}
+
+/**
+ * Sync this brokered self-host's `repositories.isInstalled` rows against GitHub's live installation-repos list:
+ * every returned repo is upserted with `isInstalled: true` (mirrors what the webhook handler already does on
+ * every forwarded event); every LOCAL repo previously marked installed under this installationId that is no
+ * longer in the fresh list is flipped to `isInstalled: false` (a repo removed from the installation, or moved
+ * out of a "selected" install's scope).
+ *
+ * No-op (`status: "skipped"`) outside broker mode -- a non-brokered self-host or cloud already gets this
+ * eagerly via real installation webhooks, and this must never run there (isOrbBrokerMode's signal, the
+ * enrollment secret's presence, is the same guard every other broker-only self-host path uses).
+ *
+ * Best-effort: any failure (broker down, GitHub throttled/erroring) returns `status: "failed"` rather than
+ * throwing, matching the fail-safe convention of every other cron sync in this codebase (e.g.
+ * registerOrbRelayTargetWithRetry) -- a sync miss self-heals on the next scheduled tick, never blocks the cron.
+ */
+export async function syncBrokeredInstalledRepos(
+  env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined } & Env,
+  fetchImpl: typeof fetch = fetch,
+): Promise<InstalledReposSyncResult> {
+  if (!isOrbBrokerMode(env)) return { status: "skipped" };
+  try {
+    const { token, installationId } = await fetchBrokeredInstallationToken(env, fetchImpl);
+    const repos = await fetchAllInstallationRepos(token, fetchImpl);
+    for (const repo of repos) {
+      await upsertRepositoryFromGitHub(env, repo, installationId);
+    }
+    const freshFullNames = new Set(repos.map((repo) => repo.full_name));
+    const previouslyInstalled = await listInstalledRepoFullNamesForInstallation(env, installationId);
+    const staleFullNames = previouslyInstalled.filter((fullName) => !freshFullNames.has(fullName));
+    await markRepositoriesRemovedFromInstallation(env, installationId, staleFullNames);
+    return { status: "synced", installationId, repoCount: repos.length, removedCount: staleFullNames.length };
+  } catch (error) {
+    return { status: "failed", reason: error instanceof Error ? error.message : "sync_failed" };
+  }
+}
@@ -32,6 +32,7 @@ import { runSelfTuneBreaker } from "../review/outcomes-wire";
 import { isRagEnabled } from "../review/rag-wire";
 import { processSubmitDraft } from "../services/draft";
 import { retryFailedRelays } from "../orb/relay";
+import { syncBrokeredInstalledRepos } from "../orb/installed-repos-sync";
 import { generateSignalSnapshots } from "./signal-snapshot";
 import { runRetentionPrune } from "./retention";
 // The 15 handlers below have no reason to move -- each is only reachable via this dispatcher (or, for
@@ -69,6 +70,9 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
     case "refresh-registry":
       await refreshRegistry(env);
       return;
+    case "sync-brokered-installed-repos":
+      await syncBrokeredInstalledRepos(env);
+      return;
     case "backfill-registered-repos":
       if (!message.repoFullName && message.requestedBy !== "test") {
         const repositories = (await listRepositories(env)).filter(

@@ -43,6 +43,7 @@ import { deterministicJitterMs, parsePositiveIntEnv } from "./queue-common";
 export const MAINTENANCE_JOB_TYPES: ReadonlySet<string> = new Set([
   "backfill-registered-repos",
   "refresh-registry",
+  "sync-brokered-installed-repos",
   "refresh-installation-health",
   "refresh-scoring-model",
   "refresh-upstream-drift",

@@ -118,6 +118,11 @@ const GITHUB_BUDGET_BACKGROUND_TYPES = new Set<string>([
   // runReviewRecapJob calls loadRepoFocusManifest directly for its one repo. Not yet cron-enqueued (manual/API
   // trigger only today, per its own doc comment), but still worth gating against a rapid repeated manual trigger.
   "generate-review-recap",
+  // syncBrokeredInstalledRepos (#5028) makes a real, paginated, authenticated `GET /installation/repositories`
+  // REST call using the brokered installation token -- unlike refresh-registry (an unauthenticated/raw-file
+  // fetch to entrius/gittensor, not the GitHub REST API), this genuinely draws down the shared installation's
+  // REST budget and must yield alongside every other budget consumer here.
+  "sync-brokered-installed-repos",
 ]);
 const PRIORITY_BY_TYPE = new Map([
   ["agent-regate-pr", AGENT_REGATE_PRIORITY],
@@ -926,6 +931,7 @@ export function jobCoalesceKey(payload: string): string | null {
     }
     switch (type) {
       case "refresh-registry":
+      case "sync-brokered-installed-repos":
       case "refresh-installation-health":
       case "refresh-scoring-model":
       case "refresh-upstream-drift":

@@ -56,6 +56,10 @@ export type JobMessage =
       type: "refresh-registry";
       requestedBy: "schedule" | "api" | "test";
     }
+  | {
+      type: "sync-brokered-installed-repos";
+      requestedBy: "schedule" | "api" | "test";
+    }
   | {
       type: "backfill-registered-repos";
       requestedBy: "schedule" | "api" | "test";

@@ -713,6 +713,46 @@ describe("worker entrypoint", () => {
     expect(sent.some((m) => m.type === "ops-alerts")).toBe(false);
   });
 
+  it("enqueues sync-brokered-installed-repos hourly ONLY in broker mode (ORB_ENROLLMENT_SECRET set, flag-OFF is byte-identical)", async () => {
+    const sentFor = async (enrollmentSecret?: string): Promise<Array<import("../../src/types").JobMessage>> => {
+      const sent: Array<import("../../src/types").JobMessage> = [];
+      const env = createTestEnv({
+        ...(enrollmentSecret === undefined ? {} : { ORB_ENROLLMENT_SECRET: enrollmentSecret }),
+        JOBS: {
+          async send(message: import("../../src/types").JobMessage) {
+            sent.push(message);
+          },
+        } as unknown as Queue,
+      });
+      const waitUntil: Promise<unknown>[] = [];
+      await worker.scheduled(controllerFor("2026-05-25T05:00:00.000Z"), env, executionContext(waitUntil));
+      await Promise.all(waitUntil);
+      return sent;
+    };
+
+    // Non-brokered (default) → no sync job; the enqueued set is unchanged from today.
+    expect((await sentFor()).some((m) => m.type === "sync-brokered-installed-repos")).toBe(false);
+    // Brokered (an enrollment secret is configured) → exactly one sync job, enqueued in the hourly window.
+    const brokered = await sentFor("orbsec_x");
+    expect(brokered.filter((m) => m.type === "sync-brokered-installed-repos")).toEqual([{ type: "sync-brokered-installed-repos", requestedBy: "schedule" }]);
+  });
+
+  it("does NOT enqueue sync-brokered-installed-repos outside the hourly window even in broker mode", async () => {
+    const sent: Array<import("../../src/types").JobMessage> = [];
+    const env = createTestEnv({
+      ORB_ENROLLMENT_SECRET: "orbsec_x",
+      JOBS: {
+        async send(message: import("../../src/types").JobMessage) {
+          sent.push(message);
+        },
+      } as unknown as Queue,
+    });
+    const waitUntil: Promise<unknown>[] = [];
+    await worker.scheduled(controllerFor("2026-05-25T05:15:00.000Z"), env, executionContext(waitUntil)); // non-hourly
+    await Promise.all(waitUntil);
+    expect(sent.some((m) => m.type === "sync-brokered-installed-repos")).toBe(false);
+  });
+
   it("enqueues the sweep-liveness-watchdog job hourly ONLY when GITTENSORY_SWEEP_WATCHDOG is ON (flag-OFF is byte-identical)", async () => {
     const sentFor = async (watchdogFlag?: string): Promise<Array<import("../../src/types").JobMessage>> => {
       const sent: Array<import("../../src/types").JobMessage> = [];

@@ -0,0 +1,108 @@
+import { describe, expect, it } from "vitest";
+import { getRepository } from "../../src/db/repositories";
+import { syncBrokeredInstalledRepos } from "../../src/orb/installed-repos-sync";
+import { createTestEnv } from "../helpers/d1";
+
+type Call = { url: string; init?: RequestInit | undefined };
+
+/** Router-style fetch stub: the broker token exchange goes to `.../v1/orb/token`, everything else is treated
+ *  as a GitHub `GET /installation/repositories` page request and answered from `pages` in order. */
+function routedFetch(args: { tokenResponse: Response; pages: Response[] }): { fetchImpl: typeof fetch; calls: Call[] } {
+  const calls: Call[] = [];
+  let pageIndex = 0;
+  const fetchImpl = (async (url: RequestInfo | URL, init?: RequestInit) => {
+    calls.push({ url: String(url), init });
+    if (String(url).includes("/v1/orb/token")) return args.tokenResponse;
+    const page = args.pages[pageIndex];
+    pageIndex += 1;
+    return page ?? Response.json({ repositories: [] });
+  }) as typeof fetch;
+  return { fetchImpl, calls };
+}
+
+function tokenResponse(installationId = 42): Response {
+  return Response.json({ token: "ghs_x", installationId, expiresAt: "2026-06-25T09:00:00Z", permissions: { contents: "write" } });
+}
+
+function repoPayload(fullName: string) {
+  const [owner, name] = fullName.split("/");
+  return { full_name: fullName, name, owner: { login: owner }, private: false, html_url: `https://github.com/${fullName}`, default_branch: "main" };
+}
+
+describe("syncBrokeredInstalledRepos", () => {
+  it("is a no-op outside broker mode (no enrollment secret)", async () => {
+    const env = createTestEnv();
+    const { fetchImpl, calls } = routedFetch({ tokenResponse: tokenResponse(), pages: [Response.json({ repositories: [repoPayload("owner/repo")] })] });
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toEqual({ status: "skipped" });
+    expect(calls).toHaveLength(0);
+  });
+
+  it("upserts every repo returned by GitHub as isInstalled, single page", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    const { fetchImpl } = routedFetch({
+      tokenResponse: tokenResponse(42),
+      pages: [Response.json({ repositories: [repoPayload("owner/repo-a"), repoPayload("owner/repo-b")] })],
+    });
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 2, removedCount: 0 });
+    await expect(getRepository(env, "owner/repo-a")).resolves.toMatchObject({ isInstalled: true, installationId: 42 });
+    await expect(getRepository(env, "owner/repo-b")).resolves.toMatchObject({ isInstalled: true, installationId: 42 });
+  });
+
+  it("paginates until a short page, following GitHub's own page-size convention", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    const fullPage = Response.json({ repositories: Array.from({ length: 100 }, (_, i) => repoPayload(`owner/repo-${i}`)) });
+    const shortPage = Response.json({ repositories: [repoPayload("owner/repo-last")] });
+    const { fetchImpl, calls } = routedFetch({ tokenResponse: tokenResponse(42), pages: [fullPage, shortPage] });
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 101, removedCount: 0 });
+    // token exchange + 2 pages
+    expect(calls).toHaveLength(3);
+    await expect(getRepository(env, "owner/repo-last")).resolves.toMatchObject({ isInstalled: true });
+  });
+
+  it("marks a previously-installed repo as no longer installed once GitHub stops returning it", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    // First sync: two repos installed.
+    const first = routedFetch({ tokenResponse: tokenResponse(42), pages: [Response.json({ repositories: [repoPayload("owner/kept"), repoPayload("owner/removed")] })] });
+    await syncBrokeredInstalledRepos(env, first.fetchImpl);
+    await expect(getRepository(env, "owner/removed")).resolves.toMatchObject({ isInstalled: true });
+
+    // Second sync: GitHub now only returns the kept repo.
+    const second = routedFetch({ tokenResponse: tokenResponse(42), pages: [Response.json({ repositories: [repoPayload("owner/kept")] })] });
+    const result = await syncBrokeredInstalledRepos(env, second.fetchImpl);
+    expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 1, removedCount: 1 });
+    await expect(getRepository(env, "owner/kept")).resolves.toMatchObject({ isInstalled: true });
+    await expect(getRepository(env, "owner/removed")).resolves.toMatchObject({ isInstalled: false, installationId: null });
+  });
+
+  it("fails safe (never throws) when the broker token exchange fails", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    const { fetchImpl } = routedFetch({ tokenResponse: new Response("nope", { status: 403 }), pages: [] });
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toMatchObject({ status: "failed" });
+    expect((result as { status: "failed"; reason: string }).reason).toMatch(/403/);
+  });
+
+  it("fails safe (never throws) when the GitHub installation-repos call errors", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    const { fetchImpl } = routedFetch({ tokenResponse: tokenResponse(42), pages: [new Response("rate limited", { status: 429 })] });
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toMatchObject({ status: "failed", reason: "installation_repositories_http_429" });
+  });
+
+  it("falls back to a generic reason when a non-Error value rejects the fetch", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    const fetchImpl = (() => Promise.reject("not an Error instance")) as typeof fetch;
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toEqual({ status: "failed", reason: "sync_failed" });
+  });
+
+  it("treats a missing repositories field on a page as an empty batch (ends pagination)", async () => {
+    const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" });
+    const { fetchImpl } = routedFetch({ tokenResponse: tokenResponse(42), pages: [Response.json({})] });
+    const result = await syncBrokeredInstalledRepos(env, fetchImpl);
+    expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 0, removedCount: 0 });
+  });
+});