JSONbored · JSONbored · Jun 24, 2026 · Jun 24, 2026
@@ -161,26 +161,19 @@ GITTENSORY_REVIEW_DRAFT=false
 #                                            #   1024-dimensional (e.g. bge-m3 or mxbai-embed-large via Ollama).
 #                                            #   Used only when RAG is enabled (GITTENSORY_REVIEW_RAG + allowlist).
 
-# --- Gittensory Orb (#1219; opt-in outcome signal collection) ---
-# Run GET /orb/setup to create the Orb GitHub App (read-only; separate from the main App).
-# Credentials are written to /data/gittensory-orb.env on callback — load them here.
+# --- Gittensory Orb (#1255; opt-in fleet-calibration export) ---
+# Orb is the central collector + analytics that aggregates anonymized gate-calibration data UP from
+# self-hosted instances. There is NO separate Orb GitHub App and NO setup wizard: your existing main App
+# already records de-noised outcomes (merged/closed + reversals) locally — flip ORB_ENABLED to ship an
+# anonymized signal to gittensory's collector. That's it: no second App, no extra secret, no wizard.
 #
-# SECURITY MODEL (this image is meant to be self-hosted by many independent maintainers):
-#   • The image bakes NO secrets. Every operator creates their OWN Orb App via /orb/setup, so the
-#     ORB_* secrets below are unique to YOUR instance and live only in YOUR /data — never shared,
-#     never sent to the collector. gittensory's own App secrets are never in the image.
-#   • Export to the central collector uses NO shared key: repo/PR identifiers are HMAC-anonymized
-#     with YOUR ORB_WEBHOOK_SECRET (so even the collector operator can't de-anonymize them), and the
-#     collector accepts the batch as untrusted, rate-limited, aggregate-only telemetry. Nothing in the
-#     container, if leaked, can compromise the collector, other operators, or the main App.
-#   • Prefer the *_FILE convention for the private key (mount a file, not an inline env value):
-#     ORB_PRIVATE_KEY_FILE=/run/secrets/orb_private_key  →  the server reads it into ORB_PRIVATE_KEY.
-# ORB_APP_ID=                                # App ID from /orb/setup callback
-# ORB_APP_SLUG=                              # App slug (human-readable name)
-# ORB_WEBHOOK_SECRET=                        # secret from /orb/setup callback — signs /orb/webhook + anonymizes export
-# ORB_PRIVATE_KEY=                           # PEM from /orb/setup callback (JSON-stringified); prefer ORB_PRIVATE_KEY_FILE
-# ORB_ENABLED=false                          # master switch: set to true to enable collection (default off)
-# ORB_AIR_GAP=false                          # set to true to keep all data local — never send to the collector
-# ORB_ANONYMIZE=true                         # HMAC-hash repo names before export (default true; false = raw names)
+# SECURITY MODEL (this image is self-hosted by many independent maintainers):
+#   • The image bakes NO secrets. repo/PR identifiers are HMAC-anonymized with YOUR own ORB_WEBHOOK_SECRET
+#     (a stable per-instance string), so even gittensory (running the collector) can never de-anonymize them.
+#   • Export carries NO shared key. The collector accepts the batch as untrusted, rate-limited, aggregate-only
+#     telemetry. Nothing in the container, if leaked, can compromise the collector, other operators, or any App.
+# ORB_ENABLED=false                          # master switch: set to true to export fleet-calibration signal (default off)
+# ORB_WEBHOOK_SECRET=<stable-random-string>  # the per-instance HMAC key used to anonymize repo/PR identifiers
+# ORB_AIR_GAP=false                          # set to true to compute locally but never send to the collector
+# ORB_ANONYMIZE=true                         # HMAC-hash repo/PR before export (default true; false = raw names)
 # ORB_COLLECTOR_URL=https://gittensory-api.aethereal.dev/v1/orb/ingest  # gittensory's hosted collector (default; override for your own)
-# ORB_SETUP_OUTPUT_PATH=/data/gittensory-orb.env          # where /orb/setup/callback writes the credentials file
@@ -0,0 +1,41 @@
+-- Gittensory Orb (#1255): turn Orb into the central fleet-calibration collector.
+--
+-- Retire the per-instance Orb GitHub App pipeline (orb_events / orb_installations were written by the
+-- now-removed /orb/webhook handler). Each self-hosted instance already records de-noised ground truth in
+-- review_audit (gate_decision + pr_outcome + reversal_*), so the exporter now reads THAT and ships an
+-- anonymized, reversal-aware signal up to the central orb_signals store.
+
+DROP TABLE IF EXISTS orb_events;
+DROP TABLE IF EXISTS orb_installations;
+
+-- orb_signals is young, continuously-regenerated telemetry (instances re-export). SQLite can't ALTER away the
+-- old table-level UNIQUE(instance_id, pr_hash) — which is wrong (two instances reviewing owner/repo#123
+-- collide) — so recreate with the correct key + the new reversal/timestamp/reason columns.
+DROP TABLE IF EXISTS orb_signals;
+CREATE TABLE IF NOT EXISTS orb_signals (
+  id                     INTEGER PRIMARY KEY,
+  instance_id            TEXT    NOT NULL,                -- SHA256(ORB_APP_ID) prefix; one-way, no PII
+  repo_hash              TEXT    NOT NULL,                -- HMAC(repo, instance secret); collector can't reverse
+  pr_hash                TEXT    NOT NULL,                -- HMAC(repo#pr, instance secret)
+  gate_verdict           TEXT,                            -- the prediction: 'merge' | 'close' | 'hold'
+  outcome                TEXT    NOT NULL CHECK (outcome IN ('merged', 'closed')),  -- realized ground truth
+  reversal_flag          TEXT    NOT NULL DEFAULT 'none' CHECK (reversal_flag IN ('none', 'reopened', 'reverted')),
+  gate_reasoncode_bucket TEXT,                            -- low-cardinality category, bucketed at source
+  time_to_close_ms       INTEGER,                         -- decision -> close cycle time (nullable)
+  decision_timestamp     TEXT,                            -- when the gate decided
+  outcome_timestamp      TEXT,                            -- when the PR resolved
+  sent_at                TEXT,
+  received_at            TEXT    NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  UNIQUE (instance_id, repo_hash, pr_hash)                -- dedup unit: one row per PR per instance, upserted
+);
+-- Supports the (verdict, outcome, reversal) confusion-matrix rollups that are the whole point of the table.
+CREATE INDEX IF NOT EXISTS orb_signals_calibration ON orb_signals (instance_id, gate_verdict, outcome, reversal_flag);
+CREATE INDEX IF NOT EXISTS orb_signals_instance ON orb_signals (instance_id, received_at);
+
+-- Export watermark per self-host instance — replaces orb_events.exported_at (review_audit is append-only, so
+-- the exporter tracks the latest exported event time and ships only newer resolved PRs / reversals).
+CREATE TABLE IF NOT EXISTS orb_export_cursor (
+  instance_hash    TEXT PRIMARY KEY,
+  last_exported_at TEXT NOT NULL DEFAULT '2000-01-01T00:00:00Z',
+  updated_at       TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
@@ -123,6 +123,7 @@ import {
 } from "../github/commands";
 import { handleGitHubWebhook } from "../github/webhook";
 import { handleOrbIngest } from "../orb/ingest";
+import { computeFleetAnalytics } from "../orb/analytics";
 import { handleMcpRequest } from "../mcp/server";
 import { buildOpenApiSpec } from "../openapi/spec";
 import { generateSignalSnapshots } from "../queue/processors";
@@ -2863,9 +2864,9 @@ export function createApp() {
 
   app.post("/v1/github/webhook", handleGitHubWebhook);
 
-  // Gittensory Orb (#1219) — central collector. Receives anonymized outcome signal batches
-  // from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender;
-  // dedup is enforced via UNIQUE(instance_id, pr_hash) in orb_signals.
+  // Gittensory Orb (#1255) — central fleet-calibration collector. Receives anonymized, reversal-aware
+  // 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);
     if (!body) return c.json({ error: "invalid_request" }, 400);
@@ -2874,6 +2875,14 @@ export function createApp() {
     return c.json(result, 200);
   });
 
+  // Fleet calibration analytics over the collected orb_signals — gate accuracy (precision / FP / reversal /
+  // cycle-time) aggregated median-robustly across the self-host fleet. Owner-only: bearer-gated by the
+  // `/v1/internal/*` middleware (INTERNAL_JOB_TOKEN). `?days=` windows the lookback (default 90).
+  app.get("/v1/internal/fleet/analytics", async (c) => {
+    const days = parsePositiveInt(c.req.query("days")) ?? 90;
+    return c.json(await computeFleetAnalytics(c.env, { windowDays: days }));
+  });
+
   // 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

@@ -0,0 +1,151 @@
+// Gittensory Orb (#1255) — fleet calibration ANALYTICS. Reads the anonymized orb_signals collected from
+// self-hosted instances and derives gate-accuracy metrics across the fleet. Aggregation is median/percentile
+// (never mean) so a single instance contributing fabricated data cannot move the fleet numbers.
+
+const MIN_DECIDED = 5; // an instance needs at least this many decided PRs to count toward the fleet median
+const OUTLIER_BAND = 0.25; // |instance precision − fleet median| beyond this flags the instance
+
+/** Per-instance confusion-matrix cell as stored. */
+interface Cell {
+  instance_id: string;
+  verdict: string | null;
+  outcome: string;
+  reversal_flag: string;
+  n: number;
+}
+
+export interface InstanceMetrics {
+  instanceId: string;
+  decided: number;
+  mergePrecision: number | null; // P(merged & not reverted | gate said merge)
+  closePrecision: number | null; // P(closed & not reopened | gate said close)
+  fpRate: number | null; // P(closed or reverted | gate said merge) — gate approved, it was wrong
+  fnRate: number | null; // P(merged or reopened | gate said close) — gate blocked, it was wrong
+  reversalRate: number; // share of decided PRs a human reversed
+}
+
+export interface FleetAnalytics {
+  windowDays: number;
+  instanceCount: number; // instances meeting MIN_DECIDED
+  fleet: {
+    mergePrecision: number | null;
+    closePrecision: number | null;
+    fpRate: number | null;
+    reversalRate: number | null;
+    cycleP50Ms: number | null;
+    cycleP95Ms: number | null;
+  };
+  instances: InstanceMetrics[];
+  outliers: Array<{ instanceId: string; metric: string; value: number; fleetMedian: number }>;
+}
+
+function median(xs: number[]): number | null {
+  if (xs.length === 0) return null;
+  const s = [...xs].sort((a, b) => a - b);
+  const mid = Math.floor(s.length / 2);
+  return s.length % 2 === 0 ? (s[mid - 1]! + s[mid]!) / 2 : s[mid]!;
+}
+
+function percentile(sorted: number[], p: number): number | null {
+  if (sorted.length === 0) return null;
+  const idx = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length));
+  return sorted[idx]!;
+}
+
+/** Fold the confusion-matrix cells for one instance into accuracy metrics (reversals count as the gate
+ *  being wrong: a reverted merge is a false positive; a reopened close is a false negative). */
+function foldInstance(instanceId: string, cells: Cell[]): InstanceMetrics {
+  let wouldMerge = 0, mergeConfirmed = 0, mergeFalse = 0;
+  let wouldClose = 0, closeConfirmed = 0, closeFalse = 0;
+  let reversals = 0, decided = 0;
+  for (const c of cells) {
+    decided += c.n;
+    if (c.reversal_flag !== "none") reversals += c.n;
+    if (c.verdict === "merge") {
+      wouldMerge += c.n;
+      if (c.outcome === "merged" && c.reversal_flag !== "reverted") mergeConfirmed += c.n;
+      else mergeFalse += c.n;
+    } else if (c.verdict === "close") {
+      wouldClose += c.n;
+      if (c.outcome === "closed" && c.reversal_flag !== "reopened") closeConfirmed += c.n;
+      else closeFalse += c.n;
+    }
+  }
+  return {
+    instanceId,
+    decided,
+    mergePrecision: wouldMerge > 0 ? mergeConfirmed / wouldMerge : null,
+    closePrecision: wouldClose > 0 ? closeConfirmed / wouldClose : null,
+    fpRate: wouldMerge > 0 ? mergeFalse / wouldMerge : null,
+    fnRate: wouldClose > 0 ? closeFalse / wouldClose : null,
+    reversalRate: reversals / decided, // decided ≥ 1 (the instance has at least one cell)
+  };
+}
+
+/** Compute fleet calibration analytics over the collected orb_signals within the window. Fail-safe → empty. */
+export async function computeFleetAnalytics(env: Env, opts: { windowDays?: number } = {}): Promise<FleetAnalytics> {
+  const windowDays = Number.isFinite(opts.windowDays) && (opts.windowDays as number) > 0 ? Math.min(opts.windowDays as number, 365) : 90;
+  // Date-only cutoff (like computeGateEval) so it compares correctly whether received_at is ISO ('…T…Z')
+  // or SQLite's CURRENT_TIMESTAMP space format ('YYYY-MM-DD HH:MM:SS').
+  const cutoff = new Date(Date.now() - windowDays * 86_400_000).toISOString().slice(0, 10);
+
+  let cells: Cell[] = [];
+  let cycle: number[] = [];
+  try {
+    const matrix = await env.DB
+      .prepare(
+        `SELECT instance_id, gate_verdict AS verdict, outcome, reversal_flag, COUNT(*) AS n
+         FROM orb_signals WHERE received_at >= ?
+         GROUP BY instance_id, gate_verdict, outcome, reversal_flag`,
+      )
+      .bind(cutoff)
+      .all<Cell>();
+    cells = matrix.results ?? [];
+    const cy = await env.DB
+      .prepare(`SELECT time_to_close_ms AS ms FROM orb_signals WHERE received_at >= ? AND time_to_close_ms IS NOT NULL ORDER BY time_to_close_ms`)
+      .bind(cutoff)
+      .all<{ ms: number }>();
+    cycle = (cy.results ?? []).map((r) => r.ms);
+  } catch {
+    return { windowDays, instanceCount: 0, fleet: { mergePrecision: null, closePrecision: null, fpRate: null, reversalRate: null, cycleP50Ms: null, cycleP95Ms: null }, instances: [], outliers: [] };
+  }
+
+  // Group cells by instance, fold each.
+  const byInstance = new Map<string, Cell[]>();
+  for (const c of cells) {
+    const list = byInstance.get(c.instance_id) ?? [];
+    list.push(c);
+    byInstance.set(c.instance_id, list);
+  }
+  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);
+  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));
+
+  const outliers: FleetAnalytics["outliers"] = [];
+  if (fleetMergeP !== null) {
+    for (const i of eligible) {
+      if (i.mergePrecision !== null && Math.abs(i.mergePrecision - fleetMergeP) > OUTLIER_BAND) {
+        outliers.push({ instanceId: i.instanceId, metric: "mergePrecision", value: i.mergePrecision, fleetMedian: fleetMergeP });
+      }
+    }
+  }
+
+  return {
+    windowDays,
+    instanceCount: eligible.length,
+    fleet: {
+      mergePrecision: fleetMergeP,
+      closePrecision: fleetCloseP,
+      fpRate: median(nums((i) => i.fpRate)),
+      reversalRate: median(nums((i) => i.reversalRate)),
+      cycleP50Ms: percentile(cycle, 50),
+      cycleP95Ms: percentile(cycle, 95),
+    },
+    instances,
+    outliers,
+  };
+}
@@ -1,18 +1,24 @@
-// Gittensory Orb (#1219) — central collector receiver.
-// Accepts anonymized outcome signal batches from self-hosted instances running exportOrbBatch.
-// No raw repo names, owner identifiers, or PR content is accepted or stored — only HMAC-anonymized
-// hashes + aggregate outcome metadata (verdict, timing).
+// Gittensory Orb (#1255) — central fleet-calibration collector receiver.
+// Accepts anonymized, reversal-aware outcome batches from self-hosted instances (exportOrbBatch).
+// No raw repo names, owner identifiers, commit SHAs, or PR content — only HMAC-anonymized hashes +
+// aggregate calibration metadata (verdict, outcome, reversal, bucketed reason, cycle time).
 
 const MAX_BATCH = 500;
 const VALID_OUTCOMES = new Set(["merged", "closed"]);
+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
 
 interface OrbIngestEvent {
   repo_hash: string;
   pr_hash: string;
-  outcome: string;
   gate_verdict?: string | null;
+  outcome: string;
+  reversal_flag?: string | null;
+  gate_reasoncode_bucket?: string | null;
   time_to_close_ms?: number | null;
-  created_at?: string | null;
+  decision_timestamp?: string | null;
+  outcome_timestamp?: string | null;
 }
 
 interface OrbIngestPayload {
@@ -22,6 +28,13 @@ interface OrbIngestPayload {
 
 export type OrbIngestResult = { accepted: number } | { error: string };
 
+/** Clamp a sender-supplied cycle time to a plausible range; null for anything implausible/absent. */
+function clampCycleMs(value: unknown): number | null {
+  if (typeof value !== "number" || !Number.isFinite(value)) return null;
+  if (value < MIN_CYCLE_MS || value > MAX_CYCLE_MS) return null;
+  return Math.round(value);
+}
+
 export async function handleOrbIngest(body: string, db: D1Database): Promise<OrbIngestResult> {
   let payload: unknown;
   try {
@@ -54,21 +67,31 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise<Orb
       continue;
     }
 
+    // Untrusted-input normalization: whitelist reversal_flag, clamp cycle time, coerce the rest to null.
+    const reversal = typeof event.reversal_flag === "string" && VALID_REVERSALS.has(event.reversal_flag) ? event.reversal_flag : "none";
+
     try {
+      // OR REPLACE: a re-exported PR (e.g. one that later gained a reversal) upserts the freshest outcome
+      // on the (instance_id, repo_hash, pr_hash) dedup key.
       const result = await db
         .prepare(
-          `INSERT OR IGNORE INTO orb_signals
-           (instance_id, repo_hash, pr_hash, outcome, gate_verdict, time_to_close_ms, sent_at)
-           VALUES (?, ?, ?, ?, ?, ?, ?)`,
+          `INSERT OR REPLACE INTO orb_signals
+           (instance_id, repo_hash, pr_hash, gate_verdict, outcome, reversal_flag, gate_reasoncode_bucket,
+            time_to_close_ms, decision_timestamp, outcome_timestamp, sent_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
         )
         .bind(
           instance_id,
           event.repo_hash,
           event.pr_hash,
-          event.outcome,
           typeof event.gate_verdict === "string" ? event.gate_verdict : null,
-          typeof event.time_to_close_ms === "number" ? event.time_to_close_ms : null,
-          typeof event.created_at === "string" ? event.created_at : null,
+          event.outcome,
+          reversal,
+          typeof event.gate_reasoncode_bucket === "string" ? event.gate_reasoncode_bucket : null,
+          clampCycleMs(event.time_to_close_ms),
+          typeof event.decision_timestamp === "string" ? event.decision_timestamp : null,
+          typeof event.outcome_timestamp === "string" ? event.outcome_timestamp : null,
+          typeof event.outcome_timestamp === "string" ? event.outcome_timestamp : null,
         )
         .run();
       if (result.meta.changes > 0) accepted++;