feat(orb): add a typed config-push write path to the relay pipeline

migrations/0168_orb_relay_pending_kind.sql

@@ -0,0 +1,5 @@
+-- Typed config-push write path (#7522, piece 1 of #4902's design): a discriminator so the pull-drain loop can
+-- stop assuming every orb_relay_pending row is a GitHub webhook. Default preserves identical behavior for every
+-- existing and future GitHub-webhook row -- dispatch behavior itself is unchanged by this migration (see the
+-- companion dispatch-side issue #7523).
+ALTER TABLE orb_relay_pending ADD COLUMN kind TEXT NOT NULL DEFAULT 'github_webhook';

src/api/routes.ts

@@ -156,7 +156,9 @@ import { handleOrbWebhook } from "../orb/webhook";
 import { handleOrbOAuthCallback } from "../orb/oauth";
 import { brokerOrbToken, isOrbBrokerEnabled, issueOrbEnrollment } from "../orb/broker";
 import {
+  enqueueConfigPushRelay,
   MAX_ORB_RELAY_REGISTER_BODY_BYTES,
+  pruneRelayPending,
   pullRelayPending,
   readOrbRelayRegisterBody,
   registerValidatedOrbRelay,
@@ -1013,6 +1015,21 @@ const killSwitchUpdateSchema = z
   })
   .strict();
 
+// Config-push write path (#7522, piece 1 of #4902's design): an operator-addressed Orb-operational notice
+// (enrollment lifecycle, capability announcement, deprecation notice) -- explicit installationIds target list
+// only, no percentage/canary selector (no rollout-percentage primitive exists in this codebase to build one on
+// top of; out of scope here). pushId doubles as the idempotency key (see enqueueConfigPushRelay's deliveryId
+// derivation), so it's constrained to the same safe-identifier shape as commandFeedbackSchema's answerId above.
+const configPushSchema = z
+  .object({
+    installationIds: z.array(z.number().int().positive()).min(1).max(500),
+    pushId: z.string().min(1).max(120).regex(/^[A-Za-z0-9_.:-]+$/),
+    message: z.string().min(1).max(500),
+    capability: z.string().min(1).max(120).optional(),
+    deprecatesAt: z.string().datetime().optional(),
+  })
+  .strict();
+
 const digestSubscriptionSchema = z
   .object({
     email: z.string().email().max(320),
@@ -1907,6 +1924,46 @@ export function createApp() {
     return c.json({ ok: true, ...verified });
   });
 
+  // Config-push write path (#7522, piece 1 of #4902's 3-piece design): an operator pushes a typed, addressed
+  // Orb-operational notice (capability announcement, deprecation notice, enrollment lifecycle) to an explicit
+  // list of installations, landing in the SAME orb_relay_pending queue the GitHub-webhook relay already uses
+  // (kind = 'config_push' -- see enqueueConfigPushRelay, src/orb/relay.ts). Write side only; the companion
+  // dispatch-side issue (#7523) is how a self-host container's drain loop tells this apart from a webhook row
+  // before touching raw_body. Scope boundary: Orb's own operational state ONLY -- never auto-applies anything
+  // that overrides an operator's own .loopover.yml/DB settings.
+  //
+  // Deliberately under /v1/app/*, NOT /v1/internal/* despite that being this issue's illustrative example path:
+  // the /v1/internal/* prefix's own middleware requires a bearer INTERNAL_JOB_TOKEN and (per requiresApiToken's
+  // explicit `/v1/internal/` exclusion) never even resolves a session identity for it -- canSessionAccessPath is
+  // never consulted -- which would make requireAppRole's session-role branch unreachable dead code for a
+  // control-panel caller. /v1/app/* is where requireAppRole's session-based gate is actually meaningful,
+  // matching the kill-switch endpoint above exactly (a bearer INTERNAL_JOB_TOKEN/api-token caller still passes
+  // requireAppRole's own non-session branch either way).
+  app.post("/v1/app/fleet/config-push", async (c) => {
+    const forbidden = await requireAppRole(c, ["operator"]);
+    if (forbidden) return forbidden;
+    const identity = await authenticateRequestIdentity(c);
+    /* v8 ignore next -- requireAppRole already rejects an unauthenticated caller before this handler runs. */
+    if (!identity) return c.json({ error: "unauthorized" }, 401);
+    const body = await c.req.json().catch(() => null);
+    const parsed = configPushSchema.safeParse(body);
+    if (!parsed.success) return c.json({ error: "invalid_config_push", issues: parsed.error.issues }, 400);
+    const { installationIds, ...payload } = parsed.data;
+    // #7611 review fix: prune ONCE for the whole request, not once per target -- enqueueConfigPushRelay no
+    // longer prunes itself (see its own doc comment) precisely so a 500-installation fan-out below can't turn
+    // into 500 redundant global TTL-prune scans/deletes against the shared orb_relay_pending table.
+    await pruneRelayPending(c.env);
+    await Promise.all(installationIds.map((installationId) => enqueueConfigPushRelay(c.env, installationId, payload)));
+    await recordAuditEvent(c.env, {
+      eventType: "operator.config_push_enqueued",
+      actor: identity.actor,
+      targetKey: `config_push#${parsed.data.pushId}`,
+      outcome: "completed",
+      metadata: { installationCount: installationIds.length, capability: payload.capability ?? null },
+    });
+    return c.json({ ok: true, pushId: parsed.data.pushId, installationCount: installationIds.length });
+  });
+
   // #5672 post-merge incident report, internal-operator side: same reporting path as the repo-scoped customer
   // route (POST /v1/repos/:owner/:repo/pulls/:number/incident-reports), for an operator filing on a customer's
   // behalf. Not scoped to one repo's session, so repoFullName/pullNumber travel in the body instead of the path.

src/db/schema.ts

@@ -873,6 +873,11 @@ export const orbRelayPending = sqliteTable(
     rawBody: text("raw_body").notNull(),
     createdAt: text("created_at").notNull().$defaultFn(() => nowIso()),
     coalesceKey: text("coalesce_key"),
+    // Discriminator (#7522): 'github_webhook' (default, every pre-existing row) vs 'config_push' (an
+    // operator-addressed capability/deprecation notice, never a GitHub payload -- see src/orb/relay.ts's
+    // enqueueConfigPushRelay). The dispatch side (#7523) branches on this before treating raw_body as a
+    // GitHubWebhookPayload.
+    kind: text("kind").notNull().default("github_webhook"),
   },
   (table) => ({
     installation: index("idx_orb_relay_pending_install").on(table.installationId, table.createdAt),

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,15 +264,20 @@ 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
 // can span hundreds of rows, so the log payload itself must stay bounded.
 const RELAY_DROP_LOG_SAMPLE_SIZE = 20;
 
-/** Drop pull-mode rows that exceeded the raw-body retention window, even if their engine never polls. */
-async function pruneRelayPending(env: Env): Promise<number> {
+/** Drop pull-mode rows that exceeded the raw-body retention window, even if their engine never polls. Exported
+ *  (#7611 review fix) so a multi-target caller like the config-push route can prune ONCE per request instead of
+ *  once per target -- see enqueueConfigPushRelay's own doc comment below for why it no longer prunes itself. */
+export async function pruneRelayPending(env: Env): Promise<number> {
   // Sampled BEFORE the delete: a plain DELETE reports only a count, and the rows are unrecoverable once gone.
   // Same WHERE predicate as the delete below, capped so a large backlog can't inflate the log payload.
   const expiring = await env.DB
@@ -340,6 +345,49 @@ export async function enqueueRelayPending(
     .run();
 }
 
+export type ConfigPushPayload = {
+  pushId: string;
+  message: string;
+  capability?: string | undefined;
+  deprecatesAt?: string | undefined;
+};
+
+/** An Orb-operational notice (#7522, piece 1 of #4902's 3-piece design) addressed to one installation — NOT a
+ *  GitHub webhook, so unlike enqueueRelayPending above this skips coalesce-key derivation entirely (that logic
+ *  assumes `rawBody` parses as a GitHubWebhookPayload; a config_push payload never does). Still shares the same
+ *  pending queue as the webhook path — same per-installation backlog cap — since both kinds drain through the
+ *  same pull loop (`kind` is how the drain side, #7523, tells them apart before touching `rawBody`).
+ *  `deliveryId` is derived from `pushId` + `installationId` (not caller-supplied) so re-posting the same push
+ *  is idempotent per target, matching every other ON CONFLICT DO NOTHING write in this file.
+ *
+ *  Deliberately does NOT prune here (#7611 review fix): enqueueRelayPending prunes per-call because it's
+ *  invoked once per individually-arriving webhook, but this function is fanned out over up to 500
+ *  `installationIds` from a SINGLE request (POST /v1/app/fleet/config-push) — pruning inside it would turn one
+ *  request into up to 500 redundant global TTL-prune scans/deletes against the shared table. The caller prunes
+ *  ONCE before the fan-out instead (see that route's own handler). */
+export async function enqueueConfigPushRelay(env: Env, installationId: number, payload: ConfigPushPayload): Promise<void> {
+  const deliveryId = `config-push:${payload.pushId}:${installationId}`;
+  await env.DB
+    .prepare(
+      "INSERT INTO orb_relay_pending (delivery_id, installation_id, event_name, raw_body, kind) VALUES (?, ?, 'config_push', ?, 'config_push') ON CONFLICT(delivery_id) DO NOTHING",
+    )
+    .bind(deliveryId, installationId, JSON.stringify(payload))
+    .run();
+  await env.DB
+    .prepare(
+      `DELETE FROM orb_relay_pending
+       WHERE installation_id = ?
+         AND delivery_id IN (
+           SELECT delivery_id FROM orb_relay_pending
+           WHERE installation_id = ?
+           ORDER BY created_at DESC, delivery_id DESC
+           LIMIT -1 OFFSET ?
+         )`,
+    )
+    .bind(installationId, installationId, RELAY_PENDING_MAX_PER_INSTALLATION)
+    .run();
+}
+
 function relayPendingCoalesceKey(eventName: string, rawBody: string): string | null {
   try {
     return githubWebhookCoalesceKey(
@@ -381,10 +429,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