fix(orb): live fleet reversal + reuse telemetry for the homepage hero metrics

migrations/0176_orb_signals_superseded_reversal.sql

@@ -0,0 +1,37 @@
+-- #8820: admit the successor-merge reversal (#8166's reversal_superseded) into the fleet-calibration signal.
+--
+-- The exporter/ingest whitelist now carries reversal_flag='superseded', but orb_signals' CHECK constraint
+-- (0060) still pins the column to ('none','reopened','reverted') — the ingest's INSERT OR REPLACE would hit
+-- the constraint and its best-effort catch would SILENTLY skip the row, so the fleet's published
+-- reversalRate stayed pinned at 0 no matter how many supersessions the instances detected. SQLite can't
+-- alter a CHECK, so rebuild the table with the widened constraint, preserving existing rows (they are
+-- continuously re-exported telemetry, but keeping them avoids a multi-day fleet-metrics blackout while
+-- instances re-fill).
+
+CREATE TABLE orb_signals_new (
+  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', 'superseded')),
+  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
+);
+
+INSERT INTO orb_signals_new (id, instance_id, repo_hash, pr_hash, gate_verdict, outcome, reversal_flag, gate_reasoncode_bucket, time_to_close_ms, decision_timestamp, outcome_timestamp, sent_at, received_at)
+  SELECT id, instance_id, repo_hash, pr_hash, gate_verdict, outcome, reversal_flag, gate_reasoncode_bucket, time_to_close_ms, decision_timestamp, outcome_timestamp, sent_at, received_at
+  FROM orb_signals;
+
+DROP TABLE orb_signals;
+ALTER TABLE orb_signals_new RENAME TO orb_signals;
+
+-- Recreate the indexes the rename does not carry over (same shapes as 0060).
+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);

migrations/0177_orb_reuse_counters.sql

@@ -0,0 +1,17 @@
+-- #8820 (reuse-rate half): live fleet source for the homepage "AI work reused" trend.
+--
+-- The public reuse-rate trend reads github_app.%cache_hit/%cache_miss audit events from THIS worker's own
+-- ledger — which froze at the self-host cutover (last event 2026-06-29), so the latest weekly buckets fell
+-- under the publish floor and the hero tile rendered a dash next to a decaying sparkline. The live signal
+-- (133k+ cache events and growing) accrues on the self-hosted instances; this table receives their
+-- day-bucketed, instance-level aggregate counters (counts only — no repos, no PRs, no content), exported on
+-- the same hourly tick as orb_signals and folded into the public trend for REGISTERED instances only (the
+-- same trust anchor computeFleetAnalytics uses).
+CREATE TABLE IF NOT EXISTS orb_reuse_counters (
+  instance_id TEXT    NOT NULL,
+  day         TEXT    NOT NULL, -- YYYY-MM-DD (UTC)
+  hits        INTEGER NOT NULL DEFAULT 0,
+  misses      INTEGER NOT NULL DEFAULT 0,
+  received_at TEXT    NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (instance_id, day) -- senders re-export a rolling window; the upsert keeps the freshest counts
+);

scripts/check-schema-drift.ts

@@ -49,6 +49,7 @@ export const RAW_SQL_ONLY_TABLES: Set<string> = new Set([
   "orb_instances",
   "orb_pr_outcomes",
   "orb_relay_failures",
+  "orb_reuse_counters",
   "orb_signals",
   "orb_webhook_events",
   "override_audit",

src/orb/analytics.ts

@@ -107,7 +107,9 @@ export function percentile(sorted: number[], p: number): number | null {
 }
 
 /** 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).
+ *  being wrong: a reverted merge is a false positive; a reopened OR superseded close is a false negative —
+ *  `superseded` (#8166) is the one-shot culture's dominant "bot was wrong" shape: the closed PR's work later
+ *  merged via a successor PR, so the close is disconfirmed exactly like a literal reopen).
  *
  *  Exported for the federated bundle export (#1970, src/orb/federated-bundle.ts): a bundle publishes this
  *  instance's own precision for #6481 to compare against the peer median computed here, so both sides MUST use
@@ -126,7 +128,7 @@ export function foldInstance(instanceId: string, cells: Cell[]): InstanceMetrics
       else mergeFalse += c.n;
     } else if (c.verdict === "close") {
       wouldClose += c.n;
-      if (c.outcome === "closed" && c.reversal_flag !== "reopened") closeConfirmed += c.n;
+      if (c.outcome === "closed" && c.reversal_flag !== "reopened" && c.reversal_flag !== "superseded") closeConfirmed += c.n;
       else closeFalse += c.n;
     }
   }

src/orb/ingest.ts

@@ -9,7 +9,7 @@ const MAX_HASH_CHARS = 128;
 const MAX_BUCKET_CHARS = 64;
 const MAX_VERDICT_CHARS = 32;
 const VALID_OUTCOMES = new Set(["merged", "closed"]);
-const VALID_REVERSALS = new Set(["none", "reopened", "reverted"]);
+const VALID_REVERSALS = new Set(["none", "reopened", "reverted", "superseded"]);
 const MIN_CYCLE_MS = 1_000; // <1s is implausible
 const MAX_CYCLE_MS = 31_536_000_000; // >1y is implausible
 
@@ -75,6 +75,23 @@ interface OrbIngestPayload {
   // #4933: optional -- an older self-host build that hasn't upgraded yet simply omits this, and the
   // instance's stored health stays whatever it last was (or NULL/unknown on first contact).
   health?: { ok: boolean };
+  // #8820: optional day-bucketed cache hit/miss aggregates for the public "AI work reused" trend (counts
+  // only). A rolling window re-sent every tick; upserted per (instance, day). Absent from older builds.
+  reuse_counters?: Array<{ day?: unknown; hits?: unknown; misses?: unknown }>;
+}
+
+/** Rolling-window bound: the sender exports ~70 days (REUSE_COUNTER_WINDOW_DAYS); anything wildly larger is
+ *  a hostile payload padding the loop, not a real export. */
+const MAX_REUSE_COUNTER_DAYS = 400;
+const MAX_REUSE_COUNT = 10_000_000; // per-day per-instance ceiling — beyond this is fabrication, not telemetry
+const REUSE_DAY_PATTERN = /^\d{4}-\d{2}-\d{2}$/;
+
+/** Clamp a sender-supplied per-day counter to a plausible non-negative integer; null rejects the row. */
+function clampReuseCount(value: unknown): number | null {
+  if (typeof value !== "number" || !Number.isFinite(value)) return null;
+  const rounded = Math.round(value);
+  if (rounded < 0 || rounded > MAX_REUSE_COUNT) return null;
+  return rounded;
 }
 
 export type OrbIngestResult = { accepted: number } | { error: string };
@@ -191,5 +208,31 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise<Orb
     }
   }
 
+  // #8820: day-bucketed reuse counters (optional field; older builds omit it). Every row is
+  // whitelist-validated (strict YYYY-MM-DD day, clamped non-negative counts) and upserted on
+  // (instance_id, day) — the sender re-exports a rolling window each tick, so REPLACE keeps the freshest
+  // counts idempotently. Malformed rows are skipped one-by-one (same best-effort posture as events above);
+  // a malformed CONTAINER (non-array) is ignored rather than failing the outcome batch riding alongside.
+  const reuseCounters = (payload as OrbIngestPayload).reuse_counters;
+  if (Array.isArray(reuseCounters)) {
+    for (const counter of reuseCounters.slice(0, MAX_REUSE_COUNTER_DAYS)) {
+      const day = typeof counter?.day === "string" && REUSE_DAY_PATTERN.test(counter.day) ? counter.day : null;
+      const hits = clampReuseCount(counter?.hits);
+      const misses = clampReuseCount(counter?.misses);
+      if (day === null || hits === null || misses === null) continue;
+      try {
+        await db
+          .prepare(
+            `INSERT OR REPLACE INTO orb_reuse_counters (instance_id, day, hits, misses, received_at)
+             VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)`,
+          )
+          .bind(instance_id, day, hits, misses)
+          .run();
+      } catch {
+        // best-effort — a counter hiccup must never fail the outcome batch
+      }
+    }
+  }
+
   return { accepted };
 }

src/selfhost/orb-collector.ts

@@ -17,6 +17,7 @@
 // can never de-anonymize).
 import { createHash, createHmac } from "node:crypto";
 import { generateAnonSecret, hmacAnonymize } from "../../packages/loopover-engine/src/telemetry/anonymize.js";
+import { AI_REVIEW_REUSE_EVENT_TYPES } from "../services/public-reuse-rate-trend";
 import { incr } from "./metrics";
 
 /** Key under which the per-instance anonymization secret is persisted in system_flags. */
@@ -33,6 +34,7 @@ interface FleetRow {
   outcome_at: string;
   reverted: number; // 0|1
   reopened: number; // 0|1
+  superseded: number; // 0|1
   event_at: string; // max(outcome_at, latest reversal time) — the export watermark unit
 }
 
@@ -41,7 +43,7 @@ interface FleetEvent {
   pr_hash: string;
   gate_verdict: string | null;
   outcome: string;
-  reversal_flag: "none" | "reopened" | "reverted";
+  reversal_flag: "none" | "reopened" | "reverted" | "superseded";
   gate_reasoncode_bucket: string;
   time_to_close_ms: number | null;
   decision_timestamp: string | null;
@@ -52,6 +54,45 @@ interface OrbExportPayload {
   instance_id: string;
   events: FleetEvent[];
   health?: { ok: boolean };
+  /** #8820: day-bucketed cache hit/miss aggregates for the public "AI work reused" trend. Counts only —
+   *  no repos, no PRs, no content. A rolling window re-sent every tick (the collector upserts per day),
+   *  so the field is self-healing and needs no cursor. Omitted when the window has no cache events. */
+  reuse_counters?: Array<{ day: string; hits: number; misses: number }>;
+}
+
+/** Rolling window the reuse counters cover — the public trend renders 8 weeks; the extra buffer keeps the
+ *  oldest visible bucket complete across week boundaries and export lag. */
+export const REUSE_COUNTER_WINDOW_DAYS = 70;
+
+/** Day-bucketed reuse counters from the local audit_events ledger. Same event population as
+ *  loadReuseRateDayRows (public-reuse-rate-trend.ts) — the LIKE convention plus ai_review's three
+ *  non-suffix-conforming reuse variants — so the fleet fold can never drift from the own-ledger count.
+ *  substr(created_at, 1, 10) is the portable day bucket (runs on the SQLite AND Postgres backends, and
+ *  tolerates this ledger's mixed 'YYYY-MM-DD hh:mm:ss' / ISO-with-T timestamp formats). */
+const REUSE_COUNTER_QUERY = `
+  SELECT substr(created_at, 1, 10) AS day,
+         SUM(CASE WHEN event_type LIKE 'github_app.%cache_hit' OR event_type IN (${AI_REVIEW_REUSE_EVENT_TYPES.map(() => "?").join(", ")}) THEN 1 ELSE 0 END) AS hits,
+         SUM(CASE WHEN event_type LIKE 'github_app.%cache_miss' THEN 1 ELSE 0 END) AS misses
+    FROM audit_events
+   WHERE (event_type LIKE 'github_app.%cache_hit' OR event_type LIKE 'github_app.%cache_miss' OR event_type IN (${AI_REVIEW_REUSE_EVENT_TYPES.map(() => "?").join(", ")}))
+     AND created_at >= ?
+   GROUP BY day`;
+
+/** Read the rolling reuse-counter window; fail-safe → [] (a counter hiccup must never block the outcome
+ *  export riding the same tick). */
+async function loadReuseCounters(db: D1Database, nowMs: number): Promise<Array<{ day: string; hits: number; misses: number }>> {
+  const sinceIso = new Date(nowMs - REUSE_COUNTER_WINDOW_DAYS * 86_400_000).toISOString();
+  try {
+    // The rows already carry exactly the export shape; SUM(CASE…) over a GROUP BY never yields SQL NULL,
+    // so no per-field fallback is needed (a missing table / failed query is the catch below).
+    const result = await db
+      .prepare(REUSE_COUNTER_QUERY)
+      .bind(...AI_REVIEW_REUSE_EVENT_TYPES, ...AI_REVIEW_REUSE_EVENT_TYPES, sinceIso)
+      .all<{ day: string; hits: number; misses: number }>();
+    return result.results;
+  } catch {
+    return [];
+  }
 }
 
 /** Stable instance identifier (hash of the Orb/App ID — no PII). A brokered instance holds no App id, so its
@@ -125,16 +166,18 @@ const FLEET_QUERY = `
     SELECT target_id,
       MAX(CASE WHEN event_type = 'reversal_reverted' THEN 1 ELSE 0 END) AS reverted,
       MAX(CASE WHEN event_type = 'reversal_reopened' THEN 1 ELSE 0 END) AS reopened,
+      MAX(CASE WHEN event_type = 'reversal_superseded' THEN 1 ELSE 0 END) AS superseded,
       MAX(created_at) AS rev_at
     FROM review_audit
-    WHERE event_type IN ('reversal_reverted', 'reversal_reopened')
+    WHERE event_type IN ('reversal_reverted', 'reversal_reopened', 'reversal_superseded')
     GROUP BY target_id
   )
-  SELECT project, target_id, verdict, reasoncode, decided_at, outcome, outcome_at, reverted, reopened, event_at
+  SELECT project, target_id, verdict, reasoncode, decided_at, outcome, outcome_at, reverted, reopened, superseded, event_at
   FROM (
     SELECT gd.project AS project, gd.target_id AS target_id, gd.verdict AS verdict, gd.reasoncode AS reasoncode,
            gd.decided_at AS decided_at, po.outcome AS outcome, po.outcome_at AS outcome_at,
            COALESCE(rev.reverted, 0) AS reverted, COALESCE(rev.reopened, 0) AS reopened,
+           COALESCE(rev.superseded, 0) AS superseded,
            CASE WHEN rev.rev_at IS NOT NULL AND rev.rev_at > po.outcome_at THEN rev.rev_at ELSE po.outcome_at END AS event_at
     FROM gd
     JOIN po ON gd.target_id = po.target_id
@@ -193,6 +236,10 @@ export async function exportOrbBatch(db: D1Database, batchSize = 200, fetchFn: t
   // otherwise, exactly as before, nothing new means nothing to do.
   if ((!results || results.length === 0) && healthOk === undefined) return 0;
 
+  // #8820: the reuse counters ride the same POST as the outcome events (same tick, same signature). Loaded
+  // AFTER the early "nothing to send" return above, so a truly idle tick still costs nothing extra.
+  const reuseCounters = await loadReuseCounters(db, Date.now());
+
   const payload: OrbExportPayload = {
     instance_id: instance,
     /* v8 ignore next -- D1's .all() always returns a `results` array (possibly empty), never omits the field;
@@ -202,13 +249,17 @@ export async function exportOrbBatch(db: D1Database, batchSize = 200, fetchFn: t
       pr_hash: anonymize ? hmacAnonymize(r.target_id, secret) : r.target_id,
       gate_verdict: r.verdict,
       outcome: r.outcome,
-      reversal_flag: r.reverted ? "reverted" : r.reopened ? "reopened" : "none",
+      // Priority mirrors signal strength: an explicit revert PR beats a reopen beats the successor-merge
+      // heuristic (#8166's reversal_superseded — the one-shot culture's dominant real "bot was wrong" shape,
+      // which this export previously DROPPED entirely, silently pinning the fleet's reversalRate at 0).
+      reversal_flag: r.reverted ? "reverted" : r.reopened ? "reopened" : r.superseded ? "superseded" : "none",
       gate_reasoncode_bucket: bucketReasonCode(r.reasoncode),
       time_to_close_ms: cycleTimeMs(r.decided_at, r.outcome_at),
       decision_timestamp: r.decided_at,
       outcome_timestamp: r.outcome_at,
     })),
     ...(healthOk !== undefined ? { health: { ok: healthOk } } : {}),
+    ...(reuseCounters.length > 0 ? { reuse_counters: reuseCounters } : {}),
   };
 
   const body = JSON.stringify(payload);

src/services/public-reuse-rate-trend.ts

@@ -26,8 +26,10 @@ export const PUBLIC_REUSE_RATE_TREND_WEEKS = 8;
 export const MIN_REUSE_RATE_TREND_SAMPLE = 5;
 
 /** ai_review reuse events that don't follow the `_cache_hit` suffix convention but are the SAME "avoided a
- *  redundant AI call" signal -- each one means the review pass reused a prior state instead of re-running. */
-const AI_REVIEW_REUSE_EVENT_TYPES = ["github_app.ai_review_frozen_reuse", "github_app.ai_review_paused_reuse", "github_app.ai_review_one_shot_reuse"] as const;
+ *  redundant AI call" signal -- each one means the review pass reused a prior state instead of re-running.
+ *  Exported for the self-host reuse-counter export (orb-collector.ts, #8820) so both sides count the exact
+ *  same event population -- a drifted copy there would silently skew the published fleet rate. */
+export const AI_REVIEW_REUSE_EVENT_TYPES = ["github_app.ai_review_frozen_reuse", "github_app.ai_review_paused_reuse", "github_app.ai_review_one_shot_reuse"] as const;
 
 export type PublicReuseRateTrendWeek = {
   /** UTC Monday (YYYY-MM-DD) that starts the bucket. */
@@ -104,11 +106,35 @@ async function loadReuseRateDayRows(env: Env, projects: string[], sinceIso: stri
   return rows.map((row) => ({ day: row.day, hits: row.hits ?? 0, misses: row.misses ?? 0 }));
 }
 
+/** Day-bucketed reuse counters exported by REGISTERED self-hosted instances (orb_reuse_counters, #8820) --
+ *  the LIVE side of this trend: the own-ledger audit_events below froze at the self-host cutover, so recent
+ *  weeks otherwise fall under the publish floor and the homepage tile renders a dash. Registration is the
+ *  same trust anchor computeFleetAnalytics uses -- open ingest stores everyone's counters, but an
+ *  unregistered stranger can't move the published rate. Deliberately NOT scoped by the own-ledger repo
+ *  allowlist (counters are instance-level counts only -- no repos to scope by), matching the fleet-accuracy
+ *  fold's own unconditional-regardless-of-allowlist behavior in public-stats.ts. */
+async function loadFleetReuseDayRows(env: Env, sinceIso: string): Promise<DayRow[]> {
+  const rows = await safeAll<{ day: string; hits: number; misses: number }>(
+    env,
+    `SELECT c.day AS day, SUM(c.hits) AS hits, SUM(c.misses) AS misses
+       FROM orb_reuse_counters c
+       JOIN orb_instances i ON i.instance_id = c.instance_id AND i.registered = 1
+      WHERE c.day >= ?
+      GROUP BY c.day`,
+    sinceIso.slice(0, 10),
+  );
+  /* v8 ignore next -- same guard shape as loadReuseRateDayRows above: SUM over a GROUP BY day of NOT NULL
+   *  integer columns always yields a defined integer, never SQL NULL; kept for defense against a future
+   *  query-shape change. */
+  return rows.map((row) => ({ day: row.day, hits: row.hits ?? 0, misses: row.misses ?? 0 }));
+}
+
 /** Assemble the public reuse-rate trend from the SAME live audit_events ledger every instrumented capability
- *  already writes to. */
+ *  already writes to, plus the registered fleet's exported day counters (#8820) -- buildPublicReuseRateTrend
+ *  sums overlapping days from both sources into the same weekly buckets. */
 export async function loadPublicReuseRateTrend(env: Env, nowMs: number = Date.now()): Promise<PublicReuseRateTrendWeek[]> {
   const projects = publicStatsProjects(env);
   const sinceIso = new Date(Date.parse(isoWeekStart(nowMs)) - (PUBLIC_REUSE_RATE_TREND_WEEKS - 1) * MS_PER_WEEK).toISOString();
-  const dayRows = await loadReuseRateDayRows(env, projects, sinceIso);
-  return buildPublicReuseRateTrend(dayRows, nowMs);
+  const [ownRows, fleetRows] = await Promise.all([loadReuseRateDayRows(env, projects, sinceIso), loadFleetReuseDayRows(env, sinceIso)]);
+  return buildPublicReuseRateTrend([...ownRows, ...fleetRows], nowMs);
 }