feat(api): add cross-repo queue health federation index (#479)

Expose operator-only federated queue pressure ranking across registered installed repos, with MCP tool, dashboard wiring, and snapshot cache.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: kiannidev

apps/gittensory-ui/public/openapi.json

@@ -12391,6 +12391,91 @@
           "weekly",
           "byProject"
         ]
+      },
+      "FederatedRepoEntry": {
+        "type": "object",
+        "properties": {
+          "repoFullName": {
+            "type": "string"
+          },
+          "burdenScore": {
+            "type": "number"
+          },
+          "level": {
+            "type": "string",
+            "enum": [
+              "low",
+              "medium",
+              "high",
+              "critical"
+            ]
+          },
+          "compositeScore": {
+            "type": "number"
+          },
+          "stalePullRequestRate": {
+            "type": "number",
+            "nullable": true
+          },
+          "pullRequestGrowth7d": {
+            "type": "number",
+            "nullable": true
+          },
+          "freshness": {
+            "type": "string",
+            "enum": [
+              "fresh",
+              "stale"
+            ]
+          },
+          "summary": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "repoFullName",
+          "burdenScore",
+          "level",
+          "compositeScore",
+          "stalePullRequestRate",
+          "pullRequestGrowth7d",
+          "freshness",
+          "summary"
+        ]
+      },
+      "FederatedQueueIndex": {
+        "type": "object",
+        "properties": {
+          "generatedAt": {
+            "type": "string"
+          },
+          "repoCount": {
+            "type": "number"
+          },
+          "limitApplied": {
+            "type": "number"
+          },
+          "source": {
+            "type": "string",
+            "enum": [
+              "snapshot",
+              "computed"
+            ]
+          },
+          "entries": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/FederatedRepoEntry"
+            }
+          }
+        },
+        "required": [
+          "generatedAt",
+          "repoCount",
+          "limitApplied",
+          "source",
+          "entries"
+        ]
       }
     },
     "parameters": {},
@@ -15262,6 +15347,49 @@
           }
         ]
       }
+    },
+    "/v1/app/queue-health/federation": {
+      "get": {
+        "parameters": [
+          {
+            "schema": {
+              "type": "string"
+            },
+            "required": false,
+            "name": "limit",
+            "in": "query"
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Ranked cross-repo queue pressure index (operator only)",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/FederatedQueueIndex"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Unauthorized"
+          },
+          "403": {
+            "description": "Insufficient role — operator access required"
+          },
+          "422": {
+            "description": "Invalid limit parameter"
+          }
+        },
+        "security": [
+          {
+            "GittensoryBearer": []
+          },
+          {
+            "GittensorySessionCookie": []
+          }
+        ]
+      }
     }
   },
   "servers": [

migrations/0074_queue_federation_cache.sql

@@ -0,0 +1,11 @@
+-- Cache table for the federated queue pressure index.
+-- TTL enforcement matches the burden forecast pattern (6-hour freshness threshold applied at read time).
+CREATE TABLE IF NOT EXISTS queue_federation_snapshots (
+  id TEXT PRIMARY KEY,
+  generated_at TEXT NOT NULL,
+  repo_count INTEGER NOT NULL DEFAULT 0,
+  payload_json TEXT NOT NULL DEFAULT '{}',
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS queue_federation_snapshots_generated_idx ON queue_federation_snapshots (generated_at);

src/api/routes.ts

@@ -172,6 +172,7 @@ import {
   MINIMUM_SUPPORTED_MCP_VERSION,
 } from "../services/mcp-compatibility";
 import { buildOperatorDashboardPayload } from "../services/operator-dashboard";
+import { buildFederatedQueueIndex, FEDERATED_QUEUE_INDEX_MAX_LIMIT } from "../services/queue-federation";
 import { buildSelfDogfoodRegistrationPack, resolveSelfDogfoodRepoFullName } from "../services/self-dogfood-registration-pack";
 import { buildSubnetInterfaceDescriptor } from "../services/subnet-interface";
 import { buildPublicRepoQuality, type PublicRepoQuality } from "../services/public-repo-quality";
@@ -1329,6 +1330,20 @@ export function createApp() {
     return c.json(await buildOperatorDashboardPayload(c.env));
   });
 
+  app.get("/v1/app/queue-health/federation", async (c) => {
+    const forbidden = await requireAppRole(c, ["operator"]);
+    if (forbidden) return forbidden;
+    const rawLimit = c.req.query("limit");
+    if (rawLimit !== undefined) {
+      const parsed = Number(rawLimit);
+      if (!Number.isInteger(parsed) || parsed < 1 || parsed > FEDERATED_QUEUE_INDEX_MAX_LIMIT) {
+        return c.json({ error: "invalid_limit", message: `limit must be an integer between 1 and ${FEDERATED_QUEUE_INDEX_MAX_LIMIT}` }, 422);
+      }
+    }
+    const limit = rawLimit !== undefined ? Number(rawLimit) : undefined;
+    return c.json(await buildFederatedQueueIndex(c.env, limit));
+  });
+
   app.get("/v1/app/notification-model", async (c) => {
     const forbidden = await requireAppRole(c, ["maintainer", "owner", "operator"]);
     if (forbidden) return forbidden;

src/db/repositories.ts

@@ -42,6 +42,7 @@ import {
   repositories,
   repoGithubTotalsSnapshots,
   repoQueueTrendSnapshots,
+  queueFederationSnapshots,
   registryDriftEvents,
   repoLabels,
   repoSnapshots,
@@ -89,6 +90,7 @@ import type {
   BountyLifecycleEventRecord,
   BountyRecord,
   BurdenForecastRecord,
+  QueueFederationSnapshotRecord,
   CheckSummaryRecord,
   CollisionEdgeRecord,
   CommandUsefulnessSummary,
@@ -2477,6 +2479,32 @@ export async function getBurdenForecast(env: Env, repoFullName: string): Promise
   };
 }
 
+const QUEUE_FEDERATION_SNAPSHOT_ID = "current";
+
+export async function upsertQueueFederationSnapshot(env: Env, snapshot: QueueFederationSnapshotRecord): Promise<void> {
+  const db = getDb(env.DB);
+  await db
+    .insert(queueFederationSnapshots)
+    .values({ id: QUEUE_FEDERATION_SNAPSHOT_ID, generatedAt: snapshot.generatedAt, repoCount: snapshot.repoCount, payloadJson: jsonString(snapshot.payload) })
+    .onConflictDoUpdate({
+      target: queueFederationSnapshots.id,
+      set: { generatedAt: snapshot.generatedAt, repoCount: snapshot.repoCount, payloadJson: jsonString(snapshot.payload) },
+    });
+}
+
+export async function getQueueFederationSnapshot(env: Env): Promise<QueueFederationSnapshotRecord | null> {
+  const db = getDb(env.DB);
+  const row = await db.select().from(queueFederationSnapshots).where(eq(queueFederationSnapshots.id, QUEUE_FEDERATION_SNAPSHOT_ID)).limit(1);
+  const first = row[0];
+  if (!first) return null;
+  return {
+    id: first.id,
+    generatedAt: first.generatedAt,
+    repoCount: first.repoCount,
+    payload: parseJson<Record<string, JsonValue>>(first.payloadJson, {}),
+  };
+}
+
 export async function persistRegistryDriftEvents(env: Env, events: RegistryDriftEventRecord[]): Promise<void> {
   const db = getDb(env.DB);
   for (const event of events) {

src/db/schema.ts

@@ -768,6 +768,14 @@ export const repoQueueTrendSnapshots = sqliteTable("repo_queue_trend_snapshots",
   generatedAt: text("generated_at").notNull().$defaultFn(() => nowIso()),
 });
 
+export const queueFederationSnapshots = sqliteTable("queue_federation_snapshots", {
+  id: text("id").primaryKey(),
+  generatedAt: text("generated_at").notNull(),
+  repoCount: integer("repo_count").notNull().default(0),
+  payloadJson: text("payload_json").notNull().default("{}"),
+  createdAt: text("created_at").notNull().default("CURRENT_TIMESTAMP"),
+});
+
 export const registryDriftEvents = sqliteTable("registry_drift_events", {
   id: text("id").primaryKey(),
   repoFullName: text("repo_full_name").notNull(),

src/mcp/server.ts

@@ -67,6 +67,7 @@ import { buildRemediationPlan } from "../services/remediation-plan";
 import { explainScoreBreakdown } from "../services/score-breakdown";
 import { loadOrComputeIssueQualityResponse } from "../services/issue-quality";
 import { loadOrComputeBurdenForecastResponse } from "../services/burden-forecast";
+import { buildFederatedQueueIndex, FEDERATED_QUEUE_INDEX_MAX_LIMIT } from "../services/queue-federation";
 import { buildMcpClientTelemetry } from "../services/client-telemetry";
 import { loadOrComputeRepoOutcomePatternsResponse } from "../services/repo-outcome-patterns";
 import { buildRepoOutcomeCalibration, outcomeCalibrationSummary } from "../services/outcome-calibration";
@@ -158,6 +159,14 @@ const fleetAnalyticsOutputSchema = {
   outliers: z.array(z.unknown()).optional(),
 };
 
+const queueHealthFederationOutputSchema = {
+  generatedAt: z.string().optional(),
+  repoCount: z.number().optional(),
+  limitApplied: z.number().optional(),
+  source: z.enum(["snapshot", "computed"]).optional(),
+  entries: z.array(z.unknown()).optional(),
+};
+
 const loginShape = {
   login: z.string().min(1),
 };
@@ -1050,6 +1059,16 @@ export class GittensoryMcp {
       async (input) => this.toolResult(await this.getBurdenForecast(input)),
     );
 
+    server.registerTool(
+      "gittensory_queue_health_federation",
+      {
+        description: "Return a ranked cross-repo queue pressure index showing the worst-burden registered repos. Operator-only.",
+        inputSchema: { limit: z.number().int().min(1).max(FEDERATED_QUEUE_INDEX_MAX_LIMIT).optional() },
+        outputSchema: queueHealthFederationOutputSchema,
+      },
+      async (input) => this.toolResult(await this.getQueueHealthFederation(input.limit)),
+    );
+
     server.registerTool(
       "gittensory_get_repo_outcome_patterns",
       {
@@ -1813,6 +1832,22 @@ export class GittensoryMcp {
     };
   }
 
+  private async getQueueHealthFederation(limit?: number): Promise<ToolPayload> {
+    if (this.identity.kind !== "session") {
+      throw new Error("Forbidden: gittensory_queue_health_federation requires operator role.");
+    }
+    const summary = await loadControlPanelRoleSummary(this.env, this.identity.actor);
+    if (!summary.roles.includes("operator")) {
+      throw new Error("Forbidden: gittensory_queue_health_federation requires operator role.");
+    }
+    const index = await buildFederatedQueueIndex(this.env, limit);
+    const criticalCount = index.entries.filter((entry) => entry.level === "critical" || entry.level === "high").length;
+    return {
+      summary: `Cross-repo queue pressure index: ${index.repoCount} repo(s) ranked, ${criticalCount} at critical/high burden.`,
+      data: index as unknown as Record<string, unknown>,
+    };
+  }
+
   private async getIssueQuality(input: { owner: string; repo: string }): Promise<ToolPayload> {
     const fullName = `${input.owner}/${input.repo}`;
     if (!(await this.canAccessRepo(fullName))) {

src/openapi/schemas.ts

@@ -177,6 +177,29 @@ export const CollisionReportSchema = z
   })
   .openapi("CollisionReport");
 
+export const FederatedRepoEntrySchema = z
+  .object({
+    repoFullName: z.string(),
+    burdenScore: z.number(),
+    level: z.enum(["low", "medium", "high", "critical"]),
+    compositeScore: z.number(),
+    stalePullRequestRate: z.number().nullable(),
+    pullRequestGrowth7d: z.number().nullable(),
+    freshness: z.enum(["fresh", "stale"]),
+    summary: z.string(),
+  })
+  .openapi("FederatedRepoEntry");
+
+export const FederatedQueueIndexSchema = z
+  .object({
+    generatedAt: z.string(),
+    repoCount: z.number(),
+    limitApplied: z.number(),
+    source: z.enum(["snapshot", "computed"]),
+    entries: z.array(FederatedRepoEntrySchema),
+  })
+  .openapi("FederatedQueueIndex");
+
 export const QueueHealthSchema = z
   .object({
     repoFullName: z.string(),

src/openapi/spec.ts

@@ -11,6 +11,8 @@ import {
   BountyLifecycleEventsSchema,
   BountySchema,
   BurdenForecastSchema,
+  FederatedQueueIndexSchema,
+  FederatedRepoEntrySchema,
   CollisionReportSchema,
   ConfigQualitySchema,
   CommandPreviewResponseSchema,
@@ -147,6 +149,8 @@ export function buildOpenApiSpec() {
   registry.register("IssueQualityReport", IssueQualityReportSchema);
   registry.register("IssueQualityResponse", IssueQualityResponseSchema);
   registry.register("BurdenForecast", BurdenForecastSchema);
+  registry.register("FederatedRepoEntry", FederatedRepoEntrySchema);
+  registry.register("FederatedQueueIndex", FederatedQueueIndexSchema);
   registry.register("ContributorScoringProfile", ContributorScoringProfileSchema);
   registry.register("ContributorStrategy", ContributorStrategySchema);
   registry.register("RewardRiskAction", RewardRiskActionSchema);
@@ -715,6 +719,17 @@ export function buildOpenApiSpec() {
       401: { description: "Unauthorized" },
     },
   });
+  registry.registerPath({
+    method: "get",
+    path: "/v1/app/queue-health/federation",
+    request: { query: z.object({ limit: z.string().optional() }) },
+    responses: {
+      200: { description: "Ranked cross-repo queue pressure index (operator only)", content: { "application/json": { schema: FederatedQueueIndexSchema } } },
+      401: { description: "Unauthorized" },
+      403: { description: "Insufficient role — operator access required" },
+      422: { description: "Invalid limit parameter" },
+    },
+  });
   for (const path of [
     "/v1/app/roles",
     "/v1/app/miner-dashboard",