feat(orb): dispatch config_push relay rows separately from the GitHub-webhook queue

src/orb/broker-client.ts

@@ -242,7 +242,7 @@ export async function drainOrbRelay(
   env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined },
   ack: string[] = [],
   fetchImpl: typeof fetch = fetch,
-): Promise<{ deliveryId: string; eventName: string; rawBody: string }[]> {
+): Promise<{ deliveryId: string; eventName: string; rawBody: string; kind: string }[]> {
   if (!isOrbBrokerMode(env)) return [];
   try {
     const base = orbBrokerBaseUrl(env);
@@ -253,11 +253,14 @@ export async function drainOrbRelay(
       signal: AbortSignal.timeout(30_000),
     });
     if (!res.ok) throw new Error(`orb_relay_drain_http_${res.status}`);
-    const body = (await res.json()) as { events?: Array<{ deliveryId?: unknown; eventName?: unknown; rawBody?: unknown }> };
-    const out: { deliveryId: string; eventName: string; rawBody: string }[] = [];
+    const body = (await res.json()) as { events?: Array<{ deliveryId?: unknown; eventName?: unknown; rawBody?: unknown; kind?: unknown }> };
+    const out: { deliveryId: string; eventName: string; rawBody: string; kind: string }[] = [];
     for (const e of body.events ?? []) {
       if (typeof e.deliveryId === "string" && typeof e.eventName === "string" && typeof e.rawBody === "string") {
-        out.push({ deliveryId: e.deliveryId, eventName: e.eventName, rawBody: e.rawBody });
+        // #7523: an older Orb server predating the `kind` column omits the field entirely -- default to
+        // 'github_webhook' (the only kind that ever existed before this) so a rolling deploy never
+        // misroutes an old-shaped event.
+        out.push({ deliveryId: e.deliveryId, eventName: e.eventName, rawBody: e.rawBody, kind: typeof e.kind === "string" ? e.kind : "github_webhook" });
         continue;
       }
       // #zero-trace-webhook-loss: a batch entry missing/mistyping one of the three required fields was

src/orb/relay.ts

@@ -264,7 +264,10 @@ const RELAY_PENDING_BATCH_SIZE = 50;
 const RELAY_PENDING_TTL_HOURS = 24;
 const RELAY_PENDING_MAX_PER_INSTALLATION = 500;
 
-export type RelayPendingEvent = { deliveryId: string; eventName: string; rawBody: string };
+// #7523: 'kind' rides along on every pulled event so a self-host drain client can tell a config_push notice
+// apart from a GitHub webhook BEFORE treating rawBody as a GitHubWebhookPayload — see
+// src/selfhost/monitored-work.ts's drainOrbRelayWithMonitor.
+export type RelayPendingEvent = { deliveryId: string; eventName: string; rawBody: string; kind: string };
 
 // Bulk-delete drop logs (this function and retryFailedRelays below) sample at most this many rows' identifying
 // info — an operator needs to see WHICH installation(s) lost events without a direct DB query, but a busy prune
@@ -419,10 +422,10 @@ export async function pullRelayPending(
       : RELAY_PENDING_BATCH_SIZE;
   const limit = Math.min(requestedLimit, RELAY_PENDING_BATCH_SIZE);
   const { results } = await env.DB
-    .prepare("SELECT delivery_id, event_name, raw_body FROM orb_relay_pending WHERE installation_id = ? ORDER BY created_at, delivery_id LIMIT ?")
+    .prepare("SELECT delivery_id, event_name, raw_body, kind FROM orb_relay_pending WHERE installation_id = ? ORDER BY created_at, delivery_id LIMIT ?")
     .bind(installationId, limit)
-    .all<{ delivery_id: string; event_name: string; raw_body: string }>();
-  return results.map((r) => ({ deliveryId: r.delivery_id, eventName: r.event_name, rawBody: r.raw_body }));
+    .all<{ delivery_id: string; event_name: string; raw_body: string; kind: string }>();
+  return results.map((r) => ({ deliveryId: r.delivery_id, eventName: r.event_name, rawBody: r.raw_body, kind: r.kind }));
 }
 
 /** Record a failed relay forward in the retry queue. Idempotent on delivery_id — a duplicate insert (e.g. from a

src/selfhost/metrics.ts

@@ -108,6 +108,7 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [
   ["loopover_orb_relay_register_consecutive_failures", { help: "Current consecutive orb relay registration failure streak, reset to 0 on any success.", type: "gauge" }],
   ["loopover_orb_relay_drain_seconds_since_last", { help: "Seconds since the pull-mode orb relay drain loop last completed successfully, or -1 if never (or in push mode).", type: "gauge" }],
   ["loopover_orb_webhook_total", { help: "Orb webhook outcomes.", type: "counter" }],
+  ["loopover_orb_config_push_received_total", { help: "Config-push relay rows received and logged by the pull-drain loop (#7523).", type: "counter" }],
   ["loopover_ai_requests_total", { help: "AI provider request outcomes.", type: "counter" }],
   ["loopover_ai_cost_usd_total", { help: "Estimated AI provider cost in USD.", type: "counter" }],
   ["loopover_ai_input_tokens_total", { help: "AI provider input tokens consumed.", type: "counter" }],

src/selfhost/monitored-work.ts

@@ -7,6 +7,11 @@ export type OrbRelayEvent = {
   deliveryId: string;
   eventName: string;
   rawBody: string;
+  // #7523: 'github_webhook' (the only kind before this) vs 'config_push' (an operator-addressed notice,
+  // never a GitHub payload -- see handleConfigPushRelayEvent below). Anything other than the literal string
+  // 'config_push' is treated as a webhook, so an unrecognized/old-shaped value degrades to the existing,
+  // unchanged behavior rather than a new failure mode.
+  kind: string;
 };
 
 export type OrbRelayDrainState = {
@@ -36,6 +41,20 @@ function orbRelayMetricEvent(eventName: string): string {
   return ORB_RELAY_METRIC_EVENTS.has(eventName) ? eventName : "other";
 }
 
+/** Receive-and-surface only (#7523, v1 scope per #4902's design comment): a config_push row's rawBody is a
+ *  JSON-stringified ConfigPushPayload (src/orb/relay.ts), never a GitHubWebhookPayload -- this deliberately
+ *  does NOT hand it to enqueueWebhookByEnv. No capability-toggle or config-mutation side effect is implemented
+ *  here; that's explicit, separately-scoped follow-up work, not something this function grows into silently. */
+function handleConfigPushRelayEvent(ev: OrbRelayEvent, log: (line: string) => void): void {
+  let payload: unknown = null;
+  try {
+    payload = JSON.parse(ev.rawBody);
+  } catch {
+    payload = null; // surfaced as-is below -- a malformed payload is still worth a visible trace, not a throw
+  }
+  log(JSON.stringify({ event: "orb_config_push_received", deliveryId: ev.deliveryId, payload }));
+}
+
 export async function runScheduledLoopWithMonitor<T>(
   cron: string,
   scheduled: () => T | Promise<T>,
@@ -86,6 +105,30 @@ export async function drainOrbRelayWithMonitor(args: {
         result: events.length > 0 ? "events" : "empty",
       });
       for (const ev of events) {
+        // #7523: a config_push row is Orb-operational state, never a GitHub payload -- branch BEFORE
+        // args.enqueue (which unconditionally does JSON.parse(rawBody) as GitHubWebhookPayload) so it's
+        // never misinterpreted. Everything else (any value other than the literal 'config_push', including
+        // 'github_webhook' and an old-shaped/missing kind) falls through to the existing path below,
+        // byte-for-byte unchanged.
+        if (ev.kind === "config_push") {
+          try {
+            handleConfigPushRelayEvent(ev, args.log ?? console.log);
+            incr("loopover_orb_config_push_received_total");
+          } catch (error) {
+            // Same per-event isolation stance as the webhook path below: don't ack, let the relay redeliver.
+            console.error(
+              JSON.stringify({
+                level: "error",
+                event: "orb_config_push_handler_threw",
+                deliveryId: ev.deliveryId,
+                error: error instanceof Error ? error.message : String(error),
+              }),
+            );
+            continue;
+          }
+          args.state.pendingAck.push(ev.deliveryId);
+          continue;
+        }
         // #audit-orb-relay-enqueue-isolation: an enqueue can throw uncaught (e.g. a D1/Postgres write failure
         // inside recordWebhookEvent, not just the anticipated failures enqueueWebhookByEnv already returns as a
         // string result) -- that must not abort the REST of this batch, or every event after the failing one

test/integration/orb-relay.test.ts

@@ -812,7 +812,7 @@ describe("pullRelayPending", () => {
     await enqueueRelayPending(e, { deliveryId: "other", installationId: 9999, eventName: "pull_request", rawBody: "{}" }); // a different install
     const events = await pullRelayPending(e, 9701);
     expect(events.map((ev) => ev.deliveryId)).toEqual(["a", "b"]); // only this install, ordered
-    expect(events[0]).toEqual({ deliveryId: "a", eventName: "pull_request", rawBody: "{}" });
+    expect(events[0]).toEqual({ deliveryId: "a", eventName: "pull_request", rawBody: "{}", kind: "github_webhook" });
   });
 
   it("ACK-deletes only the named rows, scoped to this installation (can't ack another install's row)", async () => {
@@ -1039,7 +1039,7 @@ describe("POST /v1/orb/relay/pull", () => {
     await enqueueRelayPending(e, { deliveryId: "pe-1", installationId: 8502, eventName: "pull_request", rawBody: '{"x":1}' });
     const res = await app.request("/v1/orb/relay/pull", { method: "POST", headers: { authorization: `Bearer ${secret}` } }, e);
     expect(res.status).toBe(200);
-    expect(await res.json()).toEqual({ events: [{ deliveryId: "pe-1", eventName: "pull_request", rawBody: '{"x":1}' }] });
+    expect(await res.json()).toEqual({ events: [{ deliveryId: "pe-1", eventName: "pull_request", rawBody: '{"x":1}', kind: "github_webhook" }] });
   });
 
   it("passes a valid ack[] through (acked rows are deleted), and tolerates a non-string ack entry", async () => {
@@ -1049,18 +1049,18 @@ describe("POST /v1/orb/relay/pull", () => {
     await enqueueRelayPending(e, { deliveryId: "ack-b", installationId: 8503, eventName: "issues", rawBody: "{}" });
     const res = await app.request("/v1/orb/relay/pull", { method: "POST", headers: { authorization: `Bearer ${secret}` }, body: JSON.stringify({ ack: ["ack-a", 123] }) }, e); // 123 filtered out
     expect(res.status).toBe(200);
-    expect(await res.json()).toEqual({ events: [{ deliveryId: "ack-b", eventName: "issues", rawBody: "{}" }] }); // ack-a removed
+    expect(await res.json()).toEqual({ events: [{ deliveryId: "ack-b", eventName: "issues", rawBody: "{}", kind: "github_webhook" }] }); // ack-a removed
   });
 
   it("tolerates a non-array ack field and an unparseable body (no ack, returns events)", async () => {
     const e = brokeredEnv();
     const secret = await enroll(e, 8504);
     await enqueueRelayPending(e, { deliveryId: "keep-1", installationId: 8504, eventName: "pull_request", rawBody: "{}" });
     const nonArray = await app.request("/v1/orb/relay/pull", { method: "POST", headers: { authorization: `Bearer ${secret}` }, body: JSON.stringify({ ack: "nope" }) }, e); // ack not an array
-    expect(await nonArray.json()).toEqual({ events: [{ deliveryId: "keep-1", eventName: "pull_request", rawBody: "{}" }] });
+    expect(await nonArray.json()).toEqual({ events: [{ deliveryId: "keep-1", eventName: "pull_request", rawBody: "{}", kind: "github_webhook" }] });
     const bad = await app.request("/v1/orb/relay/pull", { method: "POST", headers: { authorization: `Bearer ${secret}` }, body: "{not json" }, e); // unparseable → catch
     expect(bad.status).toBe(200);
-    expect(await bad.json()).toEqual({ events: [{ deliveryId: "keep-1", eventName: "pull_request", rawBody: "{}" }] });
+    expect(await bad.json()).toEqual({ events: [{ deliveryId: "keep-1", eventName: "pull_request", rawBody: "{}", kind: "github_webhook" }] });
   });
 
   it("REGRESSION (#4995, GITTENSORY-1C): a DB error inside pullRelayPending returns a clean 503 broker_error instead of an unhandled framework 500", async () => {

test/unit/orb-broker-client.test.ts

@@ -369,16 +369,16 @@ describe("drainOrbRelay (pull-mode drain)", () => {
     const { fetchImpl, calls } = captureFetch(
       Response.json({
         events: [
-          { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}" },
-          { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" },
+          { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}", kind: "config_push" },
+          { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" }, // no kind (older Orb) → defaults below
           { deliveryId: "bad", eventName: "x" }, // no rawBody → filtered out
         ],
       }),
     );
     const out = await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, ["prev-1"], fetchImpl);
     expect(out).toEqual([
-      { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}" },
-      { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" },
+      { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}", kind: "config_push" },
+      { deliveryId: "d2", eventName: "check_suite", rawBody: "{}", kind: "github_webhook" }, // #7523 fallback
     ]);
     expect(calls[0]?.url).toBe("https://api.loopover.ai/v1/orb/relay/pull");
     expect((calls[0]?.init?.headers as Record<string, string>).authorization).toBe("Bearer s");