feat(jobs): bind every queue through createAndWork and ban the bare call

All 104 bindings across the five registrars now go through createAndWork,
whose handler type is (jobs) => Promise<JobOutcome>. The forty-two inline
handlers return what they did. A handler that falls off the end no longer
compiles, and the healthlog/job-handler-outcome rule refuses a bare
boss.work( anywhere under src/lib/jobs except the one module that owns
the wrapper. The rule's header states its own limit: it catches the
omission, not a handler that claims ok:true after a swallow.

Retry policy, reviewed queue by queue because the rethrow changes what
runs, not only what is visible:

  - Twenty-one cron passes carry cronIsTheRetry (retryLimit 0). They are
    the retention DELETEs, the OAuth-state prunes, the hourly maintenance
    sweeps and the nightly per-sample drain — every one of them a
    statement that will fail identically a second later. pg-boss defaults
    to two retries with no delay, so converting them without this would
    have run the same doomed statement three times a night. The next tick
    is the retry and it arrives with the whole gap as backoff.
  - The policy rides on the schedule, not on createQueue: create_queue()
    ends in ON CONFLICT DO NOTHING, so a queue-level default would be
    inert on every already-running instance.
  - Everything else keeps its retries deliberately. A TLS probe, an
    off-host upload, a restore drill and a provider poll all fail
    transiently, and the reminder ticks only fail at job level when the
    pass itself could not run.

eslint-plugins/healthlog/index.js

@@ -8,6 +8,8 @@
  *   - `no-raw-palette-color` — ban raw Tailwind palette utilities in app UI.
  *   - `spacing-scale`        — no pt-/pb- overrides on gap-based Card slots,
  *                              and no off-scale `5` step on a bg-card shell.
+ *   - `job-handler-outcome`  — pg-boss bindings route through createAndWork,
+ *                              whose handler returns a JobOutcome.
  */
 
 "use strict";
@@ -17,6 +19,7 @@ const safeFetchRequired = require("./safe-fetch-required.js");
 const apiFetchRequired = require("./api-fetch-required.js");
 const noRawPaletteColor = require("./no-raw-palette-color.js");
 const spacingScale = require("./spacing-scale.js");
+const jobHandlerOutcome = require("./job-handler-outcome.js");
 
 module.exports = {
   rules: {
@@ -30,5 +33,6 @@ module.exports = {
     // named rule (now error-level too; the staged warn phase ended with
     // the semantic sweep). See the rule header.
     "no-dracula-utility": noRawPaletteColor,
+    "job-handler-outcome": jobHandlerOutcome,
   },
 };

eslint-plugins/healthlog/job-handler-outcome.js

@@ -0,0 +1,107 @@
+/**
+ * @fileoverview ESLint rule — a pg-boss queue binding must go through
+ * `createAndWork`, never a bare `boss.work(`.
+ *
+ * `boss.work(queue, options, handler)` completes a job whenever the handler's
+ * promise resolves, whatever it resolves to. Every binding in this tree used
+ * to be written that way, so a handler that caught an error, logged a warning
+ * and returned looked exactly like one that did its work. `createAndWork`
+ * (`src/lib/jobs/reminder/registrar-shared.ts`) types its handler as
+ * `(jobs) => Promise<JobOutcome>` and wraps it in `runJob`, which fails the
+ * pg-boss job on `ok: false`. Routing every binding through it is what makes
+ * "the handler returned nothing" a compile error rather than a silent pass.
+ *
+ * WHAT THIS RULE PROVES, AND WHAT IT DOES NOT
+ *
+ * It proves that no queue in this tree is bound by a path that accepts a
+ * void handler. That is all it proves. It does NOT detect a handler that
+ * returns `{ ok: true }` after swallowing an exception — that compiles, it
+ * passes this rule, and it is exactly as invisible as the defect the outcome
+ * type was written for. The rule converts an omission (forgetting to signal
+ * failure) into a commission (claiming a success that did not happen). The
+ * second is a smaller class and a reviewable one; it is not zero.
+ *
+ * Two further evasions are known and unhandled: aliasing the method
+ * (`const w = boss.work.bind(boss)`, `const { work } = boss`) and calling
+ * `work` on an object not named `boss`. Both are syntactic dodges a reviewer
+ * would see in the diff; neither is defended against here.
+ *
+ * Scope: `src/lib/jobs/`, excluding the registrar module that owns the one
+ * legal `boss.work` call and excluding test files, which mock `boss.work`
+ * and assert against it directly.
+ *
+ * @see src/lib/jobs/run-job.ts
+ * @see src/lib/jobs/job-outcome.ts
+ * @see src/lib/jobs/reminder/registrar-shared.ts
+ */
+
+"use strict";
+
+// The one module allowed to call `boss.work` — `createAndWork` lives here
+// and is the wrapper every other binding has to use.
+const EXEMPT_FILES = ["src/lib/jobs/reminder/registrar-shared.ts"];
+
+// The worker tree. A `boss.work` outside it would not be a queue binding.
+const ENFORCED_ROOTS = ["src/lib/jobs/"];
+
+function toPosix(filename) {
+  return filename.replace(/\\/g, "/");
+}
+
+function isTestFile(posix) {
+  return (
+    /\.test\.[cm]?[jt]sx?$/.test(posix) ||
+    /\.spec\.[cm]?[jt]sx?$/.test(posix) ||
+    posix.includes("/__tests__/") ||
+    posix.includes("/__mocks__/")
+  );
+}
+
+function isEnforced(filename) {
+  const posix = toPosix(filename);
+  if (!ENFORCED_ROOTS.some((root) => posix.includes(root))) return false;
+  if (EXEMPT_FILES.some((f) => posix.includes(f))) return false;
+  if (isTestFile(posix)) return false;
+  return true;
+}
+
+/** @type {import("eslint").Rule.RuleModule} */
+const jobHandlerOutcomeRule = {
+  meta: {
+    type: "problem",
+    docs: {
+      description:
+        "pg-boss queue bindings must go through createAndWork so the handler returns a JobOutcome.",
+    },
+    schema: [],
+    messages: {
+      bareBossWork:
+        'boss.work() completes a job whenever the handler resolves, so a swallowed failure reads as success. Bind the queue with createAndWork(boss, QUEUE, options, handler) from "./registrar-shared" — its handler type is (jobs) => Promise<JobOutcome> and runJob fails the job on ok:false.',
+    },
+  },
+  create(context) {
+    const filename = context.filename ?? context.getFilename?.();
+    if (!filename || !isEnforced(filename)) {
+      return {};
+    }
+
+    return {
+      CallExpression(node) {
+        const callee = node.callee;
+        if (
+          callee.type !== "MemberExpression" ||
+          callee.computed ||
+          callee.property.type !== "Identifier" ||
+          callee.property.name !== "work" ||
+          callee.object.type !== "Identifier" ||
+          callee.object.name !== "boss"
+        ) {
+          return;
+        }
+        context.report({ node, messageId: "bareBossWork" });
+      },
+    };
+  },
+};
+
+module.exports = jobHandlerOutcomeRule;

eslint.config.mjs

@@ -75,6 +75,13 @@ const eslintConfig = defineConfig([
       // trip it. Density lives on the Card itself (`gap-2 py-3 md:py-4`) or on
       // a swept `p-4 md:p-6` shell. See the rule header.
       "healthlog/spacing-scale": "error",
+      // Every pg-boss queue binding under src/lib/jobs goes through
+      // `createAndWork`, whose handler type is `(jobs) => Promise<JobOutcome>`.
+      // A bare `boss.work(` accepts a handler that resolves to nothing, which
+      // is how a swallowed failure came to read as a completed job. The rule
+      // catches the omission only — a handler that returns `{ ok: true }`
+      // after swallowing still passes, and its header says so.
+      "healthlog/job-handler-outcome": "error",
     },
   },
   // v1.28.17 — every recharts-rendering component funnels through the

src/lib/jobs/reminder/register-integration-sync.ts

@@ -9,8 +9,8 @@
  *
  * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
  * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
- * and a `boss.work(QUEUE, …, handler)` binding drains it. The queue-wiring
- * guards (`withings-queues`, `whoop-queues`, `fitbit-queues`,
+ * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The
+ * queue-wiring guards (`withings-queues`, `whoop-queues`, `fitbit-queues`,
  * `nightscout-queues`, `sleep-timeline-queues`) read THIS module.
  */
 import { PgBoss } from "pg-boss";
@@ -78,8 +78,11 @@ import {
   type IntegrationBackfillAdmissionPayload,
 } from "@/lib/jobs/integration-backfill-admission";
 import { workerLog } from "./shared";
+import { jobDone } from "@/lib/jobs/job-outcome";
 import {
   createAndSchedule,
+  createAndWork,
+  cronIsTheRetry,
   type QueuePolicyTable,
   type ScheduleEntry,
 } from "./registrar-shared";
@@ -333,7 +336,16 @@ const schedules: ScheduleEntry[] = [
   // v1.32.23 — hourly ECG catch-net (:41) so watch-only accounts pull their
   // strips without a webhook or a scale.
   [WITHINGS_ECG_SYNC_QUEUE, WITHINGS_ECG_SYNC_CRON],
-  [WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_CRON],
+  // The five OAuth-state / handoff prunes carry `cronIsTheRetry`: their
+  // handlers used to warn and carry on, they now fail the job, and a DELETE
+  // that could not run will not run any better a second later. The daily tick
+  // is the retry. The sync crons above keep the default retries — a provider
+  // outage is transient and worth retrying inside the hour.
+  [
+    WITHINGS_OAUTH_STATE_CLEANUP_QUEUE,
+    WITHINGS_OAUTH_STATE_CLEANUP_CRON,
+    cronIsTheRetry,
+  ],
   // v1.11.0 — WHOOP poll-fallback crons. Recovery/sleep/workout catch
   // dropped webhooks; cycle is the sole driver (no webhook). Staggered off
   // the Withings crons so the hourly ticks don't pile up on one boss poll.
@@ -342,11 +354,19 @@ const schedules: ScheduleEntry[] = [
   [WHOOP_WORKOUT_SYNC_QUEUE, WHOOP_WORKOUT_SYNC_CRON],
   [WHOOP_CYCLE_SYNC_QUEUE, WHOOP_CYCLE_SYNC_CRON],
   // v1.11.0 — daily 03:22 Europe/Berlin prune for expired WHOOP OAuth states.
-  [WHOOP_OAUTH_STATE_CLEANUP_QUEUE, WHOOP_OAUTH_STATE_CLEANUP_CRON],
+  [
+    WHOOP_OAUTH_STATE_CLEANUP_QUEUE,
+    WHOOP_OAUTH_STATE_CLEANUP_CRON,
+    cronIsTheRetry,
+  ],
   // v1.12.0 — hourly Fitbit poll (:08, staggered off WHOOP/Withings) + the
   // daily 03:24 Europe/Berlin prune for expired Fitbit OAuth states.
   [FITBIT_SYNC_QUEUE, FITBIT_SYNC_CRON],
-  [FITBIT_OAUTH_STATE_CLEANUP_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_CRON],
+  [
+    FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
+    FITBIT_OAUTH_STATE_CLEANUP_CRON,
+    cronIsTheRetry,
+  ],
   // v1.26.0 — hourly Google Health poll (:18, staggered off WHOOP/Fitbit/
   // Nightscout/Polar/Oura) + the daily 03:26 Europe/Berlin prune for expired
   // Google Health OAuth states.
@@ -354,6 +374,7 @@ const schedules: ScheduleEntry[] = [
   [
     GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE,
     GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_CRON,
+    cronIsTheRetry,
   ],
   // v1.17.0 — hourly Nightscout CGM poll (:11, staggered off the other sync
   // ticks).
@@ -364,7 +385,11 @@ const schedules: ScheduleEntry[] = [
   // v1.28.x — hourly Strava (:17) OAuth poll.
   [STRAVA_SYNC_QUEUE, STRAVA_SYNC_CRON],
   // v1.30.x — daily 03:28 Europe/Berlin prune for expired native OIDC handoffs.
-  [OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, OIDC_NATIVE_HANDOFF_CLEANUP_CRON],
+  [
+    OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE,
+    OIDC_NATIVE_HANDOFF_CLEANUP_CRON,
+    cronIsTheRetry,
+  ],
 ];
 
 /**
@@ -450,19 +475,30 @@ export async function registerIntegrationSyncQueues(
   // reaches this shared durable queue before doing full-history work. The
   // common group is enforced in PostgreSQL across worker processes/nodes.
 
-  await boss.work<IntegrationBackfillAdmissionPayload>(
+  await createAndWork<IntegrationBackfillAdmissionPayload>(
+    boss,
     INTEGRATION_BACKFILL_ADMISSION_QUEUE,
     {
       localConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY,
       groupConcurrency: INTEGRATION_BACKFILL_GLOBAL_CONCURRENCY,
     },
     async (jobs) => {
+      // The drain has no per-job error isolation: a runner that throws fails
+      // the job and the admission retry policy applies. So reaching the end
+      // means every admitted job did its work, and the counts each runner
+      // already returns are what the pass reports.
+      let importedTotal = 0;
+      let removedTotal = 0;
+      let deletedTotal = 0;
+      let markersTotal = 0;
+      let linkedTotal = 0;
       for (const job of jobs) {
         const { kind, data } = job.data;
         const { userId } = data;
         switch (kind) {
           case "whoop-backfill": {
             const { imported } = await runWhoopBackfillForUser(userId);
+            importedTotal += imported;
             workerLog(
               "info",
               `[whoop-backfill] user=${userId} imported=${imported}`,
@@ -471,6 +507,7 @@ export async function registerIntegrationSyncQueues(
           }
           case "fitbit-backfill": {
             const { imported } = await runFitbitBackfillForUser(userId);
+            importedTotal += imported;
             workerLog(
               "info",
               `[fitbit-backfill] user=${userId} imported=${imported}`,
@@ -479,6 +516,7 @@ export async function registerIntegrationSyncQueues(
           }
           case "google-health-backfill": {
             const { imported } = await runGoogleHealthBackfillForUser(userId);
+            importedTotal += imported;
             workerLog(
               "info",
               `[google-health-backfill] user=${userId} imported=${imported}`,
@@ -488,6 +526,7 @@ export async function registerIntegrationSyncQueues(
           case "google-health-sleep-repair": {
             const { imported } =
               await runGoogleHealthSleepRepairForUser(userId);
+            importedTotal += imported;
             workerLog(
               "info",
               `[google-health-sleep-repair] user=${userId} imported=${imported}`,
@@ -497,6 +536,8 @@ export async function registerIntegrationSyncQueues(
           case "fitbit-sleep-repair": {
             const { imported, removed } =
               await runFitbitSleepRepairForUser(userId);
+            importedTotal += imported;
+            removedTotal += removed;
             workerLog(
               "info",
               `[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`,
@@ -513,6 +554,8 @@ export async function registerIntegrationSyncQueues(
               userId,
               data.provider,
             );
+            deletedTotal += deleted;
+            importedTotal += imported;
             workerLog(
               "info",
               `[sleep-timeline-backfill] user=${userId} provider=${data.provider} deleted=${deleted} imported=${imported}`,
@@ -522,6 +565,8 @@ export async function registerIntegrationSyncQueues(
           case "lab-biomarker-backfill": {
             const { markers, linked } =
               await runLabBiomarkerBackfillForUser(userId);
+            markersTotal += markers;
+            linkedTotal += linked;
             workerLog(
               "info",
               `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`,
@@ -530,6 +575,7 @@ export async function registerIntegrationSyncQueues(
           }
           case "strava-backfill": {
             const { imported } = await runStravaBackfillForUser(userId);
+            importedTotal += imported;
             workerLog(
               "info",
               `[strava-backfill] user=${userId} imported=${imported}`,
@@ -538,30 +584,43 @@ export async function registerIntegrationSyncQueues(
           }
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        imported: importedTotal,
+        removed: removedTotal,
+        deleted: deletedTotal,
+        markers: markersTotal,
+        linked: linkedTotal,
+      });
     },
   );
 
-  await boss.work<WithingsSyncPayload>(
+  await createAndWork<WithingsSyncPayload>(
+    boss,
     WITHINGS_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleWithingsFallbackSync,
   );
-  await boss.work<WithingsActivitySyncPayload>(
+  await createAndWork<WithingsActivitySyncPayload>(
+    boss,
     WITHINGS_ACTIVITY_QUEUE,
     { localConcurrency: 1 },
     handleWithingsActivitySync,
   );
-  await boss.work<WithingsSleepSyncPayload>(
+  await createAndWork<WithingsSleepSyncPayload>(
+    boss,
     WITHINGS_SLEEP_QUEUE,
     { localConcurrency: 1 },
     handleWithingsSleepSync,
   );
-  await boss.work<WithingsEcgSyncPayload>(
+  await createAndWork<WithingsEcgSyncPayload>(
+    boss,
     WITHINGS_ECG_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleWithingsEcgSync,
   );
-  await boss.work<WithingsOAuthStateCleanupPayload>(
+  await createAndWork<WithingsOAuthStateCleanupPayload>(
+    boss,
     WITHINGS_OAUTH_STATE_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleWithingsOAuthStateCleanup,
@@ -569,29 +628,34 @@ export async function registerIntegrationSyncQueues(
   // v1.11.0 — WHOOP per-resource sync handlers. Webhook-driven per-user +
   // cron full-iteration. Serial concurrency so a backfill-heavy tick never
   // crowds the request pool and stays inside WHOOP's 100 req/min app cap.
-  await boss.work<WhoopSyncPayload>(
+  await createAndWork<WhoopSyncPayload>(
+    boss,
     WHOOP_RECOVERY_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleWhoopRecoverySync,
   );
-  await boss.work<WhoopSyncPayload>(
+  await createAndWork<WhoopSyncPayload>(
+    boss,
     WHOOP_SLEEP_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleWhoopSleepSync,
   );
-  await boss.work<WhoopSyncPayload>(
+  await createAndWork<WhoopSyncPayload>(
+    boss,
     WHOOP_WORKOUT_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleWhoopWorkoutSync,
   );
-  await boss.work<WhoopSyncPayload>(
+  await createAndWork<WhoopSyncPayload>(
+    boss,
     WHOOP_CYCLE_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleWhoopCycleSync,
   );
   // v1.11.0 — self-converging WHOOP backfill. This source queue retains the
   // per-connection singleton and forwards admitted work to the shared drain.
-  await boss.work<WhoopBackfillPayload>(
+  await createAndWork<WhoopBackfillPayload>(
+    boss,
     WHOOP_BACKFILL_QUEUE,
     { localConcurrency: WHOOP_BACKFILL_CONCURRENCY },
     async (jobs) => {
@@ -601,29 +665,34 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
-  await boss.work<WhoopOAuthStateCleanupPayload>(
+  await createAndWork<WhoopOAuthStateCleanupPayload>(
+    boss,
     WHOOP_OAUTH_STATE_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleWhoopOAuthStateCleanup,
   );
   // v1.30.x — daily sweep for the native OIDC handoff ledger.
-  await boss.work<OidcNativeHandoffCleanupPayload>(
+  await createAndWork<OidcNativeHandoffCleanupPayload>(
+    boss,
     OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleOidcNativeHandoffCleanup,
   );
   // v1.12.0 — Fitbit poll-sync (cron full-iteration; no webhook). Serial
   // concurrency so a backfill-heavy tick never crowds the request pool.
-  await boss.work<FitbitSyncPayload>(
+  await createAndWork<FitbitSyncPayload>(
+    boss,
     FITBIT_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleFitbitSync,
   );
   // v1.12.0 — self-converging Fitbit backfill. This source queue retains the
   // per-connection singleton and forwards admitted work to the shared drain.
-  await boss.work<FitbitBackfillPayload>(
+  await createAndWork<FitbitBackfillPayload>(
+    boss,
     FITBIT_BACKFILL_QUEUE,
     { localConcurrency: FITBIT_BACKFILL_CONCURRENCY },
     async (jobs) => {
@@ -633,12 +702,14 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
   // One-shot Fitbit sleep duplicate repair. This source queue retains the
   // per-connection singleton and forwards the watermark-safe bounded sleep
   // re-read to the shared drain.
-  await boss.work<FitbitSleepRepairPayload>(
+  await createAndWork<FitbitSleepRepairPayload>(
+    boss,
     FITBIT_SLEEP_REPAIR_QUEUE,
     { localConcurrency: FITBIT_SLEEP_REPAIR_CONCURRENCY },
     async (jobs) => {
@@ -648,16 +719,19 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
-  await boss.work<FitbitOAuthStateCleanupPayload>(
+  await createAndWork<FitbitOAuthStateCleanupPayload>(
+    boss,
     FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleFitbitOAuthStateCleanup,
   );
   // v1.26.0 — Google Health poll-sync (cron full-iteration; no webhook). Serial
   // concurrency so a backfill-heavy tick never crowds the request pool.
-  await boss.work<GoogleHealthSyncPayload>(
+  await createAndWork<GoogleHealthSyncPayload>(
+    boss,
     GOOGLE_HEALTH_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleGoogleHealthSync,
@@ -665,7 +739,8 @@ export async function registerIntegrationSyncQueues(
   // v1.26.0 — self-converging Google Health backfill. This source queue
   // retains the per-connection singleton and forwards admitted work to the
   // shared drain.
-  await boss.work<GoogleHealthBackfillPayload>(
+  await createAndWork<GoogleHealthBackfillPayload>(
+    boss,
     GOOGLE_HEALTH_BACKFILL_QUEUE,
     { localConcurrency: GOOGLE_HEALTH_BACKFILL_CONCURRENCY },
     async (jobs) => {
@@ -675,12 +750,14 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
   // v1.28.x — one-shot Google Health sleep duplicate repair. This source
   // queue retains the per-connection singleton and forwards the watermark-safe
   // full sleep-history re-read to the shared drain.
-  await boss.work<GoogleHealthSleepRepairPayload>(
+  await createAndWork<GoogleHealthSleepRepairPayload>(
+    boss,
     GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE,
     { localConcurrency: GOOGLE_HEALTH_SLEEP_REPAIR_CONCURRENCY },
     async (jobs) => {
@@ -690,9 +767,11 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
-  await boss.work<GoogleHealthOAuthStateCleanupPayload>(
+  await createAndWork<GoogleHealthOAuthStateCleanupPayload>(
+    boss,
     GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleGoogleHealthOAuthStateCleanup,
@@ -700,7 +779,8 @@ export async function registerIntegrationSyncQueues(
   // v1.17.1 — one-shot sleep-timeline backfill. This source queue retains the
   // per-(user, provider) singleton and forwards admitted work to the shared
   // drain.
-  await boss.work<SleepTimelineBackfillPayload>(
+  await createAndWork<SleepTimelineBackfillPayload>(
+    boss,
     SLEEP_TIMELINE_BACKFILL_QUEUE,
     { localConcurrency: SLEEP_TIMELINE_BACKFILL_CONCURRENCY },
     async (jobs) => {
@@ -710,11 +790,13 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
   // v1.18.1 — one-shot lab-biomarker backfill. This source queue retains the
   // per-user singleton and forwards admitted work to the shared drain.
-  await boss.work<LabBiomarkerBackfillPayload>(
+  await createAndWork<LabBiomarkerBackfillPayload>(
+    boss,
     LAB_BIOMARKER_BACKFILL_QUEUE,
     { localConcurrency: LAB_BIOMARKER_BACKFILL_CONCURRENCY },
     async (jobs) => {
@@ -724,12 +806,14 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
   // v1.17.0 — Nightscout CGM poll-cohort sync. The hourly cron tick (no
   // `userId`) walks every configured instance; one user's unreachable host is
   // warned, not fatal.
-  await boss.work<NightscoutSyncPayload>(
+  await createAndWork<NightscoutSyncPayload>(
+    boss,
     NIGHTSCOUT_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleNightscoutSync,
@@ -737,12 +821,14 @@ export async function registerIntegrationSyncQueues(
   // v1.17.0 (F4) — Polar + Oura OAuth poll-cohort sync. The hourly cron tick
   // (no `userId`) re-walks every connected user; one user's revoked grant is
   // warned, not fatal.
-  await boss.work<PolarSyncPayload>(
+  await createAndWork<PolarSyncPayload>(
+    boss,
     POLAR_SYNC_QUEUE,
     { localConcurrency: 1 },
     handlePolarSync,
   );
-  await boss.work<OuraSyncPayload>(
+  await createAndWork<OuraSyncPayload>(
+    boss,
     OURA_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleOuraSync,
@@ -750,14 +836,16 @@ export async function registerIntegrationSyncQueues(
   // v1.28.x — Strava OAuth poll-cohort sync (poll-only; webhook deferred). The
   // hourly cron tick (no `userId`) re-walks every connected user; one user's
   // revoked grant is warned, not fatal.
-  await boss.work<StravaSyncPayload>(
+  await createAndWork<StravaSyncPayload>(
+    boss,
     STRAVA_SYNC_QUEUE,
     { localConcurrency: 1 },
     handleStravaSync,
   );
   // v1.28.x — self-converging Strava backfill. This source queue retains the
   // per-connection singleton and forwards admitted work to the shared drain.
-  await boss.work<StravaBackfillPayload>(
+  await createAndWork<StravaBackfillPayload>(
+    boss,
     STRAVA_BACKFILL_QUEUE,
     { localConcurrency: STRAVA_BACKFILL_CONCURRENCY },
     async (jobs) => {
@@ -767,6 +855,7 @@ export async function registerIntegrationSyncQueues(
           data: job.data,
         });
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
   return allQueues;

src/lib/jobs/reminder/register-maintenance.ts

@@ -10,7 +10,7 @@
  *
  * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
  * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
- * and a `boss.work(QUEUE, …, handler)` binding drains it. The
+ * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The
  * `measurement-tombstone-cleanup`, `withings-oauth-state-cleanup`,
  * `intake-slot-dedup-queue`, `tls-pin-monitor-queue`, and `geo-backfill`
  * guards read THIS module.
@@ -131,9 +131,12 @@ import {
   type EnvironmentFetchPayload,
 } from "@/lib/jobs/environment-fetch";
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { workerLog, BOOT_BACKFILL_STAGGER_SECONDS } from "./shared";
 import {
   createAndSchedule,
+  createAndWork,
+  cronIsTheRetry,
   type QueuePolicyTable,
   type ScheduleEntry,
 } from "./registrar-shared";
@@ -411,24 +414,35 @@ const allQueues = [
   DOCUMENT_SUMMARY_CATCHUP_QUEUE,
 ];
 
+// `cronIsTheRetry` (retryLimit 0) marks the passes whose handler now returns
+// `jobFailed` where it used to warn and carry on, AND whose failure repeats:
+// a bulk DELETE or UPDATE that timed out will time out again on an immediate
+// retry, so pg-boss's default two would run the same doomed statement three
+// times a night. The queues without it are the ones whose failure is
+// transient — the TLS probe, the off-host upload, the restore drill, the
+// per-user environment fetch — and those still want the default retries.
 const schedules: ScheduleEntry[] = [
   // v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload)
   // that fans out one per-user environment fetch per opted-in account.
   [ENVIRONMENT_FETCH_QUEUE, ENVIRONMENT_FETCH_CRON],
   [DATA_BACKUP_QUEUE, DATA_BACKUP_CRON],
-  [RATE_LIMIT_CLEANUP_QUEUE, RATE_LIMIT_CLEANUP_CRON],
-  [IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON],
+  [RATE_LIMIT_CLEANUP_QUEUE, RATE_LIMIT_CLEANUP_CRON, cronIsTheRetry],
+  [IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON, cronIsTheRetry],
   [AUDIT_LOG_CLEANUP_QUEUE, AUDIT_LOG_CLEANUP_CRON],
-  [STEP_UP_ELEVATION_CLEANUP_QUEUE, STEP_UP_ELEVATION_CLEANUP_CRON],
+  [
+    STEP_UP_ELEVATION_CLEANUP_QUEUE,
+    STEP_UP_ELEVATION_CLEANUP_CRON,
+    cronIsTheRetry,
+  ],
   [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
   [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON],
-  [HOST_METRIC_QUEUE, HOST_METRIC_CRON],
-  [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON],
+  [HOST_METRIC_QUEUE, HOST_METRIC_CRON, cronIsTheRetry],
+  [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON, cronIsTheRetry],
   // v1.4.37 — hourly geo backfill. The helper is idempotent + capped
   // at 5 000 rows per pass; running it at :40 every hour catches the
   // long tail of audit rows that landed with the offline MMDB
   // missing or the online provider unreachable.
-  [GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON],
+  [GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON, cronIsTheRetry],
   // v1.12.2 ios-coord — every-6-hour TLS leaf SPKI probe (:07 off the
   // hourly sync crons). Surfaces a pinned-leaf rotation well inside the
   // ≥11-day re-pin window the iOS release owner needs.
@@ -443,14 +457,18 @@ const schedules: ScheduleEntry[] = [
   // entities. Flips IN_USE rows whose 30-day clock has blown to
   // EXPIRED at 03:30 Europe/Berlin (in the existing 02:xx–03:xx
   // maintenance window, right after idempotency-cleanup).
-  [MEDICATION_INVENTORY_EXPIRE_QUEUE, MEDICATION_INVENTORY_EXPIRE_CRON],
+  [
+    MEDICATION_INVENTORY_EXPIRE_QUEUE,
+    MEDICATION_INVENTORY_EXPIRE_CRON,
+    cronIsTheRetry,
+  ],
   // v1.4.46 — hourly auto-skip for medication intakes the user never
   // marked. Cron `5 * * * *` slots off the top-of-the-hour
   // reminder-check (:00) so the hourly ticks don't pile up on the same
   // boss poll.
-  [INTAKE_AUTO_SKIP_QUEUE, INTAKE_AUTO_SKIP_CRON],
+  [INTAKE_AUTO_SKIP_QUEUE, INTAKE_AUTO_SKIP_CRON, cronIsTheRetry],
   // v0.5.4 ios-coord — daily mood-reminder dispatch-ledger retention sweep.
-  [MOOD_REMINDER_CLEANUP_QUEUE, MOOD_REMINDER_CLEANUP_CRON],
+  [MOOD_REMINDER_CLEANUP_QUEUE, MOOD_REMINDER_CLEANUP_CRON, cronIsTheRetry],
   // v1.15.19 — daily 03:28 Europe/Berlin duplicate dose-slot dedup
   // discovery. The empty cron payload (no `userId`) is the handler's
   // signal to run the discovery fan-out instead of a per-user pass, so
@@ -458,20 +476,24 @@ const schedules: ScheduleEntry[] = [
   // a day instead of waiting for the next worker reboot.
   [INTAKE_SLOT_DEDUP_QUEUE, INTAKE_SLOT_DEDUP_CRON],
   // v1.4.49 — daily 03:35 Europe/Berlin prune for push_attempts.
-  [PUSH_ATTEMPT_CLEANUP_QUEUE, PUSH_ATTEMPT_CLEANUP_CRON],
+  [PUSH_ATTEMPT_CLEANUP_QUEUE, PUSH_ATTEMPT_CLEANUP_CRON, cronIsTheRetry],
   // v1.31.0 — daily 04:10 Europe/Berlin prune for arrival_reactions.
-  [ARRIVAL_REACTION_CLEANUP_QUEUE, ARRIVAL_REACTION_CLEANUP_CRON],
+  [
+    ARRIVAL_REACTION_CLEANUP_QUEUE,
+    ARRIVAL_REACTION_CLEANUP_CRON,
+    cronIsTheRetry,
+  ],
   // v1.7.0 — daily 03:40 Europe/Berlin prune for expired measurement
   // tombstones.
   [MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, MEASUREMENT_TOMBSTONE_CLEANUP_CRON],
   // v1.18.7 — daily 03:50 Europe/Berlin prune for stale Coach history.
-  [COACH_MESSAGE_CLEANUP_QUEUE, COACH_MESSAGE_CLEANUP_CRON],
+  [COACH_MESSAGE_CLEANUP_QUEUE, COACH_MESSAGE_CLEANUP_CRON, cronIsTheRetry],
   // v1.23 — daily 03:55 Europe/Berlin note-encryption backfill discovery.
   // The empty cron payload (no `userId`) is the handler's signal to fan out
   // one per-user job per account still holding a plaintext note.
   [NOTE_ENCRYPTION_BACKFILL_QUEUE, NOTE_ENCRYPTION_BACKFILL_CRON],
   // v1.24 — daily 04:00 Europe/Berlin prune for dead MCP connector tokens.
-  [MCP_TOKEN_CLEANUP_QUEUE, MCP_TOKEN_CLEANUP_CRON],
+  [MCP_TOKEN_CLEANUP_QUEUE, MCP_TOKEN_CLEANUP_CRON, cronIsTheRetry],
   // v1.25 — daily 04:05 Europe/Berlin medication-note encryption backfill
   // discovery. The empty cron payload (no `userId`) is the handler's signal to
   // fan out one per-user job per account still holding a plaintext medication
@@ -479,14 +501,14 @@ const schedules: ScheduleEntry[] = [
   [MED_NOTES_ENCRYPTION_BACKFILL_QUEUE, MED_NOTES_ENCRYPTION_BACKFILL_CRON],
   // Document vault — daily 04:10 Europe/Berlin purge for tombstoned
   // documents past the 30-day undo grace.
-  [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON],
+  [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON, cronIsTheRetry],
   // v1.32.1 (issue #588) — every-15-minute orphan-ImportJob sweep. Re-runs
   // the same reconcile the boot path uses, so a stuck "unpacking" row
   // whose worker crashed/restarted without the boot-time pass catching it
   // (heartbeat not yet stale, pg-boss job not yet expired) still converges
   // to a visible `failed` state within two ticks instead of staying stuck
   // forever on an otherwise-healthy worker.
-  [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON],
+  [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON, cronIsTheRetry],
 ];
 
 /**
@@ -584,61 +606,76 @@ export async function registerMaintenanceQueues(
 ): Promise<readonly string[]> {
   await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
-  await boss.work<DataBackupPayload>(
+  await createAndWork<DataBackupPayload>(
+    boss,
     DATA_BACKUP_QUEUE,
     { localConcurrency: 1 },
     handleDataBackup,
   );
-  await boss.work<RateLimitCleanupPayload>(
+  await createAndWork<RateLimitCleanupPayload>(
+    boss,
     RATE_LIMIT_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleRateLimitCleanup,
   );
-  await boss.work<IdempotencyCleanupPayload>(
+  await createAndWork<IdempotencyCleanupPayload>(
+    boss,
     IDEMPOTENCY_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleIdempotencyCleanup,
   );
-  await boss.work<AuditLogCleanupPayload>(
+  await createAndWork<AuditLogCleanupPayload>(
+    boss,
     AUDIT_LOG_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleAuditLogCleanup,
   );
-  await boss.work<StepUpElevationCleanupPayload>(
+  await createAndWork<StepUpElevationCleanupPayload>(
+    boss,
     STEP_UP_ELEVATION_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleStepUpElevationCleanup,
   );
-  await boss.work<OffhostBackupPayload>(
+  await createAndWork<OffhostBackupPayload>(
+    boss,
     OFFHOST_BACKUP_QUEUE,
     { localConcurrency: 1 },
     handleOffhostBackup,
   );
-  // prettier-ignore
-  await boss.work(RESTORE_DRILL_QUEUE, { localConcurrency: 1 }, handleRestoreDrill);
-  await boss.work<HostMetricSamplePayload>(
+  await createAndWork(
+    boss,
+    RESTORE_DRILL_QUEUE,
+    { localConcurrency: 1 },
+    handleRestoreDrill,
+  );
+  await createAndWork<HostMetricSamplePayload>(
+    boss,
     HOST_METRIC_QUEUE,
     { localConcurrency: 1 },
     handleHostMetricSample,
   );
-  await boss.work<FeedbackAggregatorPayload>(
+  await createAndWork<FeedbackAggregatorPayload>(
+    boss,
     FEEDBACK_AGGREGATOR_QUEUE,
     { localConcurrency: 1 },
     handleFeedbackAggregator,
   );
-  await boss.work<GeoBackfillPayload>(
+  await createAndWork<GeoBackfillPayload>(
+    boss,
     GEO_BACKFILL_QUEUE,
     { localConcurrency: 1 },
     handleGeoBackfill,
   );
   // v1.12.2 ios-coord — TLS leaf SPKI-change monitor. Single-flight: one
   // short outbound TLS handshake per tick, no benefit to overlapping ticks.
-  await boss.work<TlsPinMonitorPayload>(
+  await createAndWork<TlsPinMonitorPayload>(
+    boss,
     TLS_PIN_MONITOR_QUEUE,
     { localConcurrency: 1 },
     handleTlsPinMonitor,
   );
-  await boss.work<MoodReminderCleanupPayload>(
+  await createAndWork<MoodReminderCleanupPayload>(
+    boss,
     MOOD_REMINDER_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleMoodReminderCleanup,
@@ -647,7 +684,8 @@ export async function registerMaintenanceQueues(
   // matches every other cleanup queue; two ticks racing on the same
   // DELETE statement is wasted work and the second tick's payload
   // would be a no-op anyway.
-  await boss.work<PushAttemptCleanupPayload>(
+  await createAndWork<PushAttemptCleanupPayload>(
+    boss,
     PUSH_ATTEMPT_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handlePushAttemptCleanup,
@@ -655,14 +693,16 @@ export async function registerMaintenanceQueues(
   // v1.31.0 — daily prune of the data-arrival reaction markers. Single-flight
   // like every other cleanup queue: two ticks racing the same DELETE is
   // wasted work.
-  await boss.work<ArrivalReactionCleanupPayload>(
+  await createAndWork<ArrivalReactionCleanupPayload>(
+    boss,
     ARRIVAL_REACTION_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleArrivalReactionCleanup,
   );
   // v1.7.0 — daily prune of expired measurement tombstones. Single-flight
   // like every other cleanup queue.
-  await boss.work<MeasurementTombstoneCleanupPayload>(
+  await createAndWork<MeasurementTombstoneCleanupPayload>(
+    boss,
     MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleMeasurementTombstoneCleanup,
@@ -670,14 +710,16 @@ export async function registerMaintenanceQueues(
   // v1.18.7 — daily prune of the encrypted Coach conversation history.
   // Single-flight like every other cleanup queue; two ticks racing on the
   // same DELETE is wasted work and the second is a no-op.
-  await boss.work<CoachMessageCleanupPayload>(
+  await createAndWork<CoachMessageCleanupPayload>(
+    boss,
     COACH_MESSAGE_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleCoachMessageCleanup,
   );
   // MCP Phase 3 (M2) — daily prune of dead MCP connector access tokens +
   // long-revoked connection anchors. Single-flight like every other cleanup.
-  await boss.work<McpTokenCleanupPayload>(
+  await createAndWork<McpTokenCleanupPayload>(
+    boss,
     MCP_TOKEN_CLEANUP_QUEUE,
     { localConcurrency: 1 },
     handleMcpTokenCleanup,
@@ -685,17 +727,20 @@ export async function registerMaintenanceQueues(
   // Document vault — daily purge of tombstoned documents past the 30-day
   // undo grace. Single-flight like every other cleanup queue; the underlying
   // deleteMany is idempotent so a duplicate tick is a no-op.
-  await boss.work<DocumentPurgePayload>(
+  await createAndWork<DocumentPurgePayload>(
+    boss,
     DOCUMENT_PURGE_QUEUE,
     { localConcurrency: 1 },
     handleDocumentPurge,
   );
-  await boss.work<PrDetectionPayload>(
+  await createAndWork<PrDetectionPayload>(
+    boss,
     PR_DETECTION_QUEUE,
     { localConcurrency: PR_DETECTION_CONCURRENCY },
     handlePrDetection,
   );
-  await boss.work<MedicationInventoryExpirePayload>(
+  await createAndWork<MedicationInventoryExpirePayload>(
+    boss,
     MEDICATION_INVENTORY_EXPIRE_QUEUE,
     { localConcurrency: 1 },
     handleMedicationInventoryExpire,
@@ -704,7 +749,8 @@ export async function registerMaintenanceQueues(
   // Single-flight: two ticks racing against the same row pile is wasted
   // work, and the underlying `updateMany` is the canonical idempotent
   // shape anyway.
-  await boss.work<IntakeAutoSkipPayload>(
+  await createAndWork<IntakeAutoSkipPayload>(
+    boss,
     INTAKE_AUTO_SKIP_QUEUE,
     { localConcurrency: 1 },
     handleIntakeAutoSkip,
@@ -712,48 +758,64 @@ export async function registerMaintenanceQueues(
   // v1.4.34 — Apple Health export.zip ingest worker. localConcurrency
   // caps at 1 because the parse loop is CPU-bound and a concurrent
   // second 1 GB import would race the first for RSS.
-  await boss.work<AppleHealthImportPayload>(
+  await createAndWork<AppleHealthImportPayload>(
+    boss,
     APPLE_HEALTH_IMPORT_V2_QUEUE,
     { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY },
     async (jobs) => {
       // pg-boss v12 work callbacks always receive an array (batched
       // worker mode); for our concurrency-1 + batchSize-1 case we just
       // process each job sequentially.
+      //
+      // Each import is independent work, so a failed one does not abort the
+      // rest of the batch; the first failure is what the job reports.
+      let failure: JobOutcome | null = null;
       for (const job of jobs) {
-        await handleAppleHealthImport(job);
+        const outcome = await handleAppleHealthImport(job);
+        if (!outcome.ok && failure === null) failure = outcome;
       }
+      return failure ?? jobDone({ jobs: jobs.length });
     },
   );
   // Drain pre-revision-2 backlog by moving each claimed legacy job onto the
   // isolated v2 queue. This preserves the ImportJob status id without letting
   // the current parser execute under a revision-1 mirror.
-  await boss.work<AppleHealthImportPayload>(
+  await createAndWork<AppleHealthImportPayload>(
+    boss,
     APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
     { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY },
     async (jobs) => {
       for (const job of jobs) {
         await migrateLegacyAppleHealthImport(job);
       }
+      return jobDone({ jobs: jobs.length });
     },
   );
   // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight:
   // the underlying `updateMany` is idempotent and two ticks racing the same
   // read-then-write pass would just be wasted work.
-  await boss.work(
+  await createAndWork(
+    boss,
     IMPORT_JOB_RECONCILE_QUEUE,
     { localConcurrency: 1 },
     handleImportJobReconcileTick,
   );
-  await boss.work<MedicationIntakeImportQueuePayload>(
+  await createAndWork<MedicationIntakeImportQueuePayload>(
+    boss,
     MEDICATION_INTAKE_IMPORT_QUEUE,
     {
       localConcurrency: MEDICATION_INTAKE_IMPORT_CONCURRENCY,
       includeMetadata: true,
     },
     async (jobs) => {
+      // Independent per-delivery imports: run them all, report the first
+      // failure rather than losing it behind a later success.
+      let failure: JobOutcome | null = null;
       for (const job of jobs) {
-        await handleMedicationIntakeImport(job);
+        const outcome = await handleMedicationIntakeImport(job);
+        if (!outcome.ok && failure === null) failure = outcome;
       }
+      return failure ?? jobDone({ jobs: jobs.length });
     },
   );
 
@@ -768,10 +830,21 @@ export async function registerMaintenanceQueues(
   // above) whose payload omits `userId`. That tick runs the SAME discovery
   // fan-out the boot pass uses, so duplicate slots created between deploys
   // collapse within a day instead of waiting for the next worker reboot.
-  await boss.work<Partial<IntakeSlotDedupPayload>>(
+  await createAndWork<Partial<IntakeSlotDedupPayload>>(
+    boss,
     INTAKE_SLOT_DEDUP_QUEUE,
     { localConcurrency: INTAKE_SLOT_DEDUP_CONCURRENCY },
     async (jobs) => {
+      let users = 0;
+      let discoveryEnqueued = 0;
+      let discoverySkipped = 0;
+      // The discovery helper returns its error instead of throwing, so a
+      // failed fan-out would otherwise leave no trace on the job at all.
+      let discoveryFailed = 0;
+      let slotsCollapsed = 0;
+      let rowsSoftDeleted = 0;
+      let rowsNormalised = 0;
+      let daysRecomputed = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         if (!userId) {
@@ -779,6 +852,9 @@ export async function registerMaintenanceQueues(
           // still holding duplicate-slot candidates (singletonKey-coalesced,
           // identical to the boot pass).
           const result = await enqueueBootTimeIntakeSlotDedup();
+          discoveryEnqueued += result.enqueued;
+          discoverySkipped += result.skipped;
+          if (result.error) discoveryFailed++;
           workerLog(
             "info",
             `[intake-slot-dedup] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`,
@@ -787,6 +863,11 @@ export async function registerMaintenanceQueues(
         }
         try {
           const summary = await dedupeUserIntakeSlots(userId);
+          users++;
+          slotsCollapsed += summary.slotsCollapsed;
+          rowsSoftDeleted += summary.rowsSoftDeleted;
+          rowsNormalised += summary.rowsNormalised;
+          daysRecomputed += summary.daysRecomputed;
           workerLog(
             "info",
             `[intake-slot-dedup] user=${userId} slotsCollapsed=${summary.slotsCollapsed} rowsSoftDeleted=${summary.rowsSoftDeleted} rowsNormalised=${summary.rowsNormalised} daysRecomputed=${summary.daysRecomputed}`,
@@ -797,6 +878,17 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        users,
+        discovery_enqueued: discoveryEnqueued,
+        discovery_skipped: discoverySkipped,
+        discovery_failed: discoveryFailed,
+        slots_collapsed: slotsCollapsed,
+        rows_soft_deleted: rowsSoftDeleted,
+        rows_normalised: rowsNormalised,
+        days_recomputed: daysRecomputed,
+      });
     },
   );
 
@@ -806,14 +898,26 @@ export async function registerMaintenanceQueues(
   // empty-userId payload is the daily discovery tick and re-fans the per-user
   // jobs (singletonKey-coalesced, identical to the boot pass). Serial
   // concurrency so the migration never crowds the request pool.
-  await boss.work<Partial<NoteEncryptionBackfillPayload>>(
+  await createAndWork<Partial<NoteEncryptionBackfillPayload>>(
+    boss,
     NOTE_ENCRYPTION_BACKFILL_QUEUE,
     { localConcurrency: NOTE_ENCRYPTION_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let users = 0;
+      let discoveryEnqueued = 0;
+      let discoverySkipped = 0;
+      // The discovery helper returns its error instead of throwing, so a
+      // failed fan-out would otherwise leave no trace on the job at all.
+      let discoveryFailed = 0;
+      let measurements = 0;
+      let moodEntries = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         if (!userId) {
           const result = await enqueueBootTimeNoteEncryptionBackfill();
+          discoveryEnqueued += result.enqueued;
+          discoverySkipped += result.skipped;
+          if (result.error) discoveryFailed++;
           workerLog(
             "info",
             `[note-encryption-backfill] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`,
@@ -823,6 +927,9 @@ export async function registerMaintenanceQueues(
         try {
           const { measurementsMigrated, moodEntriesMigrated } =
             await runNoteEncryptionBackfillForUser(userId);
+          users++;
+          measurements += measurementsMigrated;
+          moodEntries += moodEntriesMigrated;
           workerLog(
             "info",
             `[note-encryption-backfill] user=${userId} measurements=${measurementsMigrated} moodEntries=${moodEntriesMigrated}`,
@@ -837,11 +944,21 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        users,
+        discovery_enqueued: discoveryEnqueued,
+        discovery_skipped: discoverySkipped,
+        discovery_failed: discoveryFailed,
+        measurements_migrated: measurements,
+        mood_entries_migrated: moodEntries,
+      });
     },
   );
 
   // v1.23 — admin-triggered encryption-key rotation (on-demand, singleton).
-  await boss.work<EncryptionKeyRotatePayload>(
+  await createAndWork<EncryptionKeyRotatePayload>(
+    boss,
     ENCRYPTION_KEY_ROTATE_QUEUE,
     { localConcurrency: ENCRYPTION_KEY_ROTATE_CONCURRENCY },
     handleEncryptionKeyRotate,
@@ -854,14 +971,27 @@ export async function registerMaintenanceQueues(
   // payload is the daily discovery tick and re-fans the per-user jobs
   // (singletonKey-coalesced, identical to the boot pass). Serial concurrency
   // so the migration never crowds the request pool.
-  await boss.work<Partial<MedNotesEncryptionBackfillPayload>>(
+  await createAndWork<Partial<MedNotesEncryptionBackfillPayload>>(
+    boss,
     MED_NOTES_ENCRYPTION_BACKFILL_QUEUE,
     { localConcurrency: MED_NOTES_ENCRYPTION_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let users = 0;
+      let discoveryEnqueued = 0;
+      let discoverySkipped = 0;
+      // The discovery helper returns its error instead of throwing, so a
+      // failed fan-out would otherwise leave no trace on the job at all.
+      let discoveryFailed = 0;
+      let sideEffects = 0;
+      let doseChanges = 0;
+      let inventoryItems = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         if (!userId) {
           const result = await enqueueBootTimeMedNotesEncryptionBackfill();
+          discoveryEnqueued += result.enqueued;
+          discoverySkipped += result.skipped;
+          if (result.error) discoveryFailed++;
           workerLog(
             "info",
             `[med-notes-encryption-backfill] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`,
@@ -874,6 +1004,10 @@ export async function registerMaintenanceQueues(
             doseChangesMigrated,
             inventoryItemsMigrated,
           } = await runMedNotesEncryptionBackfillForUser(userId);
+          users++;
+          sideEffects += sideEffectsMigrated;
+          doseChanges += doseChangesMigrated;
+          inventoryItems += inventoryItemsMigrated;
           workerLog(
             "info",
             `[med-notes-encryption-backfill] user=${userId} sideEffects=${sideEffectsMigrated} doseChanges=${doseChangesMigrated} inventoryItems=${inventoryItemsMigrated}`,
@@ -888,6 +1022,16 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        users,
+        discovery_enqueued: discoveryEnqueued,
+        discovery_skipped: discoverySkipped,
+        discovery_failed: discoveryFailed,
+        side_effects_migrated: sideEffects,
+        dose_changes_migrated: doseChanges,
+        inventory_items_migrated: inventoryItems,
+      });
     },
   );
 
@@ -895,16 +1039,25 @@ export async function registerMaintenanceQueues(
   // trigger endpoint sends one per-user job; this handler indexes that user's
   // not-yet-indexed documents (consent + budget gated, bounded + resumable).
   // Serial concurrency so the provider calls never crowd the request pool.
-  await boss.work<ContentIndexBackfillPayload>(
+  await createAndWork<ContentIndexBackfillPayload>(
+    boss,
     CONTENT_INDEX_BACKFILL_QUEUE,
     { localConcurrency: CONTENT_INDEX_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let users = 0;
+      let skipped = 0;
+      let documentsIndexed = 0;
       for (const job of jobs) {
         const { userId } = job.data;
-        if (!userId) continue;
+        if (!userId) {
+          skipped++;
+          continue;
+        }
         try {
           const { indexed, reason } =
             await runContentIndexBackfillForUser(userId);
+          users++;
+          documentsIndexed += indexed;
           workerLog(
             "info",
             `[document-content-index-backfill] user=${userId} indexed=${indexed} reason=${reason}`,
@@ -919,6 +1072,12 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        users,
+        skipped,
+        documents_indexed: documentsIndexed,
+      });
     },
   );
 
@@ -926,15 +1085,22 @@ export async function registerMaintenanceQueues(
   // (one job per stored document); provider-first (vision) when configured +
   // consented, else local text-layer extraction. Serial concurrency so the
   // provider calls + PDF parse never crowd the request pool.
-  await boss.work<DocumentIndexPayload>(
+  await createAndWork<DocumentIndexPayload>(
+    boss,
     DOCUMENT_INDEX_QUEUE,
     { localConcurrency: DOCUMENT_INDEX_CONCURRENCY },
     async (jobs) => {
+      let documents = 0;
+      let skipped = 0;
       for (const job of jobs) {
         const { userId, documentId } = job.data;
-        if (!userId || !documentId) continue;
+        if (!userId || !documentId) {
+          skipped++;
+          continue;
+        }
         try {
           await runDocumentIndex(job.data);
+          documents++;
         } catch (err) {
           recordError();
           workerLog(
@@ -945,6 +1111,7 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({ jobs: jobs.length, documents, skipped });
     },
   );
 
@@ -952,15 +1119,22 @@ export async function registerMaintenanceQueues(
   // upload (one job per stored document); decodes + downscales the original to
   // a small encrypted JPEG. Pure local compute; serial concurrency so the
   // canvas/pdfjs decode never crowds the request pool.
-  await boss.work<DocumentThumbnailPayload>(
+  await createAndWork<DocumentThumbnailPayload>(
+    boss,
     DOCUMENT_THUMBNAIL_QUEUE,
     { localConcurrency: DOCUMENT_THUMBNAIL_CONCURRENCY },
     async (jobs) => {
+      let documents = 0;
+      let skipped = 0;
       for (const job of jobs) {
         const { userId, documentId } = job.data;
-        if (!userId || !documentId) continue;
+        if (!userId || !documentId) {
+          skipped++;
+          continue;
+        }
         try {
           await runDocumentThumbnail(job.data);
+          documents++;
         } catch (err) {
           recordError();
           workerLog(
@@ -971,6 +1145,7 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({ jobs: jobs.length, documents, skipped });
     },
   );
 
@@ -979,15 +1154,22 @@ export async function registerMaintenanceQueues(
   // the `documentsAutoAiRead` opt-in is ON (egress consent + budget gated) and
   // persists the summary encrypted. Serial concurrency so the provider call
   // never crowds the request pool.
-  await boss.work<DocumentSummaryPayload>(
+  await createAndWork<DocumentSummaryPayload>(
+    boss,
     DOCUMENT_SUMMARY_QUEUE,
     { localConcurrency: DOCUMENT_SUMMARY_CONCURRENCY },
     async (jobs) => {
+      let documents = 0;
+      let skipped = 0;
       for (const job of jobs) {
         const { userId, documentId } = job.data;
-        if (!userId || !documentId) continue;
+        if (!userId || !documentId) {
+          skipped++;
+          continue;
+        }
         try {
           await runDocumentSummaryJob(job.data);
+          documents++;
         } catch (err) {
           recordError();
           workerLog(
@@ -998,6 +1180,7 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({ jobs: jobs.length, documents, skipped });
     },
   );
 
@@ -1007,15 +1190,24 @@ export async function registerMaintenanceQueues(
   // documents (the upload-time enqueue no-opped while the flag was OFF). It
   // only enqueues — the opt-in, egress consent and budget gates all still run
   // per document inside the summary job. Serial concurrency.
-  await boss.work<SummaryCatchUpPayload>(
+  await createAndWork<SummaryCatchUpPayload>(
+    boss,
     DOCUMENT_SUMMARY_CATCHUP_QUEUE,
     { localConcurrency: DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY },
     async (jobs) => {
+      let users = 0;
+      let skipped = 0;
+      let summariesEnqueued = 0;
       for (const job of jobs) {
         const { userId } = job.data;
-        if (!userId) continue;
+        if (!userId) {
+          skipped++;
+          continue;
+        }
         try {
-          await runSummaryCatchUpForUser(userId);
+          const { enqueued } = await runSummaryCatchUpForUser(userId);
+          users++;
+          summariesEnqueued += enqueued;
         } catch (err) {
           recordError();
           workerLog(
@@ -1026,6 +1218,12 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        users,
+        skipped,
+        summaries_enqueued: summariesEnqueued,
+      });
     },
   );
 
@@ -1033,15 +1231,24 @@ export async function registerMaintenanceQueues(
   // discovery sends one per-user job; this handler fans out per-document
   // thumbnail jobs for that user's not-yet-thumbnailed documents. Serial
   // concurrency; cheap discovery + enqueue only.
-  await boss.work<ThumbnailBackfillPayload>(
+  await createAndWork<ThumbnailBackfillPayload>(
+    boss,
     DOCUMENT_THUMBNAIL_BACKFILL_QUEUE,
     { localConcurrency: DOCUMENT_THUMBNAIL_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let users = 0;
+      let skipped = 0;
+      let thumbnailsEnqueued = 0;
       for (const job of jobs) {
         const { userId } = job.data;
-        if (!userId) continue;
+        if (!userId) {
+          skipped++;
+          continue;
+        }
         try {
           const { enqueued } = await runThumbnailBackfillForUser(userId);
+          users++;
+          thumbnailsEnqueued += enqueued;
           workerLog(
             "info",
             `[document-thumbnail-backfill] user=${userId} enqueued=${enqueued}`,
@@ -1056,6 +1263,12 @@ export async function registerMaintenanceQueues(
           throw err;
         }
       }
+      return jobDone({
+        jobs: jobs.length,
+        users,
+        skipped,
+        thumbnails_enqueued: thumbnailsEnqueued,
+      });
     },
   );
 
@@ -1063,13 +1276,20 @@ export async function registerMaintenanceQueues(
   // payload) fans out one per-user job per opted-in account; the queue also
   // serves the on-demand backfill payloads from the settings surface. Serial
   // concurrency so the staggered outbound fetches never crowd the request pool.
-  await boss.work<EnvironmentFetchPayload>(
+  await createAndWork<EnvironmentFetchPayload>(
+    boss,
     ENVIRONMENT_FETCH_QUEUE,
     { localConcurrency: ENVIRONMENT_FETCH_CONCURRENCY },
     async (jobs) => {
+      // A discovery tick and a per-user backfill share this queue and are
+      // independent of each other, so the pass runs to the end and reports
+      // the first failure it saw.
+      let failure: JobOutcome | null = null;
       for (const job of jobs) {
-        await handleEnvironmentFetch(boss, job.data);
+        const outcome = await handleEnvironmentFetch(boss, job.data);
+        if (!outcome.ok && failure === null) failure = outcome;
       }
+      return failure ?? jobDone({ jobs: jobs.length });
     },
   );
 

src/lib/jobs/reminder/register-rollup.ts

@@ -10,7 +10,7 @@
  *
  * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
  * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
- * and a `boss.work(QUEUE, …, handler)` binding drains it. The
+ * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The
  * `drain-cumulative-queue`, `mean-consolidation-queue`, and
  * `stress-strain-retention-queue` (dense-retention facet) guards read THIS
  * module.
@@ -104,9 +104,12 @@ import {
 } from "./shared";
 import {
   createAndSchedule,
+  createAndWork,
+  cronIsTheRetry,
   type QueuePolicyTable,
   type ScheduleEntry,
 } from "./registrar-shared";
+import { jobDone, jobFailed } from "@/lib/jobs/job-outcome";
 // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows.
 // Collapses each user × cumulative-type × calendar-day bucket into one
 // `stats:…` row so the list view stops painting hundreds of step chunks per
@@ -170,7 +173,7 @@ const schedules: ScheduleEntry[] = [
   // v1.4.37 W7c — nightly fold of per-sample APPLE_HEALTH cumulative
   // rows into one row per day per type. Slots between the
   // audit-log cleanup (03:15) and the feedback aggregator (04:00).
-  [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON],
+  [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON, cronIsTheRetry],
 ];
 
 /**
@@ -274,7 +277,8 @@ export async function registerRollupQueues(
   // the DAY bucket is already recomputed synchronously by the hook
   // itself. Concurrency-2 keeps two recomputes in flight without
   // crowding the dashboard request pool.
-  await boss.work<RollupRecomputePayload>(
+  await createAndWork<RollupRecomputePayload>(
+    boss,
     ROLLUP_RECOMPUTE_QUEUE,
     { localConcurrency: ROLLUP_RECOMPUTE_CONCURRENCY },
     async (jobs) => {
@@ -287,6 +291,10 @@ export async function registerRollupQueues(
           to: new Date(payload.to),
         });
       }
+      // No per-bucket isolation here on purpose: a recompute that throws
+      // fails the job, so reaching this line means every bucket in the batch
+      // was folded.
+      return jobDone({ buckets_recomputed: jobs.length });
     },
   );
 
@@ -295,18 +303,25 @@ export async function registerRollupQueues(
   // full `recomputeUserRollups` against the default 5-year window
   // across every granularity. Serial concurrency so the populator
   // never crowds the dashboard request pool.
-  await boss.work<RollupFullBackfillPayload>(
+  await createAndWork<RollupFullBackfillPayload>(
+    boss,
     ROLLUP_FULL_BACKFILL_QUEUE,
     { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let rowsUpsertedTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         const { rowsUpserted, durationMs } = await recomputeUserRollups(userId);
+        rowsUpsertedTotal += rowsUpserted;
         workerLog(
           "info",
           `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
         );
       }
+      return jobDone({
+        users: jobs.length,
+        rows_upserted: rowsUpsertedTotal,
+      });
     },
   );
 
@@ -315,19 +330,29 @@ export async function registerRollupQueues(
   // step rows; this handler collapses them into one daily-total row per
   // calendar day and soft-deletes the originals. Serial concurrency so
   // the populator never crowds the dashboard request pool.
-  await boss.work<StepConsolidationPayload>(
+  await createAndWork<StepConsolidationPayload>(
+    boss,
     STEP_CONSOLIDATION_QUEUE,
     { localConcurrency: STEP_CONSOLIDATION_CONCURRENCY },
     async (jobs) => {
+      let daysTotal = 0;
+      let legacyRowsTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         const { daysConsolidated, legacyRowsSoftDeleted } =
           await runStepConsolidationForUser(userId);
+        daysTotal += daysConsolidated;
+        legacyRowsTotal += legacyRowsSoftDeleted;
         workerLog(
           "info",
           `[step-consolidation] user=${userId} days=${daysConsolidated} legacyRowsSoftDeleted=${legacyRowsSoftDeleted}`,
         );
       }
+      return jobDone({
+        users: jobs.length,
+        days_consolidated: daysTotal,
+        legacy_rows_soft_deleted: legacyRowsTotal,
+      });
     },
   );
 
@@ -335,18 +360,39 @@ export async function registerRollupQueues(
   // GOOGLE_HEALTH / FITBIT daily-total step rows the pre-fix consolidation
   // swept and removes the shadow MANUAL totals it minted. Serial
   // concurrency so the repair never crowds the dashboard request pool.
-  await boss.work<StepConsolidationRepairPayload>(
+  await createAndWork<StepConsolidationRepairPayload>(
+    boss,
     STEP_CONSOLIDATION_REPAIR_QUEUE,
     { localConcurrency: STEP_CONSOLIDATION_REPAIR_CONCURRENCY },
     async (jobs) => {
+      let rowsResurrected = 0;
+      let wedgeSkipped = 0;
+      let manualMintsRemoved = 0;
+      let daysRecomputed = 0;
+      let failures = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         const summary = await runStepConsolidationRepairForUser(userId);
+        rowsResurrected += summary.rowsResurrected;
+        wedgeSkipped += summary.wedgeSkipped;
+        manualMintsRemoved += summary.manualMintsRemoved;
+        daysRecomputed += summary.daysRecomputed;
+        // The repair isolates its own per-day failures and reports them as a
+        // count, so they ride out as a fact rather than failing the job.
+        failures += summary.failures;
         workerLog(
           "info",
           `[step-consolidation-repair] user=${userId} resurrected=${summary.rowsResurrected} wedgeSkipped=${summary.wedgeSkipped} manualMintsRemoved=${summary.manualMintsRemoved} daysRecomputed=${summary.daysRecomputed} failures=${summary.failures}`,
         );
       }
+      return jobDone({
+        users: jobs.length,
+        rows_resurrected: rowsResurrected,
+        wedge_skipped: wedgeSkipped,
+        manual_mints_removed: manualMintsRemoved,
+        days_recomputed: daysRecomputed,
+        failures,
+      });
     },
   );
 
@@ -356,18 +402,28 @@ export async function registerRollupQueues(
   // it and re-runs detection silently so the honest re-derived best takes
   // its place. Serial concurrency so the repair never crowds the
   // dashboard request pool.
-  await boss.work<CumulativePrRederivePayload>(
+  await createAndWork<CumulativePrRederivePayload>(
+    boss,
     CUMULATIVE_PR_REDERIVE_QUEUE,
     { localConcurrency: CUMULATIVE_PR_REDERIVE_CONCURRENCY },
     async (jobs) => {
+      let rowsDeleted = 0;
+      let rowsReinserted = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         const summary = await runCumulativePrRederivationForUser(userId);
+        rowsDeleted += summary.rowsDeleted;
+        rowsReinserted += summary.rowsReinserted;
         workerLog(
           "info",
           `[cumulative-pr-rederive] user=${userId} rowsDeleted=${summary.rowsDeleted} rowsReinserted=${summary.rowsReinserted}`,
         );
       }
+      return jobDone({
+        users: jobs.length,
+        rows_deleted: rowsDeleted,
+        rows_reinserted: rowsReinserted,
+      });
     },
   );
 
@@ -376,19 +432,29 @@ export async function registerRollupQueues(
   // mean-type rows; this handler collapses each completed day to its
   // mean and soft-deletes the originals. Serial concurrency so the
   // populator never crowds the dashboard request pool.
-  await boss.work<MeanConsolidationPayload>(
+  await createAndWork<MeanConsolidationPayload>(
+    boss,
     MEAN_CONSOLIDATION_QUEUE,
     { localConcurrency: MEAN_CONSOLIDATION_CONCURRENCY },
     async (jobs) => {
+      let daysTotal = 0;
+      let perSampleRowsTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         const { daysConsolidated, perSampleRowsSoftDeleted } =
           await runMeanConsolidationForUser(userId);
+        daysTotal += daysConsolidated;
+        perSampleRowsTotal += perSampleRowsSoftDeleted;
         workerLog(
           "info",
           `[mean-consolidation] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`,
         );
       }
+      return jobDone({
+        users: jobs.length,
+        days_consolidated: daysTotal,
+        per_sample_rows_soft_deleted: perSampleRowsTotal,
+      });
     },
   );
 
@@ -399,10 +465,14 @@ export async function registerRollupQueues(
   // soft-deletes the originals, keeping the in-window intra-day shape intact
   // for the Stress engine. Serial concurrency so the backfill never crowds
   // the request pool.
-  await boss.work<DenseIntradayRetentionPayload>(
+  await createAndWork<DenseIntradayRetentionPayload>(
+    boss,
     DENSE_INTRADAY_RETENTION_QUEUE,
     { localConcurrency: DENSE_INTRADAY_RETENTION_CONCURRENCY },
     async (jobs) => {
+      let daysTotal = 0;
+      let perSampleRowsTotal = 0;
+      let derivedRestingRowsTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         try {
@@ -411,6 +481,9 @@ export async function registerRollupQueues(
             perSampleRowsSoftDeleted,
             derivedRestingRowsUpserted,
           } = await runDenseIntradayRetentionForUser(userId);
+          daysTotal += daysConsolidated;
+          perSampleRowsTotal += perSampleRowsSoftDeleted;
+          derivedRestingRowsTotal += derivedRestingRowsUpserted;
           workerLog(
             "info",
             `[dense-intraday-retention] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted} derivedRestingRowsUpserted=${derivedRestingRowsUpserted}`,
@@ -425,6 +498,12 @@ export async function registerRollupQueues(
           throw err;
         }
       }
+      return jobDone({
+        users: jobs.length,
+        days_consolidated: daysTotal,
+        per_sample_rows_soft_deleted: perSampleRowsTotal,
+        derived_resting_rows_upserted: derivedRestingRowsTotal,
+      });
     },
   );
 
@@ -435,10 +514,15 @@ export async function registerRollupQueues(
   // Self-converging: a rebuilt day drops out of the discovery pairing, so
   // the pass runs once per install. Serial concurrency so the rebuild never
   // crowds the request pool.
-  await boss.work<DenseIntradayHourlyRebuildPayload>(
+  await createAndWork<DenseIntradayHourlyRebuildPayload>(
+    boss,
     DENSE_INTRADAY_HOURLY_REBUILD_QUEUE,
     { localConcurrency: DENSE_INTRADAY_HOURLY_REBUILD_CONCURRENCY },
     async (jobs) => {
+      let daysRebuiltTotal = 0;
+      let hourlyRowsTotal = 0;
+      let dailyRowsRetiredTotal = 0;
+      let daysSkippedTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         try {
@@ -448,6 +532,10 @@ export async function registerRollupQueues(
             dailyRowsRetired,
             daysSkippedNoTombstones,
           } = await runDenseIntradayHourlyRebuildForUser(userId);
+          daysRebuiltTotal += daysRebuilt;
+          hourlyRowsTotal += hourlyRowsUpserted;
+          dailyRowsRetiredTotal += dailyRowsRetired;
+          daysSkippedTotal += daysSkippedNoTombstones;
           workerLog(
             "info",
             `[dense-intraday-hourly-rebuild] user=${userId} daysRebuilt=${daysRebuilt} hourlyRowsUpserted=${hourlyRowsUpserted} dailyRowsRetired=${dailyRowsRetired} daysSkippedNoTombstones=${daysSkippedNoTombstones}`,
@@ -462,6 +550,13 @@ export async function registerRollupQueues(
           throw err;
         }
       }
+      return jobDone({
+        users: jobs.length,
+        days_rebuilt: daysRebuiltTotal,
+        hourly_rows_upserted: hourlyRowsTotal,
+        daily_rows_retired: dailyRowsRetiredTotal,
+        days_skipped_no_tombstones: daysSkippedTotal,
+      });
     },
   );
 
@@ -471,7 +566,8 @@ export async function registerRollupQueues(
   // No current read path consumes these buckets — they exist so a
   // future cross-granularity reader can ship without a backfill
   // step. Concurrency-2 mirrors the measurement-rollup worker.
-  await boss.work<MoodRollupRecomputePayload>(
+  await createAndWork<MoodRollupRecomputePayload>(
+    boss,
     MOOD_ROLLUP_RECOMPUTE_QUEUE,
     { localConcurrency: MOOD_ROLLUP_RECOMPUTE_CONCURRENCY },
     async (jobs) => {
@@ -483,6 +579,7 @@ export async function registerRollupQueues(
           to: new Date(payload.to),
         });
       }
+      return jobDone({ buckets_recomputed: jobs.length });
     },
   );
 
@@ -491,10 +588,12 @@ export async function registerRollupQueues(
   // but zero rollup rows; this handler folds the full 5-year window
   // across every granularity. Concurrency-1 so the populator never
   // crowds the request pool.
-  await boss.work<MoodRollupFullBackfillPayload>(
+  await createAndWork<MoodRollupFullBackfillPayload>(
+    boss,
     MOOD_ROLLUP_FULL_BACKFILL_QUEUE,
     { localConcurrency: MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let rowsUpsertedTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         try {
@@ -508,6 +607,7 @@ export async function registerRollupQueues(
             userId,
             { replace: true },
           );
+          rowsUpsertedTotal += rowsUpserted;
           workerLog(
             "info",
             `[mood-rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
@@ -522,6 +622,10 @@ export async function registerRollupQueues(
           throw err;
         }
       }
+      return jobDone({
+        users: jobs.length,
+        rows_upserted: rowsUpsertedTotal,
+      });
     },
   );
 
@@ -530,15 +634,18 @@ export async function registerRollupQueues(
   // but zero rollup coverage; this handler folds the trailing 90-day
   // window per account. Concurrency-1 so the populator never crowds
   // the request pool.
-  await boss.work<MedicationComplianceBackfillPayload>(
+  await createAndWork<MedicationComplianceBackfillPayload>(
+    boss,
     MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
     { localConcurrency: MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY },
     async (jobs) => {
+      let rowsUpsertedTotal = 0;
       for (const job of jobs) {
         const { userId } = job.data;
         try {
           const { rowsUpserted, durationMs } =
             await recomputeUserMedicationCompliance(userId);
+          rowsUpsertedTotal += rowsUpserted;
           workerLog(
             "info",
             `[medication-compliance-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
@@ -553,6 +660,10 @@ export async function registerRollupQueues(
           throw err;
         }
       }
+      return jobDone({
+        users: jobs.length,
+        rows_upserted: rowsUpsertedTotal,
+      });
     },
   );
 
@@ -563,10 +674,26 @@ export async function registerRollupQueues(
   // Concurrency-1 so the drain never crowds the dashboard request pool
   // and a long backfill on the maintainer's account (300 k+ measurement rows)
   // stays a single sequential walk.
-  await boss.work<DrainCumulativePayload>(
+  await createAndWork<DrainCumulativePayload>(
+    boss,
     DRAIN_CUMULATIVE_QUEUE,
     { localConcurrency: 1 },
     async (jobs) => {
+      // The three drains below are independent passes sharing one nightly
+      // tick, so one failing pass must not stop the other two from running.
+      // The first failure is held back and reported once the tick is over —
+      // that keeps the existing order and still fails the job, which is what
+      // the swallowed errors never did.
+      let firstFailure: { reason: string; cause: unknown } | undefined;
+      let passesFailed = 0;
+      let bucketsCollapsed = 0;
+      let perSampleRowsDeleted = 0;
+      let meanDaysConsolidated = 0;
+      let meanRowsSoftDeleted = 0;
+      let denseDaysConsolidated = 0;
+      let denseRowsSoftDeleted = 0;
+      let denseSkipped = false;
+
       for (const job of jobs) {
         try {
           const summary = await drainPerSampleCumulative(getWorkerPrisma(), {
@@ -574,6 +701,8 @@ export async function registerRollupQueues(
             cutoffHours: DRAIN_CUMULATIVE_CUTOFF_HOURS,
             log: (line) => workerLog("info", line),
           });
+          bucketsCollapsed += summary.totals.bucketsCollapsed;
+          perSampleRowsDeleted += summary.totals.perSampleRowsDeleted;
           workerLog(
             "info",
             `[drain-cumulative] triggeredAt=${job.data.triggeredAt} usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}`,
@@ -581,6 +710,8 @@ export async function registerRollupQueues(
         } catch (err) {
           recordError();
           await reportWorkerError("drain-cumulative", err);
+          passesFailed++;
+          firstFailure ??= { reason: "cumulative drain failed", cause: err };
         }
 
         // v1.8.5 — fold the daily-MEAN drain onto the same nightly tick as
@@ -600,6 +731,8 @@ export async function registerRollupQueues(
             cutoffHours: MEAN_CONSOLIDATION_CUTOFF_HOURS,
             log: (line) => workerLog("info", line),
           });
+          meanDaysConsolidated += meanSummary.totals.daysConsolidated;
+          meanRowsSoftDeleted += meanSummary.totals.perSampleRowsSoftDeleted;
           workerLog(
             "info",
             `[mean-consolidation] triggeredAt=${job.data.triggeredAt} usersScanned=${meanSummary.totals.usersScanned} daysConsolidated=${meanSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${meanSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${meanSummary.totals.dailyRowsUpserted} daysFailed=${meanSummary.totals.daysFailed}`,
@@ -609,6 +742,8 @@ export async function registerRollupQueues(
           await reportWorkerError("mean-consolidation", err, {
             tick: "nightly",
           });
+          passesFailed++;
+          firstFailure ??= { reason: "mean consolidation failed", cause: err };
         }
 
         // v1.10.0 WX-E — fold the dense intra-day retention drain onto the
@@ -624,6 +759,7 @@ export async function registerRollupQueues(
         // shipped.
         try {
           if (!DENSE_INTRADAY_RETENTION_ENABLED) {
+            denseSkipped = true;
             workerLog(
               "info",
               "[dense-intraday-retention] disabled via DENSE_INTRADAY_RETENTION_ENABLED — skipping the nightly walk (operator kill-switch)",
@@ -636,6 +772,9 @@ export async function registerRollupQueues(
                 log: (line) => workerLog("info", line),
               },
             );
+            denseDaysConsolidated += denseSummary.totals.daysConsolidated;
+            denseRowsSoftDeleted +=
+              denseSummary.totals.perSampleRowsSoftDeleted;
             workerLog(
               "info",
               `[dense-intraday-retention] triggeredAt=${job.data.triggeredAt} usersScanned=${denseSummary.totals.usersScanned} daysConsolidated=${denseSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${denseSummary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${denseSummary.totals.hourlyRowsUpserted} dailyRowsRetired=${denseSummary.totals.dailyRowsRetired}`,
@@ -646,8 +785,26 @@ export async function registerRollupQueues(
           await reportWorkerError("dense-intraday-retention", err, {
             tick: "nightly",
           });
+          passesFailed++;
+          firstFailure ??= { reason: "dense retention failed", cause: err };
         }
       }
+
+      const did = {
+        passes_failed: passesFailed,
+        buckets_collapsed: bucketsCollapsed,
+        per_sample_rows_deleted: perSampleRowsDeleted,
+        mean_days_consolidated: meanDaysConsolidated,
+        mean_rows_soft_deleted: meanRowsSoftDeleted,
+        dense_days_consolidated: denseDaysConsolidated,
+        dense_rows_soft_deleted: denseRowsSoftDeleted,
+        dense_skipped: denseSkipped,
+      };
+
+      if (firstFailure) {
+        return jobFailed(firstFailure.reason, firstFailure.cause, did);
+      }
+      return jobDone(did);
     },
   );
 

src/lib/jobs/reminder/register-status.ts

@@ -10,8 +10,8 @@
  *
  * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
  * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
- * and a `boss.work(QUEUE, …, handler)` binding drains it. The status-wiring
- * guards (`status-cron-candidates`, `insight-pregenerate`,
+ * and a `createAndWork(boss, QUEUE, …, handler)` binding drains it. The
+ * status-wiring guards (`status-cron-candidates`, `insight-pregenerate`,
  * `insight-status-generate`, `recovery-score-queue`,
  * `stress-strain-retention-queue`, `period-narrative-warm`,
  * `coach-memory-refresh-queue`) read THIS module.
@@ -101,9 +101,11 @@ import {
   type InsightStatusGeneratePayload,
 } from "@/lib/jobs/insight-status-generate";
 import { withBackgroundEvent } from "@/lib/logging/background";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { getWorkerPrisma, workerLog } from "./shared";
 import {
   createAndSchedule,
+  createAndWork,
   insightRetryOptions,
   type QueuePolicyTable,
   type ScheduleEntry,
@@ -363,37 +365,44 @@ export async function registerStatusQueues(
 ): Promise<readonly string[]> {
   await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
-  await boss.work<GeneralStatusPayload>(
+  await createAndWork<GeneralStatusPayload>(
+    boss,
     GENERAL_STATUS_QUEUE,
     { localConcurrency: 1 },
     handleGeneralStatusGenerate,
   );
-  await boss.work<BloodPressureStatusPayload>(
+  await createAndWork<BloodPressureStatusPayload>(
+    boss,
     BLOOD_PRESSURE_STATUS_QUEUE,
     { localConcurrency: 1 },
     handleBloodPressureStatusGenerate,
   );
-  await boss.work<WeightStatusPayload>(
+  await createAndWork<WeightStatusPayload>(
+    boss,
     WEIGHT_STATUS_QUEUE,
     { localConcurrency: 1 },
     handleWeightStatusGenerate,
   );
-  await boss.work<PulseStatusPayload>(
+  await createAndWork<PulseStatusPayload>(
+    boss,
     PULSE_STATUS_QUEUE,
     { localConcurrency: 1 },
     handlePulseStatusGenerate,
   );
-  await boss.work<BmiStatusPayload>(
+  await createAndWork<BmiStatusPayload>(
+    boss,
     BMI_STATUS_QUEUE,
     { localConcurrency: 1 },
     handleBmiStatusGenerate,
   );
-  await boss.work<MoodStatusPayload>(
+  await createAndWork<MoodStatusPayload>(
+    boss,
     MOOD_STATUS_QUEUE,
     { localConcurrency: 1 },
     handleMoodStatusGenerate,
   );
-  await boss.work<MedicationComplianceStatusPayload>(
+  await createAndWork<MedicationComplianceStatusPayload>(
+    boss,
     MEDICATION_COMPLIANCE_STATUS_QUEUE,
     { localConcurrency: 1 },
     handleMedicationComplianceStatusGenerate,
@@ -409,7 +418,8 @@ export async function registerStatusQueues(
   // (the cohort walk is itself sequential per user, and the content-hash
   // gate + per-user budget gate inside the handler cover the rare
   // double-tick overlap the old single slot serialised away).
-  await boss.work<InsightPregeneratePayload>(
+  await createAndWork<InsightPregeneratePayload>(
+    boss,
     INSIGHT_PREGENERATE_QUEUE,
     { localConcurrency: 2 },
     handleInsightPregenerateJob,
@@ -420,17 +430,23 @@ export async function registerStatusQueues(
   // night's samples into one job, and the handler's marker re-check makes a
   // rare double-fire a no-op, so one slot is enough and bounds the provider
   // concurrency the same way the nightly path does.
-  await boss.work<MorningDigestRefreshPayload>(
+  await createAndWork<MorningDigestRefreshPayload>(
+    boss,
     MORNING_DIGEST_REFRESH_QUEUE,
     { localConcurrency: 1 },
-    async (jobs) => {
-      await withBackgroundEvent("job.morning_digest_refresh", async (evt) => {
+    async (jobs): Promise<JobOutcome> => {
+      return withBackgroundEvent("job.morning_digest_refresh", async (evt) => {
+        // The counts the batch already decides on: one refresh per job, and
+        // the finalised subset that reaches the morning push seam.
+        let refreshed = 0;
+        let finalised = 0;
         for (const job of jobs) {
           try {
             const result = await runMorningDigestRefresh(
               getWorkerPrisma(),
               job.data,
             );
+            refreshed++;
             evt.addMeta(
               "morning_refresh",
               `${result.status}:${result.comprehensive ?? "none"}`,
@@ -442,6 +458,7 @@ export async function registerStatusQueues(
             // there, so a user who never opted in or is outside the window is a
             // clean no-op. The fallback cron covers the no-sleep-arrived case.
             if (result.status === "finalised") {
+              finalised++;
               const pushResult = await maybeDispatchDailyBriefing(
                 getWorkerPrisma(),
                 job.data.userId,
@@ -457,6 +474,7 @@ export async function registerStatusQueues(
             throw err;
           }
         }
+        return jobDone({ refreshed, finalised });
       });
     },
   );
@@ -464,7 +482,8 @@ export async function registerStatusQueues(
   // read-only status route on a cold card. Low concurrency so a first
   // visit that cold-misses several cards can't saturate the Prisma pool
   // or fan out an unbounded number of concurrent provider calls.
-  await boss.work<InsightStatusGeneratePayload>(
+  await createAndWork<InsightStatusGeneratePayload>(
+    boss,
     INSIGHT_STATUS_GENERATE_QUEUE,
     { localConcurrency: INSIGHT_STATUS_GENERATE_CONCURRENCY },
     handleInsightStatusGenerate,
@@ -474,56 +493,84 @@ export async function registerStatusQueues(
   // eligible user and upserts one `COMPUTED RECOVERY_SCORE` row per scored
   // day (idempotent — a re-fire overwrites in place). Single-flight so two
   // ticks never double-walk the cohort.
-  await boss.work(RECOVERY_SCORE_QUEUE, { localConcurrency: 1 }, async () => {
-    try {
-      const summary = await runRecoveryScore(getWorkerPrisma());
-      workerLog(
-        "info",
-        `[recovery-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
-      );
-    } catch (err) {
-      recordError();
-      workerLog("error", "[recovery-score] pass failed", err);
-      throw err;
-    }
-  });
+  await createAndWork(
+    boss,
+    RECOVERY_SCORE_QUEUE,
+    { localConcurrency: 1 },
+    async (): Promise<JobOutcome> => {
+      try {
+        const summary = await runRecoveryScore(getWorkerPrisma());
+        workerLog(
+          "info",
+          `[recovery-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
+        );
+        return jobDone({
+          considered: summary.considered,
+          stored: summary.stored,
+          insufficient: summary.insufficient,
+          errored: summary.errored,
+        });
+      } catch (err) {
+        recordError();
+        workerLog("error", "[recovery-score] pass failed", err);
+        throw err;
+      }
+    },
+  );
   // v1.15.20 — proactive Coach nudge. Single-flight; the push-attempts
   // ledger caps a user at one nudge per rolling week, so an overlapping
   // tick would only waste reads. Deterministic triggers, no AI call.
-  await boss.work(COACH_NUDGE_QUEUE, { localConcurrency: 1 }, async () => {
-    await withBackgroundEvent("job.coach_nudge", async (evt) => {
-      try {
-        const summary = await runCoachNudgeTick(getWorkerPrisma(), new Date());
-        evt.setBackground({
-          task_name: "job.coach_nudge",
-          result: {
+  await createAndWork(
+    boss,
+    COACH_NUDGE_QUEUE,
+    { localConcurrency: 1 },
+    async (): Promise<JobOutcome> => {
+      return withBackgroundEvent("job.coach_nudge", async (evt) => {
+        try {
+          const summary = await runCoachNudgeTick(
+            getWorkerPrisma(),
+            new Date(),
+          );
+          evt.setBackground({
+            task_name: "job.coach_nudge",
+            result: {
+              candidates_scanned: summary.candidatesScanned,
+              dispatched: summary.dispatched,
+              persisted: summary.persisted,
+              skipped_opted_out: summary.skippedOptedOut,
+              skipped_no_provider: summary.skippedNoProvider,
+              skipped_recent_nudge: summary.skippedRecentNudge,
+              skipped_recent_engagement: summary.skippedRecentEngagement,
+              skipped_no_trigger: summary.skippedNoTrigger,
+              skipped_no_channel: summary.skippedNoChannel,
+              failed: summary.failed,
+            },
+          });
+          // A per-user failure never fails this pass — it rides out as the
+          // `failed` count, exactly as it does in the wide event.
+          return jobDone({
             candidates_scanned: summary.candidatesScanned,
             dispatched: summary.dispatched,
             persisted: summary.persisted,
-            skipped_opted_out: summary.skippedOptedOut,
-            skipped_no_provider: summary.skippedNoProvider,
-            skipped_recent_nudge: summary.skippedRecentNudge,
-            skipped_recent_engagement: summary.skippedRecentEngagement,
-            skipped_no_trigger: summary.skippedNoTrigger,
-            skipped_no_channel: summary.skippedNoChannel,
             failed: summary.failed,
-          },
-        });
-      } catch (err) {
-        evt.setError(err);
-        recordError();
-        throw err;
-      }
-    });
-  });
+          });
+        } catch (err) {
+          evt.setError(err);
+          recordError();
+          throw err;
+        }
+      });
+    },
+  );
   // v1.16.11 — medication low-stock pass. Single-flight; the
   // per-medication stamp makes a re-fire idempotent (already-notified
   // crossings skip), so an overlapping tick would only waste reads.
-  await boss.work(
+  await createAndWork(
+    boss,
     MEDICATION_LOW_STOCK_QUEUE,
     { localConcurrency: 1 },
-    async () => {
-      await withBackgroundEvent("job.medication_low_stock", async (evt) => {
+    async (): Promise<JobOutcome> => {
+      return withBackgroundEvent("job.medication_low_stock", async (evt) => {
         try {
           const summary = await runMedicationLowStockTick(
             getWorkerPrisma(),
@@ -544,6 +591,15 @@ export async function registerStatusQueues(
               failed: summary.failed,
             },
           });
+          // A per-medication failure never fails this pass — it rides out as
+          // the `failed` count.
+          return jobDone({
+            users_scanned: summary.usersScanned,
+            medications_evaluated: summary.medicationsEvaluated,
+            notified: summary.notified,
+            rearmed: summary.rearmed,
+            failed: summary.failed,
+          });
         } catch (err) {
           evt.setError(err);
           recordError();
@@ -556,7 +612,8 @@ export async function registerStatusQueues(
   // to claim a durable unique row, so two slots would only race each other for
   // a claim exactly one of them can win. It makes no provider call — a
   // module-graph test pins that — so one slot costs nothing in throughput.
-  await boss.work<DataArrival>(
+  await createAndWork<DataArrival>(
+    boss,
     DATA_ARRIVAL_QUEUE,
     { localConcurrency: 1 },
     handleDataArrival,
@@ -564,7 +621,8 @@ export async function registerStatusQueues(
   // v1.31.0 — the arrival reaction line. Single-flight: this is the spine's
   // ONE provider call, and a second concurrent slot would only race the same
   // durable `generatedAt` claim exactly one of them can win.
-  await boss.work<ReactionLineJob>(
+  await createAndWork<ReactionLineJob>(
+    boss,
     REACTION_LINE_QUEUE,
     { localConcurrency: 1 },
     handleReactionLine,
@@ -573,7 +631,8 @@ export async function registerStatusQueues(
   // path the spine deliberately does not walk, so it stays off the request pool
   // and cannot fan out concurrent completions when several workouts land in one
   // sync.
-  await boss.work<WorkoutInsightGeneratePayload>(
+  await createAndWork<WorkoutInsightGeneratePayload>(
+    boss,
     WORKOUT_INSIGHT_GENERATE_QUEUE,
     { localConcurrency: WORKOUT_INSIGHT_GENERATE_CONCURRENCY },
     handleWorkoutInsightGenerate,
@@ -583,84 +642,131 @@ export async function registerStatusQueues(
   // is suppressed), and the tick only loads a digest for opted-in users at
   // their exact local fallback hour. No provider call — it reads the cached
   // digest and dispatches the deterministic line + cached lead.
-  await boss.work(DAILY_BRIEFING_QUEUE, { localConcurrency: 1 }, async () => {
-    await withBackgroundEvent("job.daily_briefing", async (evt) => {
-      try {
-        const summary = await runDailyBriefingTick(
-          getWorkerPrisma(),
-          new Date(),
-        );
-        evt.setBackground({
-          task_name: "job.daily_briefing",
-          result: {
+  await createAndWork(
+    boss,
+    DAILY_BRIEFING_QUEUE,
+    { localConcurrency: 1 },
+    async (): Promise<JobOutcome> => {
+      return withBackgroundEvent("job.daily_briefing", async (evt) => {
+        try {
+          const summary = await runDailyBriefingTick(
+            getWorkerPrisma(),
+            new Date(),
+          );
+          evt.setBackground({
+            task_name: "job.daily_briefing",
+            result: {
+              candidates_scanned: summary.candidatesScanned,
+              in_slot: summary.inSlot,
+              sent: summary.sent,
+              suppressed_frequency: summary.suppressedFrequency,
+              no_digest: summary.noDigest,
+              opted_out: summary.optedOut,
+              module_off: summary.moduleOff,
+              no_channel: summary.noChannel,
+              outside_window: summary.outsideWindow,
+              failed: summary.failed,
+            },
+          });
+          // Most ticks land outside every user's slot; that is the tick doing
+          // its job, so a zero `sent` is a success with zero counts.
+          return jobDone({
             candidates_scanned: summary.candidatesScanned,
             in_slot: summary.inSlot,
             sent: summary.sent,
-            suppressed_frequency: summary.suppressedFrequency,
-            no_digest: summary.noDigest,
-            opted_out: summary.optedOut,
-            module_off: summary.moduleOff,
-            no_channel: summary.noChannel,
-            outside_window: summary.outsideWindow,
             failed: summary.failed,
-          },
-        });
-      } catch (err) {
-        evt.setError(err);
-        recordError();
-        throw err;
-      }
-    });
-  });
+          });
+        } catch (err) {
+          evt.setError(err);
+          recordError();
+          throw err;
+        }
+      });
+    },
+  );
   // v1.10.0 — computed scores (WX-E). Nightly Stress-score (HRV-derived
   // proxy) compute + store. Single-flight so two ticks never double-walk
   // the cohort. The runner iterates every eligible user and upserts one
   // `COMPUTED STRESS_SCORE` row per scored day (idempotent — a re-fire
   // overwrites in place).
-  await boss.work(STRESS_SCORE_QUEUE, { localConcurrency: 1 }, async () => {
-    try {
-      const summary = await runStressScore(getWorkerPrisma());
-      workerLog(
-        "info",
-        `[stress-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
-      );
-    } catch (err) {
-      recordError();
-      workerLog("error", "[stress-score] pass failed", err);
-      throw err;
-    }
-  });
+  await createAndWork(
+    boss,
+    STRESS_SCORE_QUEUE,
+    { localConcurrency: 1 },
+    async (): Promise<JobOutcome> => {
+      try {
+        const summary = await runStressScore(getWorkerPrisma());
+        workerLog(
+          "info",
+          `[stress-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
+        );
+        return jobDone({
+          considered: summary.considered,
+          stored: summary.stored,
+          insufficient: summary.insufficient,
+          errored: summary.errored,
+        });
+      } catch (err) {
+        recordError();
+        workerLog("error", "[stress-score] pass failed", err);
+        throw err;
+      }
+    },
+  );
   // v1.10.0 — computed scores (WX-E). Nightly Strain-score (Banister TRIMP
   // cardio-load) compute + store. Single-flight; upserts one `COMPUTED
   // STRAIN_SCORE` row per scored day (idempotent).
-  await boss.work(STRAIN_SCORE_QUEUE, { localConcurrency: 1 }, async () => {
-    try {
-      const summary = await runStrainScore(getWorkerPrisma());
-      workerLog(
-        "info",
-        `[strain-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
-      );
-    } catch (err) {
-      recordError();
-      workerLog("error", "[strain-score] pass failed", err);
-      throw err;
-    }
-  });
+  await createAndWork(
+    boss,
+    STRAIN_SCORE_QUEUE,
+    { localConcurrency: 1 },
+    async (): Promise<JobOutcome> => {
+      try {
+        const summary = await runStrainScore(getWorkerPrisma());
+        workerLog(
+          "info",
+          `[strain-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
+        );
+        return jobDone({
+          considered: summary.considered,
+          stored: summary.stored,
+          insufficient: summary.insufficient,
+          errored: summary.errored,
+        });
+      } catch (err) {
+        recordError();
+        workerLog("error", "[strain-score] pass failed", err);
+        throw err;
+      }
+    },
+  );
   // v1.11.0 — period-narrative warm. A scheduled tick (no `userId`) runs the
   // boundary-gated nightly fan-out; a `userId` payload runs a single-user warm
   // enqueued by the read-only GET on a cold/stale read. Single-flight so two
   // ticks never double-walk the cohort; the per-user budget gate covers the
   // fan-out and the enqueue `singletonKey` covers the single-user path.
-  await boss.work<PeriodNarrativePayload>(
+  await createAndWork<PeriodNarrativePayload>(
+    boss,
     PERIOD_NARRATIVE_QUEUE,
     { localConcurrency: 1 },
-    async (jobs) => {
+    async (jobs): Promise<JobOutcome> => {
+      // Two modes on one queue, so the outcome counts them separately: a
+      // non-boundary night is a scheduled pass that generated nothing, which
+      // is a success with zero counts, not an idle queue.
+      let singleUser = 0;
+      let scheduled = 0;
+      let generated = 0;
+      let failed = 0;
       for (const job of jobs) {
         try {
           if (job.data?.userId) {
             await warmOneNarrative(job.data);
+            singleUser++;
           } else {
             const summary = await runPeriodNarrativeWarm(getWorkerPrisma());
+            scheduled++;
+            generated += summary.generated;
+            failed += summary.failed;
             workerLog(
               "info",
               `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
@@ -674,26 +780,45 @@ export async function registerStatusQueues(
           throw err;
         }
       }
+      return jobDone({
+        single_user: singleUser,
+        scheduled,
+        generated,
+        failed,
+      });
     },
   );
   // v1.11.1 — combined Coach memory refresh: rolling conversation summary +
   // durable fact extraction for one long conversation. localConcurrency 1 so a
   // burst of long-conversation turns can't fan out concurrent provider calls;
   // each step is budget-gated inside runStatusCompletion and fault-isolated.
-  await boss.work<CoachMemoryRefreshPayload>(
+  await createAndWork<CoachMemoryRefreshPayload>(
+    boss,
     COACH_MEMORY_REFRESH_QUEUE,
     { localConcurrency: 1 },
-    async (jobs) => {
+    async (jobs): Promise<JobOutcome> => {
+      // An incomplete payload is skipped rather than failed, so it has to be
+      // countable — otherwise a queue full of them reads as a clean run.
+      let refreshed = 0;
+      let skippedIncomplete = 0;
       for (const job of jobs) {
-        if (!job.data?.conversationId || !job.data?.userId) continue;
+        if (!job.data?.conversationId || !job.data?.userId) {
+          skippedIncomplete++;
+          continue;
+        }
         try {
           await runCoachMemoryRefresh(job.data);
+          refreshed++;
         } catch (err) {
           recordError();
           workerLog("error", "[coach-memory-refresh] failed", err);
           throw err;
         }
       }
+      return jobDone({
+        refreshed,
+        skipped_incomplete: skippedIncomplete,
+      });
     },
   );
 
@@ -701,11 +826,12 @@ export async function registerStatusQueues(
   // reminder to `due` and minting plan-review reminders are both idempotent
   // across re-fires (an already-due reminder is not re-flipped; a plan whose
   // reviewDate was cleared is not re-minted). No provider call, no push.
-  await boss.work(
+  await createAndWork(
+    boss,
     COACH_REMINDER_SWEEP_QUEUE,
     { localConcurrency: 1 },
-    async () => {
-      await withBackgroundEvent("job.coach_reminder_sweep", async (evt) => {
+    async (): Promise<JobOutcome> => {
+      return withBackgroundEvent("job.coach_reminder_sweep", async (evt) => {
         try {
           const summary = await runCoachReminderSweep(
             getWorkerPrisma(),
@@ -719,6 +845,11 @@ export async function registerStatusQueues(
               errored: summary.errored,
             },
           });
+          return jobDone({
+            reminders_due: summary.remindersDue,
+            plan_reviews_minted: summary.planReviewsMinted,
+            errored: summary.errored,
+          });
         } catch (err) {
           evt.setError(err);
           recordError();
@@ -732,11 +863,12 @@ export async function registerStatusQueues(
   // reminders the sweep minted and writes the grounded before/after read-back
   // into the plan's encrypted outcome column, flipping it to `reviewed`.
   // Idempotent (a reviewed plan is not re-processed). No provider call, no push.
-  await boss.work(
+  await createAndWork(
+    boss,
     COACH_PLAN_REVIEW_QUEUE,
     { localConcurrency: 1 },
-    async () => {
-      await withBackgroundEvent("job.coach_plan_review", async (evt) => {
+    async (): Promise<JobOutcome> => {
+      return withBackgroundEvent("job.coach_plan_review", async (evt) => {
         try {
           const summary = await runCoachPlanReviewTick(
             getWorkerPrisma(),
@@ -750,6 +882,11 @@ export async function registerStatusQueues(
               errored: summary.errored,
             },
           });
+          return jobDone({
+            reviewed: summary.reviewed,
+            insufficient: summary.insufficient,
+            errored: summary.errored,
+          });
         } catch (err) {
           evt.setError(err);
           recordError();

src/lib/jobs/reminder/registrar-shared.ts

@@ -167,6 +167,32 @@ export const insightRetryOptions = {
   retryBackoff: true,
 } as const;
 
+/**
+ * Retry policy for a cron-driven pass whose failure mode is deterministic.
+ *
+ * Handlers now return a `JobOutcome`, and the retention cleanups that used to
+ * warn-and-return fail their job instead. That is the point — but it walks
+ * into pg-boss's queue default of `retryLimit: 2` with no delay and no
+ * backoff, so a bulk DELETE that timed out against a large trailing edge
+ * would run twice more, immediately, against the same rows. Three doomed
+ * sequential scans a night is a worse outcome than the silence it replaced.
+ *
+ * For this class the next cron tick IS the retry, and it arrives with the
+ * whole gap as backoff. So the schedule declares zero retries: the failure is
+ * recorded once, reaches the operator once, and the pass tries again on its
+ * own cadence.
+ *
+ * It rides on the SCHEDULE rather than on `createQueue`, deliberately.
+ * pg-boss's `create_queue()` ends in ON CONFLICT DO NOTHING, so a queue-level
+ * default would be inert on every already-running instance — the same trap
+ * `reconcileQueuePolicies` above exists to work around. Send options apply to
+ * every job the cron mints, new database or old.
+ *
+ * Use it only where the failure genuinely repeats. A transient failure (a TLS
+ * probe, an S3 fetch, a provider call) wants the default retries, not this.
+ */
+export const cronIsTheRetry = { retryLimit: 0 } as const;
+
 /**
  * Create every queue in `queues`, then schedule every cron in `schedules`.
  * Centralised so each registrar provisions before it schedules in the exact