feat(orb): registration-gate the fleet + cap ingest body (das-github-mirror model)

migrations/0061_orb_instances.sql

@@ -0,0 +1,17 @@
+-- Gittensory Orb (#1255) — instance registration gate, modeled on das-github-mirror's `registered=false`
+-- default. Every self-host instance that POSTs anonymized batches to /v1/orb/ingest is recorded here on
+-- first contact, but its signals only count toward fleet calibration once an operator REGISTERS it
+-- (registered=1). This is the fleet's trust anchor: ingest stays open + frictionless (no shared secret —
+-- the topology has no per-instance key the collector could verify), but a stranger — or a ring of them —
+-- cannot move the fleet median until a human opts them in. Signals are still stored for everyone (so a
+-- later registration is retroactive); computeFleetAnalytics is what filters to registered instances.
+CREATE TABLE IF NOT EXISTS orb_instances (
+  instance_id TEXT PRIMARY KEY NOT NULL,
+  -- 0 until an operator opts the instance into fleet calibration; computeFleetAnalytics counts only registered.
+  registered INTEGER NOT NULL DEFAULT 0,
+  first_seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  last_seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  registered_at TEXT
+);
+
+CREATE INDEX IF NOT EXISTS orb_instances_registered_idx ON orb_instances(registered);

src/api/routes.ts

@@ -122,7 +122,7 @@ import {
   type GittensoryMentionCommandName,
 } from "../github/commands";
 import { handleGitHubWebhook } from "../github/webhook";
-import { handleOrbIngest } from "../orb/ingest";
+import { handleOrbIngest, readOrbIngestBody } from "../orb/ingest";
 import { computeFleetAnalytics } from "../orb/analytics";
 import { handleMcpRequest } from "../mcp/server";
 import { buildOpenApiSpec } from "../openapi/spec";
@@ -2868,7 +2868,10 @@ export function createApp() {
   // outcome batches from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender;
   // dedup is enforced via UNIQUE(instance_id, repo_hash, pr_hash) in orb_signals. Rate-limited (strict, #1254).
   app.post("/v1/orb/ingest", async (c) => {
-    const body = await c.req.text().catch(() => null);
+    // Open ingress (no shared secret — the fleet topology has no per-instance key the collector could
+    // verify), bounded by a hard body ceiling so it can't be used to make us buffer unbounded input.
+    const body = await readOrbIngestBody(c.req.raw, c.req.header("content-length"));
+    if (body === null) return c.json({ error: "payload_too_large" }, 413);
     if (!body) return c.json({ error: "invalid_request" }, 400);
     const result = await handleOrbIngest(body, c.env.DB);
     if ("error" in result) return c.json(result, 400);
@@ -2883,6 +2886,40 @@ export function createApp() {
     return c.json(await computeFleetAnalytics(c.env, { windowDays: days }));
   });
 
+  // Orb instance registry — the fleet trust gate. Every self-host instance that ingests is recorded here,
+  // but only REGISTERED ones count toward fleet calibration (computeFleetAnalytics). Bearer-gated by the
+  // `/v1/internal/*` middleware (INTERNAL_JOB_TOKEN). List shows pending + registered instances with their
+  // stored-signal counts so an operator knows what they're opting in before they register it.
+  app.get("/v1/internal/orb/instances", async (c) => {
+    const rows = await c.env.DB
+      .prepare(
+        `SELECT i.instance_id AS instanceId, i.registered AS registered, i.first_seen_at AS firstSeenAt,
+                i.last_seen_at AS lastSeenAt, i.registered_at AS registeredAt,
+                (SELECT COUNT(*) FROM orb_signals s WHERE s.instance_id = i.instance_id) AS signalCount
+         FROM orb_instances i ORDER BY i.last_seen_at DESC`,
+      )
+      .all<{ instanceId: string; registered: number; firstSeenAt: string; lastSeenAt: string; registeredAt: string | null; signalCount: number }>();
+    return c.json({ instances: (rows.results ?? []).map((r) => ({ ...r, registered: r.registered === 1 })) });
+  });
+
+  // Opt an instance into (or out of) fleet calibration. Body: { instanceId, registered? } (registered
+  // defaults true). Upserts so an operator can register an instance that has ingested but isn't recorded yet.
+  app.post("/v1/internal/orb/instances/register", async (c) => {
+    const payload = (await c.req.json().catch(() => null)) as { instanceId?: unknown; registered?: unknown } | null;
+    const instanceId = typeof payload?.instanceId === "string" ? payload.instanceId : "";
+    if (!instanceId) return c.json({ error: "instanceId required" }, 400);
+    const registered = payload?.registered === false ? 0 : 1;
+    await c.env.DB
+      .prepare(
+        `INSERT INTO orb_instances (instance_id, registered, registered_at) VALUES (?, ?, CURRENT_TIMESTAMP)
+         ON CONFLICT(instance_id) DO UPDATE SET registered = excluded.registered,
+           registered_at = CASE WHEN excluded.registered = 1 THEN CURRENT_TIMESTAMP ELSE NULL END`,
+      )
+      .bind(instanceId, registered)
+      .run();
+    return c.json({ instanceId, registered: registered === 1 });
+  });
+
   // Convergence (ops / observability, flag GITTENSORY_REVIEW_OPS). Cross-repo review-OUTCOME aggregate (gate-block
   // ledger + recommendation/slop calibration) for an operator dashboard. Bearer-gated by the `/v1/internal/*`
   // middleware above (INTERNAL_JOB_TOKEN). Flag-OFF (default) → 404, so the endpoint does not exist and the

src/orb/analytics.ts

@@ -91,6 +91,7 @@ export async function computeFleetAnalytics(env: Env, opts: { windowDays?: numbe
 
   let cells: Cell[] = [];
   let cycle: number[] = [];
+  let registered = new Set<string>();
   try {
     const matrix = await env.DB
       .prepare(
@@ -106,6 +107,10 @@ export async function computeFleetAnalytics(env: Env, opts: { windowDays?: numbe
       .bind(cutoff)
       .all<{ ms: number }>();
     cycle = (cy.results ?? []).map((r) => r.ms);
+    // The fleet trust gate: only operator-registered instances count toward the median (open ingest stores
+    // everyone's signals, but a stranger can't move calibration until a human opts them in — #1255).
+    const reg = await env.DB.prepare(`SELECT instance_id FROM orb_instances WHERE registered = 1`).all<{ instance_id: string }>();
+    registered = new Set((reg.results ?? []).map((r) => r.instance_id));
   } catch {
     return { windowDays, instanceCount: 0, fleet: { mergePrecision: null, closePrecision: null, fpRate: null, reversalRate: null, cycleP50Ms: null, cycleP95Ms: null }, instances: [], outliers: [] };
   }
@@ -119,8 +124,9 @@ export async function computeFleetAnalytics(env: Env, opts: { windowDays?: numbe
   }
   const instances = [...byInstance.entries()].map(([id, cs]) => foldInstance(id, cs)).sort((a, b) => a.instanceId.localeCompare(b.instanceId));
 
-  // Fleet = median across instances with enough volume (robust to a single bad contributor).
-  const eligible = instances.filter((i) => i.decided >= MIN_DECIDED);
+  // Fleet = median across REGISTERED instances with enough volume (robust to a single bad contributor and
+  // to unregistered/untrusted senders — registration is the fleet's trust anchor).
+  const eligible = instances.filter((i) => i.decided >= MIN_DECIDED && registered.has(i.instanceId));
   const nums = (sel: (i: InstanceMetrics) => number | null): number[] => eligible.map(sel).filter((v): v is number => v !== null);
   const fleetMergeP = median(nums((i) => i.mergePrecision));
   const fleetCloseP = median(nums((i) => i.closePrecision));

src/orb/ingest.ts

@@ -9,6 +9,42 @@ const VALID_REVERSALS = new Set(["none", "reopened", "reverted"]);
 const MIN_CYCLE_MS = 1_000; // <1s is implausible
 const MAX_CYCLE_MS = 31_536_000_000; // >1y is implausible
 
+// 1 MiB comfortably holds a full MAX_BATCH (500) of small anonymized events (~hashes + numbers) with
+// headroom, while bounding how much a hostile sender can make the collector buffer. Mirrors the
+// body limit das-github-mirror puts in front of its open webhook ingress.
+export const MAX_ORB_INGEST_BODY_BYTES = 1_048_576;
+
+function parseContentLength(header: string | null | undefined): number | null {
+  if (typeof header !== "string") return null;
+  const n = Number(header);
+  return Number.isInteger(n) && n >= 0 ? n : null;
+}
+
+/** Read the request body with a hard byte ceiling so a hostile sender can't make us buffer unbounded
+ *  input. Returns null when the body exceeds MAX_ORB_INGEST_BODY_BYTES (the caller answers 413). */
+export async function readOrbIngestBody(request: Request, contentLengthHeader: string | null | undefined): Promise<string | null> {
+  const declared = parseContentLength(contentLengthHeader);
+  if (declared !== null && declared > MAX_ORB_INGEST_BODY_BYTES) return null;
+
+  const stream = request.body;
+  if (!stream) return "";
+  const reader = stream.getReader();
+  const decoder = new TextDecoder();
+  let total = 0;
+  let out = "";
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    total += value.byteLength;
+    if (total > MAX_ORB_INGEST_BODY_BYTES) {
+      await reader.cancel();
+      return null;
+    }
+    out += decoder.decode(value, { stream: true });
+  }
+  return out + decoder.decode();
+}
+
 interface OrbIngestEvent {
   repo_hash: string;
   pr_hash: string;
@@ -55,6 +91,19 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise<Orb
     return { error: "invalid_payload" };
   }
 
+  // Record the instance on first contact (registered=0 by default) and bump last_seen. The registration
+  // gate lives in computeFleetAnalytics: signals are stored for everyone, but only registered instances
+  // count toward the fleet median — so open ingest can't be used to skew calibration (the das-github-mirror
+  // model: every source is seen, trusted only once an operator opts it in).
+  try {
+    await db
+      .prepare(`INSERT INTO orb_instances (instance_id) VALUES (?) ON CONFLICT(instance_id) DO UPDATE SET last_seen_at = CURRENT_TIMESTAMP`)
+      .bind(instance_id)
+      .run();
+  } catch {
+    // best-effort: never fail ingest because the instance bookkeeping hiccupped
+  }
+
   const batch = events.slice(0, MAX_BATCH);
   let accepted = 0;
 

test/integration/orb-ingest.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from "vitest";
 import { createApp } from "../../src/api/routes";
-import { handleOrbIngest } from "../../src/orb/ingest";
+import { handleOrbIngest, MAX_ORB_INGEST_BODY_BYTES, readOrbIngestBody } from "../../src/orb/ingest";
 import { createTestEnv, TestD1Database } from "../helpers/d1";
 
 describe("handleOrbIngest()", () => {
@@ -127,6 +127,61 @@ describe("handleOrbIngest()", () => {
     const db = { prepare: () => ({ bind: () => ({ run: () => Promise.resolve({ meta: { changes: 0 } }) }) }) } as unknown as D1Database;
     expect(await ingest(db, [ev()])).toEqual({ accepted: 0 });
   });
+
+  it("records the instance on first contact (registered=0) and bumps last_seen on re-ingest", async () => {
+    const db = makeDb();
+    await ingest(db, [ev({ pr_hash: "i1" })], "instX");
+    const row = await (db as unknown as TestD1Database)
+      .prepare("SELECT registered, first_seen_at, last_seen_at FROM orb_instances WHERE instance_id=?")
+      .bind("instX")
+      .first<{ registered: number; first_seen_at: string; last_seen_at: string }>();
+    expect(row?.registered).toBe(0); // not trusted until an operator registers it
+    await ingest(db, [ev({ pr_hash: "i2" })], "instX"); // same instance again → still one row
+    const cnt = await (db as unknown as TestD1Database).prepare("SELECT COUNT(*) AS n FROM orb_instances WHERE instance_id=?").bind("instX").first<{ n: number }>();
+    expect(cnt?.n).toBe(1);
+  });
+
+  it("does not fail ingest if the instance bookkeeping upsert throws", async () => {
+    // First prepare() (orb_instances upsert) rejects; ingest must still process the batch best-effort.
+    let call = 0;
+    const db = {
+      prepare: (sql: string) => {
+        call++;
+        if (sql.includes("orb_instances")) return { bind: () => ({ run: () => Promise.reject(new Error("boom")) }) };
+        return new TestD1Database().prepare(sql);
+      },
+    } as unknown as D1Database;
+    expect(await ingest(db, [ev()])).toBeTruthy();
+    expect(call).toBeGreaterThan(0);
+  });
+});
+
+describe("readOrbIngestBody()", () => {
+  const reqWithBody = (body: BodyInit, headers?: Record<string, string>) =>
+    new Request("http://collector/v1/orb/ingest", { method: "POST", body, ...(headers ? { headers } : {}) });
+
+  it("reads a normal body", async () => {
+    expect(await readOrbIngestBody(reqWithBody("hello"), "5")).toBe("hello");
+  });
+
+  it("returns '' when there is no request body", async () => {
+    expect(await readOrbIngestBody(new Request("http://collector", { method: "POST" }), null)).toBe("");
+  });
+
+  it("rejects (null) when the declared content-length exceeds the cap — without reading", async () => {
+    expect(await readOrbIngestBody(reqWithBody("tiny"), String(MAX_ORB_INGEST_BODY_BYTES + 1))).toBeNull();
+  });
+
+  it("ignores a non-numeric content-length and reads normally", async () => {
+    expect(await readOrbIngestBody(reqWithBody("ok"), "not-a-number")).toBe("ok");
+  });
+
+  it("rejects (null) when the streamed body exceeds the cap with no declared length", async () => {
+    const big = new Uint8Array(MAX_ORB_INGEST_BODY_BYTES + 8);
+    const stream = new ReadableStream<Uint8Array>({ start(ctrl) { ctrl.enqueue(big); ctrl.close(); } });
+    const req = new Request("http://collector", { method: "POST", body: stream, ...({ duplex: "half" } as object) });
+    expect(await readOrbIngestBody(req, null)).toBeNull();
+  });
 });
 
 describe("POST /v1/orb/ingest route", () => {
@@ -150,6 +205,66 @@ describe("POST /v1/orb/ingest route", () => {
     const res = await app.request("/v1/orb/ingest", { method: "POST", body: "" }, createTestEnv());
     expect(res.status).toBe(400);
   });
+
+  it("returns 413 when the body exceeds the ingest byte ceiling", async () => {
+    const huge = "x".repeat(MAX_ORB_INGEST_BODY_BYTES + 16);
+    const res = await app.request("/v1/orb/ingest", { method: "POST", body: huge }, createTestEnv());
+    expect(res.status).toBe(413);
+    expect(((await res.json()) as { error: string }).error).toBe("payload_too_large");
+  });
+});
+
+describe("Orb instance registry routes (/v1/internal/orb/instances)", () => {
+  const app = createApp();
+  const auth = { authorization: "Bearer dev-internal-token" };
+  const ingestOne = (env: Env, instance: string) =>
+    app.request("/v1/orb/ingest", { method: "POST", body: JSON.stringify({ instance_id: instance, events: [{ repo_hash: "r", pr_hash: `${instance}-p`, outcome: "merged" }] }) }, env);
+
+  it("lists ingested instances as unregistered with their stored-signal count", async () => {
+    const env = createTestEnv();
+    await ingestOne(env, "inst-a");
+    const res = await app.request("/v1/internal/orb/instances", { headers: auth }, env);
+    expect(res.status).toBe(200);
+    const { instances } = (await res.json()) as { instances: Array<{ instanceId: string; registered: boolean; signalCount: number }> };
+    expect(instances).toEqual([expect.objectContaining({ instanceId: "inst-a", registered: false, signalCount: 1 })]);
+  });
+
+  it("401 without the internal token", async () => {
+    expect((await app.request("/v1/internal/orb/instances", {}, createTestEnv())).status).toBe(401);
+  });
+
+  it("registers an instance (and can unregister it)", async () => {
+    const env = createTestEnv();
+    await ingestOne(env, "inst-b");
+    const reg = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({ instanceId: "inst-b" }) }, env);
+    expect(((await reg.json()) as { registered: boolean }).registered).toBe(true);
+    const off = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({ instanceId: "inst-b", registered: false }) }, env);
+    expect(((await off.json()) as { registered: boolean }).registered).toBe(false);
+  });
+
+  it("registers an instance that has not ingested yet (upsert)", async () => {
+    const env = createTestEnv();
+    const reg = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({ instanceId: "never-seen" }) }, env);
+    expect(reg.status).toBe(200);
+    const list = (await (await app.request("/v1/internal/orb/instances", { headers: auth }, env)).json()) as { instances: Array<{ instanceId: string; registered: boolean }> };
+    expect(list.instances).toEqual([expect.objectContaining({ instanceId: "never-seen", registered: true })]);
+  });
+
+  it("400 when instanceId is missing", async () => {
+    const res = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({}) }, createTestEnv());
+    expect(res.status).toBe(400);
+  });
+
+  it("400 on a non-JSON register body (json().catch → null)", async () => {
+    const res = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: "{bad" }, createTestEnv());
+    expect(res.status).toBe(400);
+  });
+
+  it("tolerates a list query that omits results (rows.results ?? [])", async () => {
+    const env = { ...createTestEnv(), DB: { prepare: () => ({ all: () => Promise.resolve({}) }) } } as unknown as Env;
+    const res = await app.request("/v1/internal/orb/instances", { headers: auth }, env);
+    expect(((await res.json()) as { instances: unknown[] }).instances).toEqual([]);
+  });
 });
 
 describe("GET /v1/internal/fleet/analytics route", () => {

test/unit/mcp-fleet-analytics.test.ts

@@ -22,6 +22,8 @@ async function seedMergeSignals(env: Env, instance: string, n: number): Promise<
       .bind(instance, `repo${seq}`, `pr${seq++}`)
       .run();
   }
+  // Register the instance so it counts toward the fleet (only registered instances are aggregated).
+  await env.DB.prepare(`INSERT INTO orb_instances (instance_id, registered) VALUES (?, 1) ON CONFLICT(instance_id) DO UPDATE SET registered=1`).bind(instance).run();
 }
 
 describe("gittensory_get_fleet_analytics MCP tool", () => {

test/unit/operator-dashboard.test.ts

@@ -47,6 +47,9 @@ describe("operator dashboard payload", () => {
     await seed("good1", 5, "merged"); // precision 1.0
     await seed("good2", 5, "merged"); // precision 1.0
     await seed("bad", 5, "closed"); // precision 0.0 → outlier vs the median (1.0)
+    for (const id of ["good1", "good2", "bad"]) {
+      await env.DB.prepare(`INSERT INTO orb_instances (instance_id, registered) VALUES (?, 1)`).bind(id).run(); // only registered instances count
+    }
     const payload = await buildOperatorDashboardPayload(env);
     expect(payload.fleetMetrics.instanceCount).toBe(3);
     expect(payload.fleetMetrics.outliers.map((o) => o.instanceId)).toContain("bad");