feat(jobs): tell admins when a system job keeps failing

A full rollup backfill that failed on every boot showed only on the
admin page's card of failing background jobs. From the account's side
the only symptom was empty dashboard cards, and nothing pointed the
operator at the card (#1298).

A new hourly evaluation reads the newest terminal runs of every queue
whose failure has no account-facing surface (the system audience in
subsystem-surface.ts) from pg-boss's own ledger. A failed run adds its
attempts, retries included, and the first successful run ends the
streak. Three failed attempts in a row send every admin a SYSTEM_ALERT
once, a reminder at most once a day while the queue keeps failing, and
one notice when it runs successfully again: the decision the backup
health notices use, with the same compare-and-set claim, so two workers
never send a notice twice.

The memory is one row per announced queue in system_job_alert_notices
(migration 0490). The notice names the queue, the streak and the last
error, and opens the admin status page.

src/__tests__/error-level-discipline-guard.test.ts

@@ -163,6 +163,11 @@ const ALLOWED: Readonly<Record<string, { count: number; reason: string }>> = {
     count: 1,
     reason: "No /proc/diskstats outside Linux; see captureHostMetric.",
   },
+  "lib/jobs/system-job-alert.ts::readTerminalRuns": {
+    count: 1,
+    reason:
+      "No pgboss schema on a web-only deployment; null means no queue to ask, and the evaluation then judges nothing.",
+  },
   "lib/jobs/job-failures.ts::readActiveJobsStartedBefore": {
     count: 1,
     reason:

src/lib/data-wipe/wipe-plan.ts

@@ -358,6 +358,8 @@ export const INSTANCE_SCOPED: Readonly<Record<string, string>> = {
     "the monthly restore drill's last verdict for the instance's off-host bucket; says nothing about any account",
   RecoveryReadinessNotice:
     "per backup signal, what the admins were last told; a fact about the instance's backups, not about any account",
+  SystemJobAlertNotice:
+    "per background queue, what the admins were last told about its failures; a fact about the instance's jobs, not about any account",
   RecoveryKitConfirmation:
     "the instance's recovery kit confirmation (key ids and fingerprints); the confirming admin is a plain id, not a relation",
   DatabaseBackupRun:

src/lib/jobs/__tests__/system-job-alert.test.ts

@@ -0,0 +1,87 @@
+/**
+ * The system job alert's verdict on one queue's ledger (#1298). The notice
+ * flow itself runs against a real pg-boss in
+ * `tests/integration/system-job-alert.test.ts`.
+ */
+import { describe, expect, it } from "vitest";
+
+import {
+  judgeQueueRuns,
+  SYSTEM_JOB_FAILURE_STREAK,
+  watchedQueues,
+  type TerminalRun,
+} from "@/lib/jobs/system-job-alert";
+
+const at = new Date("2026-10-29T10:00:00.000Z");
+
+function run(overrides: Partial<TerminalRun> = {}): TerminalRun {
+  return {
+    state: "failed",
+    retryCount: 0,
+    allFailed: false,
+    completedOn: at,
+    error: "boom",
+    expireSeconds: null,
+    ...overrides,
+  };
+}
+
+describe("judgeQueueRuns", () => {
+  it("is inactive with no run left in the ledger", () => {
+    expect(judgeQueueRuns([]).state).toBe("inactive");
+  });
+
+  it("counts every attempt of a run, so one job that used up its retries is a streak", () => {
+    const verdict = judgeQueueRuns([run({ retryCount: 3 })]);
+    expect(verdict).toMatchObject({ state: "failing", failures: 4 });
+    expect(verdict.lastError).toBe("boom");
+  });
+
+  it("calls the streak at the threshold, not before", () => {
+    const below = Array.from({ length: SYSTEM_JOB_FAILURE_STREAK - 1 }, () =>
+      run(),
+    );
+    expect(judgeQueueRuns(below).state).toBe("pending");
+    expect(judgeQueueRuns([...below, run()]).state).toBe("failing");
+  });
+
+  it("ends the streak at the newest success", () => {
+    const verdict = judgeQueueRuns([
+      run({ error: "newest" }),
+      run({ state: "completed", error: null }),
+      run(),
+      run(),
+      run(),
+    ]);
+    expect(verdict).toMatchObject({
+      state: "ok",
+      failures: 1,
+      lastError: "newest",
+    });
+  });
+
+  it("counts a completed run that says every candidate failed", () => {
+    const allFailed = run({ state: "completed", allFailed: true });
+    expect(judgeQueueRuns([allFailed, allFailed, allFailed]).state).toBe(
+      "failing",
+    );
+  });
+
+  it("explains a bare timeout the way the admin card does", () => {
+    const verdict = judgeQueueRuns([
+      run({ error: "job timed out", expireSeconds: 3600, retryCount: 2 }),
+    ]);
+    expect(verdict.lastError).toMatch(
+      /^job timed out: did not finish within its 1 h limit/,
+    );
+  });
+});
+
+describe("watchedQueues", () => {
+  it("watches the system queues and none with an account-facing surface", () => {
+    const queues = watchedQueues();
+    expect(queues).toContain("rollup-full-backfill");
+    expect(queues).toContain("system-job-alert");
+    expect(queues).not.toContain("google-health-sync");
+  });
+});

src/lib/jobs/job-failures.ts

@@ -145,7 +145,7 @@ function explainTimeout(message: string, expireSeconds: number | null): string {
  * serialises to `{ message, stack, … }`; a thrown non-Error can be any JSON at
  * all, so the raw text is the fallback rather than an invented placeholder.
  */
-function presentError(
+export function presentError(
   raw: string | null,
   expireSeconds: number | null = null,
 ): string {

src/lib/jobs/job-outcome.ts

@@ -307,6 +307,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "summary_pending_healed",
   "suppressed_client_managed",
   "suppressed_discreet",
+  // The hourly system job alert: queues failing now, notices sent, admins.
+  "system_job_admins",
+  "system_job_notices",
+  "system_jobs_failing",
   "thumbnails_enqueued",
   "tls_pin_monitor_known_count",
   "tls_pin_monitor_outcome",

src/lib/jobs/queue-runtime.ts

@@ -386,6 +386,9 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
   "recovery-readiness": short(
     "A few indexed reads, one row per account from two backup ledgers, and at most one notice per signal.",
   ),
+  "system-job-alert": short(
+    "The newest fifty terminal runs of each system queue over pg-boss's completion index, and at most one notice per queue.",
+  ),
   "integration-watchdog-hourly": short(
     "A few indexed reads per account with a connection and no provider call; a 14-day history count only for a silent source.",
   ),

src/lib/jobs/registry/status.ts

@@ -4,7 +4,8 @@
  * the morning digest refresh, the data-arrival spine with its reaction line
  * and workout insight, the nightly scores, the period-narrative warm, the
  * Coach memory refresh, nudge, reminder sweep and plan review, the low-stock
- * pass and the daily-briefing fallback.
+ * pass, the daily-briefing fallback, and the hourly alert to the admins
+ * about a system queue that keeps failing.
  *
  * Most queues here are keyless cron ticks, so they carry no policy. The
  * LLM-bound warm queues (pre-generate, period narrative, Coach memory) pass
@@ -77,6 +78,12 @@ import {
   WORKOUT_INSIGHT_GENERATE_QUEUE,
 } from "@/lib/jobs/workout-insight-generate-shared";
 
+import { cronIsTheRetry } from "@/lib/jobs/reminder/registrar-shared";
+import {
+  SYSTEM_JOB_ALERT_CRON,
+  SYSTEM_JOB_ALERT_QUEUE,
+} from "@/lib/jobs/system-job-alert-queue";
+
 import { defineJob, type JobDefinition, type JobGroup } from "./define-job";
 
 const GENERAL_STATUS_QUEUE = "insights-general-status";
@@ -261,6 +268,18 @@ const jobs: readonly JobDefinition[] = [
     work: { localConcurrency: 1 },
     load: async () => (await handlers()).handleDailyBriefing,
   }),
+  // Hourly: tell the admins about a system queue that keeps failing, the way
+  // the recovery readiness evaluation tells them about a backup. Single-flight
+  // and no retries; the next tick is the retry, and the notice claim keeps
+  // two evaluations from sending the same notice.
+  defineJob({
+    queue: SYSTEM_JOB_ALERT_QUEUE,
+    cron: SYSTEM_JOB_ALERT_CRON,
+    scheduleOptions: cronIsTheRetry,
+    work: { localConcurrency: 1 },
+    load: async () =>
+      (await import("@/lib/jobs/system-job-alert")).handleSystemJobAlert,
+  }),
 ];
 
 export const STATUS_GROUP: JobGroup = {

src/lib/jobs/subsystem-surface.ts

@@ -67,6 +67,7 @@ const subsystemSurface = {
   "coach-plan-review": { audience: "system" },
   "medication-low-stock": { audience: "system" },
   "daily-briefing": { audience: "system" },
+  "system-job-alert": { audience: "system" },
 
   // register-rollup.ts
   "rollup-recompute": { audience: "system" },

src/lib/jobs/system-job-alert-queue.ts

@@ -0,0 +1,12 @@
+/**
+ * The queue facts of `system-job-alert.ts`: its name and schedule. They live
+ * apart from the job so the worker registry can read them without loading it.
+ */
+export const SYSTEM_JOB_ALERT_QUEUE = "system-job-alert";
+
+/**
+ * Hourly at :43, a minute no other schedule uses. A notice goes out once per
+ * change whatever the cadence, and an hour is soon enough to hear that a
+ * background job keeps failing.
+ */
+export const SYSTEM_JOB_ALERT_CRON = "43 * * * *";

src/lib/jobs/system-job-alert.ts

@@ -0,0 +1,303 @@
+/**
+ * Push to every admin when a background job of the instance keeps failing:
+ * once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` runs in a row, a
+ * reminder at most once a day while it keeps failing, and once when it runs
+ * successfully again. Nothing else.
+ *
+ * Why (#1298). A full rollup backfill that failed on every boot showed only
+ * on the admin page's card of failing background jobs. From the account's
+ * side the only symptom was empty dashboard cards, and nothing pointed an
+ * operator at the card. The recovery readiness evaluation already pushes a
+ * stopped backup; this does the same for the queues whose failure has no
+ * account-facing status of its own (the `system` audience in
+ * `subsystem-surface.ts`).
+ *
+ * The streak is read from pg-boss's own ledger, newest terminal run first: a
+ * run that failed (or completed and said every candidate failed,
+ * `did.all_failed`) adds its attempts, retries included, and the first run
+ * that succeeded ends the streak. A queue with no terminal run left in the
+ * ledger is not judged at all.
+ *
+ * The memory is one row per announced queue (`SystemJobAlertNotice`), claimed
+ * with a compare-and-set before the notice is sent, exactly as the readiness
+ * notices are (`readiness-notify.ts`, whose decision this reuses), so two
+ * workers send a notice once and a crash between the claim and the send loses
+ * a notice rather than repeating it. Delivery is the ordinary localised path,
+ * as the other operator alerts use: a `SYSTEM_ALERT` to every admin, in their
+ * locale, on the channels they enabled.
+ */
+import type { Job } from "pg-boss";
+
+import { ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account";
+import { prisma } from "@/lib/db";
+import { presentError } from "@/lib/jobs/job-failures";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { SUBSYSTEM_SURFACE } from "@/lib/jobs/subsystem-surface";
+import { withBackgroundEvent } from "@/lib/logging/background";
+import { dispatchLocalisedNotification } from "@/lib/notifications/dispatch-localised";
+import {
+  decideNotice,
+  type NoticeKind,
+  type StoredNotice,
+} from "@/lib/recovery/readiness-notify";
+
+/** Failed runs in a row, retries included, before the admins are told. */
+export const SYSTEM_JOB_FAILURE_STREAK = 3;
+
+/** Newest terminal runs read per queue. */
+const LOOKBACK_ROWS = 50;
+
+/** Where the notice takes the admin: the page with the failing-jobs card. */
+const ADMIN_STATUS_PATH = "/admin/system-status";
+
+/** One terminal run as the ledger holds it, newest first. */
+export interface TerminalRun {
+  state: "completed" | "failed";
+  /** Retries pg-boss ran before this run settled. */
+  retryCount: number;
+  /** The run completed but reported that every candidate failed. */
+  allFailed: boolean;
+  completedOn: Date;
+  error: string | null;
+  expireSeconds: number | null;
+}
+
+/** What the ledger says about one queue. */
+export interface QueueVerdict {
+  /**
+   * `failing`: the streak reached the threshold. `ok`: the newest run, or a
+   * run after the last failures, succeeded. `pending`: fewer failures than
+   * the threshold and no success to judge by. `inactive`: no run left.
+   */
+  state: "failing" | "ok" | "pending" | "inactive";
+  /** Failed attempts in a row, newest first. */
+  failures: number;
+  /** The newest failure's message, redacted and truncated. */
+  lastError: string;
+}
+
+/** Pure: judge one queue's newest terminal runs. */
+export function judgeQueueRuns(runs: readonly TerminalRun[]): QueueVerdict {
+  if (runs.length === 0)
+    return { state: "inactive", failures: 0, lastError: "" };
+  let failures = 0;
+  let lastError: string | null = null;
+  let succeeded = false;
+  for (const run of runs) {
+    if (run.state === "completed" && !run.allFailed) {
+      succeeded = true;
+      break;
+    }
+    failures += run.retryCount + 1;
+    lastError ??= presentError(run.error, run.expireSeconds);
+  }
+  const state =
+    failures >= SYSTEM_JOB_FAILURE_STREAK
+      ? "failing"
+      : succeeded
+        ? "ok"
+        : "pending";
+  return { state, failures, lastError: lastError ?? "" };
+}
+
+/** The queues this alert watches: every one whose failure has no other surface. */
+export function watchedQueues(): string[] {
+  return Object.entries(SUBSYSTEM_SURFACE)
+    .filter(([, surface]) => surface.audience === "system")
+    .map(([queue]) => queue);
+}
+
+/**
+ * The newest terminal runs of each queue, newest first, one statement over
+ * pg-boss's `(name, completed_on)` index. `null` when the queue schema is not
+ * there (a web-only deployment): no queue to ask is not a healthy queue.
+ */
+export async function readTerminalRuns(
+  queues: readonly string[],
+): Promise<Map<string, TerminalRun[]> | null> {
+  try {
+    const rows = await prisma.$queryRaw<
+      Array<{
+        name: string;
+        state: string;
+        retry_count: number;
+        all_failed: string | null;
+        completed_on: Date;
+        error: string | null;
+        expire_seconds: number | null;
+      }>
+    >`
+      SELECT q.name, j.state, j.retry_count, j.all_failed, j.completed_on,
+             j.error, j.expire_seconds
+      FROM unnest(${queues as string[]}::text[]) AS q(name)
+      CROSS JOIN LATERAL (
+        SELECT state, retry_count, completed_on, expire_seconds,
+               output->'did'->>'all_failed' AS all_failed,
+               COALESCE(
+                 output->>'message',
+                 output->'value'->>'message',
+                 output::text
+               ) AS error
+        FROM pgboss.job
+        WHERE name = q.name
+          AND completed_on IS NOT NULL
+          AND state IN ('completed', 'failed')
+        ORDER BY completed_on DESC
+        LIMIT ${LOOKBACK_ROWS}
+      ) j
+      ORDER BY q.name, j.completed_on DESC
+    `;
+    const byQueue = new Map<string, TerminalRun[]>();
+    for (const row of rows) {
+      const runs = byQueue.get(row.name) ?? [];
+      runs.push({
+        state: row.state === "failed" ? "failed" : "completed",
+        retryCount: Number(row.retry_count ?? 0),
+        allFailed: row.all_failed === "true",
+        completedOn: row.completed_on,
+        error: row.error,
+        expireSeconds: row.expire_seconds,
+      });
+      byQueue.set(row.name, runs);
+    }
+    return byQueue;
+  } catch {
+    return null;
+  }
+}
+
+function titleKey(kind: NoticeKind): string {
+  if (kind === "recovered")
+    return "notifications.systemJobAlert.titleRecovered";
+  if (kind === "reminder") return "notifications.systemJobAlert.titleReminder";
+  return "notifications.systemJobAlert.titleFailing";
+}
+
+export interface SystemJobAlertOutcome {
+  /** Queues at or past the threshold now. */
+  failing: number;
+  /** Notices claimed and sent, by queue. */
+  sent: Array<{ queue: string; kind: NoticeKind }>;
+  /** Admins each notice went to. */
+  admins: number;
+}
+
+/**
+ * Judge every watched queue against what the admins were last told, claim
+ * every notice that is due, and send it.
+ */
+export async function notifySystemJobFailures(
+  now: Date = new Date(),
+): Promise<SystemJobAlertOutcome> {
+  const outcome: SystemJobAlertOutcome = { failing: 0, sent: [], admins: 0 };
+  const runs = await readTerminalRuns(watchedQueues());
+  if (runs === null) return outcome;
+
+  // account-deletion: one row per queue of the instance; no account in it.
+  const rows = await prisma.systemJobAlertNotice.findMany();
+  const byQueue = new Map(rows.map((row) => [row.queue, row]));
+  const admins = await prisma.user.findMany({
+    where: { role: "ADMIN", ...ACTIVE_ACCOUNT },
+    select: { id: true },
+  });
+  outcome.admins = admins.length;
+
+  for (const queue of watchedQueues()) {
+    const verdict = judgeQueueRuns(runs.get(queue) ?? []);
+    if (verdict.state === "failing") outcome.failing += 1;
+    // Too few failures to call it a streak and no success to call it over.
+    if (verdict.state === "pending") continue;
+
+    const row = byQueue.get(queue);
+    const stored: StoredNotice | null = row
+      ? {
+          notifiedState:
+            row.notifiedState === "failing" || row.notifiedState === "ok"
+              ? row.notifiedState
+              : null,
+          notifiedAt: row.notifiedAt,
+        }
+      : null;
+    const decision = decideNotice(stored, verdict.state, now);
+    // With no admin yet there is nobody to tell, so nothing is claimed: the
+    // first admin hears about a queue that is still failing.
+    if (!decision.next || admins.length === 0) continue;
+
+    // Claim: only the evaluation that moves the row from what it read sends.
+    const claimed = row
+      ? // account-deletion: the one queue row read above; no account in it.
+        (
+          await prisma.systemJobAlertNotice.updateMany({
+            where: {
+              queue,
+              notifiedState: row.notifiedState,
+              notifiedAt: row.notifiedAt,
+            },
+            data: {
+              notifiedState: decision.next.notifiedState,
+              notifiedAt: decision.next.notifiedAt,
+            },
+          })
+        ).count
+      : (
+          await prisma.systemJobAlertNotice.createMany({
+            data: [
+              {
+                queue,
+                notifiedState: decision.next.notifiedState,
+                notifiedAt: decision.next.notifiedAt,
+              },
+            ],
+            skipDuplicates: true,
+          })
+        ).count;
+    if (claimed !== 1 || decision.send === null) continue;
+
+    for (const admin of admins) {
+      await dispatchLocalisedNotification({
+        userId: admin.id,
+        eventType: "SYSTEM_ALERT",
+        titleKey: titleKey(decision.send),
+        messageKey:
+          decision.send === "recovered"
+            ? "notifications.systemJobAlert.bodyRecovered"
+            : "notifications.systemJobAlert.bodyFailing",
+        params: {
+          queue,
+          failures: verdict.failures,
+          error: verdict.lastError,
+        },
+        metadata: {
+          source: "system-job-alert",
+          url: ADMIN_STATUS_PATH,
+          queue,
+          kind: decision.send,
+        },
+      });
+    }
+    outcome.sent.push({ queue, kind: decision.send });
+  }
+  return outcome;
+}
+
+/** The hourly evaluation. Single-flight; the next tick is the retry. */
+export async function handleSystemJobAlert(
+  jobs: Job<object>[],
+): Promise<JobOutcome> {
+  void jobs;
+  return withBackgroundEvent("job.system_job_alert", async (evt) => {
+    try {
+      const outcome = await notifySystemJobFailures();
+      const did = {
+        system_jobs_failing: outcome.failing,
+        system_job_notices: outcome.sent.length,
+        system_job_admins: outcome.admins,
+      };
+      for (const [key, value] of Object.entries(did)) evt.addMeta(key, value);
+      return jobDone(did);
+    } catch (err) {
+      evt.addWarning(`system-job-alert failed: ${err}`);
+      return jobFailed("system job alert evaluation failed", err);
+    }
+  });
+}