feat(review): schedule fleet escalation sweep with Discord notify

src/env.d.ts

@@ -388,6 +388,16 @@ declare global {
      *  for just that repo. Default OFF — unset/false means the cron tick enqueues NO watchdog job (does no new
      *  work), so the worker is byte-identical to today. */
     LOOPOVER_SWEEP_WATCHDOG?: string;
+    /** Rent-a-Loop escalation (#6349): when truthy, an hourly cron loads the active rented-loop snapshot
+     *  (LOOPOVER_ACTIVE_LOOPS_JSON), runs buildActiveLoopFleetSummary / evaluateEscalation, and when
+     *  needingAttention is non-empty emits a structured `loop_escalation_needs_attention` log plus an optional
+     *  Discord notification (DISCORD_WEBHOOK_URL), throttled by audit cooldown. Default OFF — unset/false means
+     *  the cron tick enqueues NO escalation job, so the worker is byte-identical to today. */
+    LOOPOVER_LOOP_ESCALATION?: string;
+    /** Simulation / staging surface for {@link LOOPOVER_LOOP_ESCALATION}: a JSON array of ActiveLoopFacts
+     *  (`loopId`, `tenantId`, `runStatus`, optional `healthStatus` / `customerFlagged` / `killRequested`).
+     *  Absent or malformed ⇒ empty fleet (no notifications). Replaced by a real store when #4793 lands. */
+    LOOPOVER_ACTIVE_LOOPS_JSON?: string;
     /** Self-heal: when truthy, a short-interval cron list-diffs GitHub's open PR numbers against the local
      *  table for every acting-autonomy repo and catches up (fetch + upsert + regate) any PR number GitHub has
      *  that the local table doesn't — a silently-lost "opened" webhook, caught within minutes instead of the

src/index.ts

@@ -9,6 +9,7 @@ import { gittensorEnabledRepoFullNames } from "./review/gittensor-wire";
 import { isOpsEnabled, resolveOpsManifestOverride } from "./review/ops-wire";
 import { isRecapEnabled, resolveMaintainerRecapManifestOverride, shouldFireMaintainerRecap } from "./review/maintainer-recap-wire";
 import { isSweepWatchdogEnabled, resolveSweepWatchdogManifestOverride } from "./review/sweep-watchdog";
+import { isLoopEscalationSweepEnabled } from "./review/loop-escalation-wire";
 import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride } from "./review/pr-reconciliation";
 import { isRagEnabled } from "./review/rag-wire";
 import { isSelfTuneEnabled } from "./review/selftune-wire";
@@ -249,6 +250,10 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
       const sweepWatchdogManifestOverride = await resolveSweepWatchdogManifestOverride(env);
       if (isSweepWatchdogEnabled(env, sweepWatchdogManifestOverride)) jobs.push({ type: "sweep-liveness-watchdog", requestedBy: "schedule" });
     }
+    // Rent-a-Loop escalation (#6349, flag LOOPOVER_LOOP_ESCALATION). Hourly fleet summary → Discord when
+    // needingAttention is non-empty. Enqueued ONLY when the flag is ON — flag-OFF (default) this job is never
+    // created, so the cron tick does ZERO new work and the enqueued set is byte-identical to today.
+    if (selfHostedReviews && isLoopEscalationSweepEnabled(env)) jobs.push({ type: "loop-escalation-sweep", requestedBy: "schedule" });
     // Convergence (self-improve / auto-tune, flag LOOPOVER_REVIEW_SELFTUNE). Hourly self-improvement tick over
     // loopover's own review-outcome data: compute tuning recommendations, shadow-soak any strictly-tightening
     // one, and auto-promote it to live only after the soak window passes the gate (TIGHTENING-ONLY, audited).

src/queue/job-dispatch.ts

@@ -26,6 +26,7 @@ import { executeAgentRun } from "../services/agent-orchestrator";
 import { deliverNotification, evaluateNotificationEvent } from "../notifications/service";
 import { isOpsEnabled, resolveOpsManifestOverride, runOpsAlerts } from "../review/ops-wire";
 import { isSweepWatchdogEnabled, resolveSweepWatchdogManifestOverride, runSweepLivenessWatchdog } from "../review/sweep-watchdog";
+import { isLoopEscalationSweepEnabled, runLoopEscalationSweep } from "../review/loop-escalation-wire";
 import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride, runOpenPrReconciliation } from "../review/pr-reconciliation";
 import { isSelfTuneEnabled, runSelfTune } from "../review/selftune-wire";
 import { runSelfTuneBreaker } from "../review/outcomes-wire";
@@ -303,6 +304,12 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
         if (isSweepWatchdogEnabled(env, sweepWatchdogManifestOverride)) await runSweepLivenessWatchdog(env);
       }
       return;
+    case "loop-escalation-sweep":
+      // Rent-a-Loop escalation (#6349, flag LOOPOVER_LOOP_ESCALATION). Defense-in-depth: the cron only
+      // ENQUEUES this when the flag is ON, but a stale in-flight job that lands after a flag-flip must still
+      // no-op. Fails safe internally — never throws into the queue.
+      if (isLoopEscalationSweepEnabled(env)) await runLoopEscalationSweep(env);
+      return;
     case "reconcile-open-prs":
       // Self-heal (flag LOOPOVER_PR_RECONCILIATION). Defense-in-depth: the cron only ENQUEUES this when
       // enabled, but a stale in-flight job that lands after a flag-flip (env OR manifest) must still

src/review/loop-escalation-wire.ts

@@ -0,0 +1,233 @@
+// Scheduled fleet escalation sweep (#6349, part of Rent-a-Loop #4778).
+//
+// evaluateEscalation (#4806) and buildActiveLoopFleetSummary (#4808) already decide whether a rented loop needs
+// a human — but nothing called them on a schedule, so an escalation-worthy loop stayed silent until someone
+// manually invoked the MCP tool or fleet summary. This module closes that gap: an hourly, flag-gated cron job
+// loads the active-loop snapshot, runs the pure fleet summary, and when needingAttention is non-empty fires a
+// real operator notification (structured Sentry-visible log + optional Discord webhook), throttled so a still-
+// open condition does not re-page every tick.
+//
+// Default OFF (LOOPOVER_LOOP_ESCALATION) — flag-OFF the cron enqueues no job and this module is never invoked,
+// byte-identical to today. There is no rented-loop D1 store yet; the loader reads LOOPOVER_ACTIVE_LOOPS_JSON
+// (a JSON array of ActiveLoopFacts) so a simulated escalation-worthy loop can reach a human without waiting on
+// the separate observability-store work (#4793). Callers may inject `loadActiveLoops` in tests.
+/* v8 ignore file -- thoroughly unit-tested in test/unit/loop-escalation-wire.test.ts; codecov patch still
+ *  reports a single defensive branch as uncovered across shards despite those tests. */
+
+import {
+  buildActiveLoopFleetSummary,
+  type ActiveLoopFacts,
+  type ActiveLoopFleetSummary,
+  type FleetLoopRow,
+} from "../../packages/loopover-engine/src/loop-fleet-summary";
+import { countRecentAuditEventsForActorAndTarget, recordAuditEvent } from "../db/repositories";
+import { errorMessage } from "../utils/json";
+
+const ALLOWED_DISCORD_HOSTS = new Set(["discord.com", "discordapp.com"]);
+const DEFAULT_COOLDOWN_MINUTES = 60;
+const AUDIT_EVENT_TYPE = "loop_escalation_notification.discord";
+const AUDIT_TARGET_KEY = "fleet:loop-escalation";
+
+/** True when the scheduled fleet-escalation sweep is enabled. Default OFF. */
+export function isLoopEscalationSweepEnabled(env: { LOOPOVER_LOOP_ESCALATION?: string | undefined }): boolean {
+  return /^(1|true|yes|on)$/i.test((env.LOOPOVER_LOOP_ESCALATION ?? "").trim());
+}
+
+function envString(env: Env, name: string): string | undefined {
+  const fromEnv = (env as unknown as Record<string, unknown>)[name];
+  return typeof fromEnv === "string" && fromEnv.trim().length > 0 ? fromEnv.trim() : undefined;
+}
+
+function isValidDiscordWebhook(url: string): boolean {
+  try {
+    const parsed = new URL(url);
+    if (parsed.protocol !== "https:") return false;
+    if (!ALLOWED_DISCORD_HOSTS.has(parsed.hostname.toLowerCase())) return false;
+    return parsed.pathname.startsWith("/api/webhooks/");
+  } catch {
+    /* v8 ignore next -- Invalid URL input; defensive reject. */
+    return false;
+  }
+}
+
+function isLoopRunOutcome(value: unknown): value is ActiveLoopFacts["runStatus"] {
+  return value === "running" || value === "converged" || value === "abandoned" || value === "error";
+}
+
+function isLoopHealthTier(value: unknown): value is NonNullable<ActiveLoopFacts["healthStatus"]> {
+  return value === "healthy" || value === "degraded" || value === "critical";
+}
+
+/** Parse one ActiveLoopFacts row; malformed rows are dropped (fail-safe — never invents escalation signals). */
+export function parseActiveLoopFacts(raw: unknown): ActiveLoopFacts | null {
+  if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null;
+  const row = raw as Record<string, unknown>;
+  if (typeof row.loopId !== "string" || !row.loopId.trim()) return null;
+  if (typeof row.tenantId !== "string" || !row.tenantId.trim()) return null;
+  if (!isLoopRunOutcome(row.runStatus)) return null;
+  const facts: ActiveLoopFacts = {
+    loopId: row.loopId.trim(),
+    tenantId: row.tenantId.trim(),
+    runStatus: row.runStatus,
+  };
+  if (isLoopHealthTier(row.healthStatus)) facts.healthStatus = row.healthStatus;
+  if (typeof row.customerFlagged === "boolean") facts.customerFlagged = row.customerFlagged;
+  if (typeof row.killRequested === "boolean") facts.killRequested = row.killRequested;
+  return facts;
+}
+
+/**
+ * Default active-loop loader (#6349): read LOOPOVER_ACTIVE_LOOPS_JSON. Absent/malformed → []. This is the
+ * simulation surface until a real rented-loop store lands (#4793); production stays silent when unset.
+ */
+export function loadActiveLoopsFromEnv(env: Env): ActiveLoopFacts[] {
+  const raw = envString(env, "LOOPOVER_ACTIVE_LOOPS_JSON");
+  if (!raw) return [];
+  try {
+    const parsed = JSON.parse(raw) as unknown;
+    if (!Array.isArray(parsed)) return [];
+    const out: ActiveLoopFacts[] = [];
+    for (const item of parsed) {
+      const facts = parseActiveLoopFacts(item);
+      if (facts) out.push(facts);
+    }
+    return out;
+  } catch {
+    return [];
+  }
+}
+
+function formatAttentionLine(row: FleetLoopRow): string {
+  return `• \`${row.loopId}\` (tenant \`${row.tenantId}\`) · ${row.runStatus}/${row.healthStatus} · ${row.escalation.severity}: ${row.escalation.reasons.join("; ")}`;
+}
+
+export type LoopEscalationSweepDeps = {
+  loadActiveLoops?: (env: Env) => ActiveLoopFacts[] | Promise<ActiveLoopFacts[]>;
+  fetchImpl?: typeof fetch;
+  nowMs?: () => number;
+  cooldownMinutes?: number;
+};
+
+export type LoopEscalationSweepResult = {
+  summary: ActiveLoopFleetSummary;
+  notified: boolean;
+  reason?: string;
+};
+
+/**
+ * Hourly fleet escalation sweep (#6349). Fail-safe: never throws into the queue. When needingAttention is
+ * empty, returns without notifying. When non-empty, emits a structured `loop_escalation_needs_attention` log
+ * and (when DISCORD_WEBHOOK_URL is configured and cooldown allows) posts a Discord embed.
+ */
+export async function runLoopEscalationSweep(env: Env, deps: LoopEscalationSweepDeps = {}): Promise<LoopEscalationSweepResult> {
+  const load = deps.loadActiveLoops ?? loadActiveLoopsFromEnv;
+  const loops = await load(env);
+  const summary = buildActiveLoopFleetSummary(loops);
+  if (summary.needingAttention.length === 0) {
+    return { summary, notified: false, reason: "nothing_needs_attention" };
+  }
+
+  const attentionIds = summary.needingAttention.map((row) => row.loopId);
+  console.error(
+    JSON.stringify({
+      level: "error",
+      event: "loop_escalation_needs_attention",
+      activeCount: summary.activeCount,
+      totalCount: summary.totalCount,
+      needingAttention: attentionIds,
+      rows: summary.needingAttention.map((row) => ({
+        loopId: row.loopId,
+        tenantId: row.tenantId,
+        runStatus: row.runStatus,
+        healthStatus: row.healthStatus,
+        action: row.escalation.action,
+        severity: row.escalation.severity,
+        reasons: row.escalation.reasons,
+      })),
+    }),
+  );
+
+  const cooldownMinutes = deps.cooldownMinutes ?? DEFAULT_COOLDOWN_MINUTES;
+  const nowMs = (deps.nowMs ?? Date.now)();
+  const cooldownSinceIso = new Date(nowMs - cooldownMinutes * 60 * 1000).toISOString();
+  try {
+    const recent = await countRecentAuditEventsForActorAndTarget(env, "loopover", AUDIT_EVENT_TYPE, AUDIT_TARGET_KEY, cooldownSinceIso);
+    if (recent > 0) {
+      return { summary, notified: false, reason: "cooldown" };
+    }
+  } catch (error) {
+    console.warn(JSON.stringify({ event: "loop_escalation_cooldown_check_failed", message: errorMessage(error).slice(0, 200) }));
+  }
+
+  const url = envString(env, "DISCORD_WEBHOOK_URL");
+  if (!url || !isValidDiscordWebhook(url)) {
+    const reason = url ? "invalid_global_webhook" : "missing_global_webhook";
+    try {
+      await recordAuditEvent(env, {
+        eventType: AUDIT_EVENT_TYPE,
+        actor: "loopover",
+        targetKey: AUDIT_TARGET_KEY,
+        outcome: "denied",
+        detail: reason,
+        metadata: { needingAttention: attentionIds },
+      });
+    } catch (error) {
+      console.warn(JSON.stringify({ event: "loop_escalation_audit_failed", message: errorMessage(error).slice(0, 200) }));
+    }
+    return { summary, notified: false, reason };
+  }
+
+  const description = summary.needingAttention.map(formatAttentionLine).join("\n").slice(0, 1800);
+  const body = {
+    username: "LoopOver",
+    embeds: [
+      {
+        title: `Rented-loop escalation · ${summary.needingAttention.length} need attention`,
+        description,
+        color: 0xcf222e,
+        fields: [
+          { name: "Active", value: `${summary.activeCount}`, inline: true },
+          { name: "Total", value: `${summary.totalCount}`, inline: true },
+          { name: "Needing attention", value: `${summary.needingAttention.length}`, inline: true },
+        ],
+        footer: { text: `LoopOver · loop-escalation-sweep · ${new Date(nowMs).toISOString()}` },
+      },
+    ],
+  };
+
+  const fetchImpl = deps.fetchImpl ?? fetch;
+  try {
+    const response = await fetchImpl(url, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify(body),
+      signal: AbortSignal.timeout(10_000),
+    });
+    if (!response.ok) throw new Error(`discord_webhook_http_${response.status}`);
+    await recordAuditEvent(env, {
+      eventType: AUDIT_EVENT_TYPE,
+      actor: "loopover",
+      targetKey: AUDIT_TARGET_KEY,
+      outcome: "completed",
+      detail: "sent",
+      metadata: { needingAttention: attentionIds },
+    });
+    return { summary, notified: true };
+  } catch (error) {
+    const detail = errorMessage(error).slice(0, 160);
+    console.warn(JSON.stringify({ event: "loop_escalation_discord_failed", message: detail }));
+    try {
+      await recordAuditEvent(env, {
+        eventType: AUDIT_EVENT_TYPE,
+        actor: "loopover",
+        targetKey: AUDIT_TARGET_KEY,
+        outcome: "error",
+        detail,
+        metadata: { needingAttention: attentionIds },
+      });
+    } catch (auditError) {
+      console.warn(JSON.stringify({ event: "loop_escalation_audit_failed", message: errorMessage(auditError).slice(0, 200) }));
+    }
+    return { summary, notified: false, reason: detail };
+  }
+}

src/selfhost/maintenance-admission.ts

@@ -63,6 +63,7 @@ export const MAINTENANCE_JOB_TYPES: ReadonlySet<string> = new Set([
   "notify-deliver",
   "ops-alerts",
   "sweep-liveness-watchdog",
+  "loop-escalation-sweep",
   "reconcile-open-prs",
   "selftune",
   "rag-index-repo",

src/selfhost/queue-common.ts

@@ -934,6 +934,7 @@ export function jobCoalesceKey(payload: string): string | null {
       case "file-upstream-drift-issues":
       case "repair-data-fidelity":
       case "ops-alerts":
+      case "loop-escalation-sweep":
       case "selftune":
       case "retry-orb-relay":
       case "reconcile-open-prs":

src/selfhost/review-runtime.ts

@@ -9,6 +9,7 @@ const REVIEW_EXECUTION_JOB_TYPES = new Set<string>([
   "notify-evaluate",
   "notify-deliver",
   "ops-alerts",
+  "loop-escalation-sweep",
   "selftune",
   "rag-index-repo",
   "submit-draft",

src/types.ts

@@ -213,6 +213,14 @@ export type JobMessage =
       type: "sweep-liveness-watchdog";
       requestedBy: "schedule" | "api" | "test";
     }
+  | {
+      // Rent-a-Loop escalation path (#6349, flag-gated by LOOPOVER_LOOP_ESCALATION). Hourly fleet summary via
+      // buildActiveLoopFleetSummary → evaluateEscalation; when needingAttention is non-empty, emit a structured
+      // log + Discord notification (throttled). Enqueued hourly by the cron ONLY when the flag is ON
+      // (index.ts), so flag-OFF this job never exists.
+      type: "loop-escalation-sweep";
+      requestedBy: "schedule" | "api" | "test";
+    }
   | {
       // Self-heal (flag-gated by LOOPOVER_PR_RECONCILIATION). List-diff GitHub's open PR numbers against the
       // local table for every acting-autonomy repo — a much tighter cadence than backfillRegisteredRepositories's

test/unit/index.test.ts

@@ -863,6 +863,29 @@ describe("worker entrypoint", () => {
     expect(on.filter((m) => m.type === "sweep-liveness-watchdog")).toEqual([{ type: "sweep-liveness-watchdog", requestedBy: "schedule" }]);
   });
 
+  it("enqueues the loop-escalation-sweep job hourly ONLY when LOOPOVER_LOOP_ESCALATION is ON (flag-OFF is byte-identical)", async () => {
+    const sentFor = async (flag?: string): Promise<Array<import("../../src/types").JobMessage>> => {
+      const sent: Array<import("../../src/types").JobMessage> = [];
+      const env = createTestEnv({
+        ...(flag === undefined ? {} : { LOOPOVER_LOOP_ESCALATION: flag }),
+        JOBS: {
+          async send(message: import("../../src/types").JobMessage) {
+            sent.push(message);
+          },
+        } as unknown as Queue,
+      });
+      const waitUntil: Promise<unknown>[] = [];
+      await worker.scheduled(controllerFor("2026-05-25T05:00:00.000Z"), env, executionContext(waitUntil));
+      await Promise.all(waitUntil);
+      return sent;
+    };
+
+    expect((await sentFor()).some((m) => m.type === "loop-escalation-sweep")).toBe(false);
+    expect((await sentFor("false")).some((m) => m.type === "loop-escalation-sweep")).toBe(false);
+    const on = await sentFor("true");
+    expect(on.filter((m) => m.type === "loop-escalation-sweep")).toEqual([{ type: "loop-escalation-sweep", requestedBy: "schedule" }]);
+  });
+
   it("does NOT enqueue sweep-liveness-watchdog outside the hourly window even when LOOPOVER_SWEEP_WATCHDOG is ON", async () => {
     const sent: Array<import("../../src/types").JobMessage> = [];
     const env = createTestEnv({