feat(jobs): return an outcome from every named queue handler

Sixty-two handler functions across twenty-four modules now resolve to a
JobOutcome instead of undefined. The counts they already wrote to their
wide event ride out in `did`, so a run that deleted nothing and a run
that could not run are no longer the same shape.

Three rules decided each conversion:

  - a catch that swallowed becomes jobFailed, keeping its warning line;
  - a catch that already rethrew keeps its throw, because the throw
    already fails the job and rewrapping it buries the stack;
  - a fan-out pass is judged on the pass. One account's revoked grant
    belongs on that account's integration ledger, not in a queue-wide
    failure that retries the whole cohort, so per-leg failures ride out
    as a users_failed count — visible for the first time.

The retention cleanups are the behaviour change worth naming: every one
of them used to warn and return, so a purge that had not run for weeks
looked like a purge with nothing to do. They fail now. Their registrar
sets retryLimit accordingly in the commit that converts the bindings.

JobFailure carries the thrown error's stack; without it every one of the
101 queues would report the same runJob frame.

src/lib/jobs/apple-health-import-worker.ts

@@ -31,6 +31,7 @@ import type { Job } from "pg-boss";
 
 import { extractExportXml } from "@/lib/import/unzip-export-xml";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import {
   streamParseExportXml,
   type ImportJobProgress,
@@ -66,23 +67,26 @@ export const IMPORT_JOB_RECONCILE_CRON = "*/15 * * * *";
  * the same `reconcileOrphanImportJobs()` the boot path calls — the
  * heartbeat + live pg-boss-job check already make it safe to re-run
  * on a healthy worker (a live import's fresh heartbeat and `active`
- * pg-boss state both keep its row untouched). Errors are logged, not
- * rethrown — a reconcile miss must never spam pg-boss's retry/backoff
- * machinery for a background sweep this cheap to just re-run in 15
- * minutes.
+ * pg-boss state both keep its row untouched). A sweep that throws now
+ * reports a failed outcome instead of only leaving a warning behind:
+ * this pass exists to end a silent stall, and a stall in the pass
+ * itself was the one condition nothing surfaced. The next tick is 15
+ * minutes away regardless, so nothing rests on the retry.
  */
 export async function handleImportJobReconcileTick(
   jobs: Job<object>[],
-): Promise<void> {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent(
+  return withBackgroundEvent(
     "job.apple_health_import_reconcile",
     async (evt) => {
       try {
         await reconcileOrphanImportJobs();
       } catch (err) {
         evt.addWarning(`apple-health-import-reconcile failed: ${err}`);
+        return jobFailed("apple health import reconcile failed", err);
       }
+      return jobDone();
     },
   );
 }
@@ -230,7 +234,7 @@ async function writeProgress(
  */
 export async function handleAppleHealthImport(
   job: Job<AppleHealthImportPayload>,
-): Promise<void> {
+): Promise<JobOutcome> {
   const { userId, uploadPath, uploadBytes, triggeredByAdminId } = job.data;
   const prisma = getWorkerPrisma();
 
@@ -280,7 +284,7 @@ export async function handleAppleHealthImport(
         ` — row is already terminal (${importJob.status}); the staged upload` +
         " was consumed by the first run and a re-run could only mask its outcome",
     );
-    return;
+    return jobDone({ skipped: "already_terminal" });
   }
 
   // Extracted-XML path, hoisted so the failure path can clean it up —
@@ -364,6 +368,12 @@ export async function handleAppleHealthImport(
     // the end, scoped to the user's full measurement span, so
     // post-import reads of the analytics + comprehensive surfaces
     // hit the warm rollup table on first paint.
+    //
+    // The miss rides out as a fact rather than as a failed job: the
+    // import is committed and its mirror row already reads `done`, so
+    // failing here would report a completed import as failed and
+    // contradict the row the status poll surfaces to the operator.
+    let rollupFailed = false;
     try {
       const span = await prisma.measurement.aggregate({
         where: { userId },
@@ -382,6 +392,7 @@ export async function handleAppleHealthImport(
     } catch (rollupErr) {
       // Rollup failure is non-fatal — the next read falls through to
       // live aggregation. Log but don't poison the import.
+      rollupFailed = true;
       console.warn(
         `[apple-health-import] Rollup recompute failed for user ${userId}`,
         rollupErr,
@@ -392,6 +403,13 @@ export async function handleAppleHealthImport(
     // periodically swept on the host.
     safeUnlink(unzip.xmlPath);
     safeUnlink(uploadPath);
+
+    return jobDone({
+      records_read: result.totals.recordsRead,
+      rows_upserted: result.totals.rowsUpserted,
+      duration_ms: result.totals.durationMs,
+      rollup_failed: rollupFailed,
+    });
   } catch (err) {
     // v1.28.33 (issue #486) — a missing staging file is an operational
     // condition, not a parse failure: `/tmp` is wiped on a container

src/lib/jobs/data-arrival.ts

@@ -30,6 +30,7 @@
 import type { Job } from "pg-boss";
 
 import { invalidateUserDashboardSnapshot } from "@/lib/cache/invalidate";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { annotate } from "@/lib/logging/context";
 import { withBackgroundEvent } from "@/lib/logging/background";
 
@@ -248,14 +249,17 @@ export async function runDataArrival(
 
 export async function handleDataArrival(
   jobs: Job<DataArrival>[],
-): Promise<void> {
-  await withBackgroundEvent("job.data_arrival", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.data_arrival", async (evt) => {
+    let processed = 0;
+    let skipped = 0;
     for (const job of jobs) {
       const arrival = job.data;
       try {
         const outcome = await runDataArrival(getWorkerPrisma(), arrival);
 
         if (outcome.status === "skipped") {
+          skipped++;
           annotate({
             action: { name: `arrival.${arrival.kind}.skipped` },
             meta: { reason: outcome.reason, source: arrival.source },
@@ -264,6 +268,7 @@ export async function handleDataArrival(
           continue;
         }
 
+        processed++;
         annotate({
           action: { name: `arrival.${arrival.kind}.processed` },
           meta: {
@@ -285,5 +290,9 @@ export async function handleDataArrival(
         throw err;
       }
     }
+    // Both statuses are successes — a refused arrival is a decision the spine
+    // made, not work it failed to do. The counts are what make the split
+    // between them readable at all.
+    return jobDone({ total: jobs.length, processed, skipped });
   });
 }

src/lib/jobs/document-purge.ts

@@ -11,6 +11,7 @@
  */
 import { type Job } from "pg-boss";
 import { withBackgroundEvent } from "@/lib/logging/background";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { getWorkerPrisma } from "@/lib/jobs/reminder/shared";
 import type { PrismaClient } from "@/generated/prisma/client";
 
@@ -43,16 +44,22 @@ export async function purgeExpiredDocumentTombstones(
   return count;
 }
 
-export async function handleDocumentPurge(jobs: Job<DocumentPurgePayload>[]) {
+export async function handleDocumentPurge(
+  jobs: Job<DocumentPurgePayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.document_tombstone_purge", async (evt) => {
+  return withBackgroundEvent("job.document_tombstone_purge", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const purged = await purgeExpiredDocumentTombstones(prisma);
       evt.setAction({ name: "documents.vault.purge" });
       evt.addMeta("document_purge_deleted", purged);
+      // A night with no expired tombstone purges nothing and is still a run
+      // that did its work — the zero is the fact, not an absence of one.
+      return jobDone({ document_purge_deleted: purged });
     } catch (err) {
       evt.addWarning(`document-tombstone-purge failed: ${err}`);
+      return jobFailed("document tombstone purge failed", err);
     }
   });
 }

src/lib/jobs/encryption-key-rotate.ts

@@ -24,6 +24,7 @@
  */
 import { type Job } from "pg-boss";
 import { withBackgroundEvent } from "@/lib/logging/background";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { auditLog } from "@/lib/auth/audit";
 import {
   rotateCorpus,
@@ -60,8 +61,8 @@ export async function runEncryptionKeyRotation(): Promise<{
 
 export async function handleEncryptionKeyRotate(
   jobs: Job<EncryptionKeyRotatePayload>[],
-) {
-  await withBackgroundEvent("job.encryption_key_rotate", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.encryption_key_rotate", async (evt) => {
     const requestedBy = jobs[0]?.data?.requestedByUserId ?? null;
     try {
       const result = await runEncryptionKeyRotation();
@@ -78,6 +79,15 @@ export async function handleEncryptionKeyRotate(
           errors: result.totalErrors,
         },
       });
+      // Per-row errors are the fail-closed skip of a row under a key the
+      // deployment no longer configures. The pass still ran, so they ride out
+      // as a count rather than failing (and retrying) the whole corpus.
+      return jobDone({
+        rotate_active_key_id: result.activeKeyId,
+        rotate_scanned: result.totalScanned,
+        rotate_rotated: result.totalRotated,
+        rotate_errors: result.totalErrors,
+      });
     } catch (err) {
       evt.addWarning(`encryption-key-rotate failed: ${err}`);
       await auditLog("admin.encryption.rotate.failed", {

src/lib/jobs/environment-fetch.ts

@@ -33,6 +33,7 @@ import {
   utcDayKey,
 } from "@/lib/environment/service";
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { workerLog } from "./reminder/shared";
 
 export const ENVIRONMENT_FETCH_QUEUE = "environment-fetch";
@@ -152,14 +153,14 @@ export async function enqueueEnvironmentFetchDiscovery(
 export async function handleEnvironmentFetch(
   boss: PgBoss,
   payload: EnvironmentFetchPayload,
-): Promise<void> {
+): Promise<JobOutcome> {
   if (!payload.userId) {
     const { enqueued, skipped } = await enqueueEnvironmentFetchDiscovery(boss);
     workerLog(
       "info",
       `[environment-fetch] discovery enqueued=${enqueued} skipped=${skipped}`,
     );
-    return;
+    return jobDone({ users_enqueued: enqueued, users_skipped: skipped });
   }
 
   const userId = payload.userId;
@@ -168,7 +169,7 @@ export async function handleEnvironmentFetch(
       "info",
       `[environment-fetch] user=${userId} module disabled — skipping`,
     );
-    return;
+    return jobDone({ skipped: "environment module disabled" });
   }
 
   const { startDate, endDate } = resolveRange(payload);
@@ -179,7 +180,11 @@ export async function handleEnvironmentFetch(
       "error",
       `[environment-fetch] user=${userId} range ${startDate}..${endDate} (${span}d) exceeds cap — skipping`,
     );
-    return;
+    // A refused range is not an opt-out: the job was asked to fill days and
+    // filled none, so it says so instead of completing quietly on a stderr line.
+    return jobFailed("environment backfill range exceeds cap", undefined, {
+      span_days: span,
+    });
   }
 
   try {
@@ -192,6 +197,11 @@ export async function handleEnvironmentFetch(
       "info",
       `[environment-fetch] user=${userId} ${startDate}..${endDate} stored=${result.stored} skipped=${result.skipped} fetches=${result.fetches}`,
     );
+    return jobDone({
+      days_stored: result.stored,
+      days_skipped: result.skipped,
+      fetches: result.fetches,
+    });
   } catch (err) {
     recordError();
     workerLog("error", `[environment-fetch] user=${userId} failed`, err);

src/lib/jobs/job-outcome.ts

@@ -17,6 +17,13 @@
  * signal a failure) into commission (claiming a success that did not
  * happen), which is a smaller and much more reviewable class of defect.
  *
+ * A fan-out pass is judged on the pass, not on its legs. A nightly sweep
+ * over every connected account is `ok: true` when the sweep ran, even if one
+ * account's grant was revoked: that belongs on the account's own integration
+ * ledger, and failing the queue would retry the whole cohort over one user's
+ * expired token. The per-leg failures ride out as a count in `did`, which is
+ * what makes them visible at all — today they are visible nowhere.
+ *
  * `did` is the small pinned-shape bag of facts about the run — counts,
  * identifiers, booleans. It rides into the failure report and, once the
  * failure ledger lands, into the operator surface. Keep it to scalars that

src/lib/jobs/medication-intake-import.ts

@@ -3,6 +3,7 @@ import { Prisma, type PrismaClient } from "@/generated/prisma/client";
 
 import { invalidateUserMedications } from "@/lib/cache/invalidate";
 import { prisma, toJson } from "@/lib/db";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { consumeImportedIntakesBatch } from "@/lib/medications/inventory/consumption";
 import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
 import {
@@ -711,9 +712,19 @@ async function recordWorkerFailure(
 
 export async function handleMedicationIntakeImport(
   job: Job<MedicationIntakeImportQueuePayload>,
-): Promise<void> {
+): Promise<JobOutcome> {
   try {
-    await processMedicationIntakeImportJob(job.data.jobId);
+    const result = await processMedicationIntakeImportJob(job.data.jobId);
+    // A null result is a delivery that found the row already terminal or gone.
+    // There was nothing left to import, which is an absence of work rather
+    // than a failure to do it.
+    return result === null
+      ? jobDone({ finalized: false })
+      : jobDone({
+          finalized: true,
+          imported: result.imported,
+          skipped: result.skipped,
+        });
   } catch (error) {
     await recordWorkerFailure(job, error);
     throw error;

src/lib/jobs/reaction-line.ts

@@ -29,6 +29,7 @@ import type { Job } from "pg-boss";
 
 import { prisma } from "@/lib/db";
 import { randomUUID } from "node:crypto";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { annotate } from "@/lib/logging/context";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { resolveProviderChain } from "@/lib/ai/provider";
@@ -677,13 +678,16 @@ export async function runReactionLine(
 
 export async function handleReactionLine(
   jobs: Job<ReactionLineJob>[],
-): Promise<void> {
-  await withBackgroundEvent("job.reaction_line", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.reaction_line", async (evt) => {
+    let generated = 0;
+    let skipped = 0;
     for (const job of jobs) {
       try {
         const outcome = await runReactionLine(job.data);
 
         if (outcome.status === "skipped") {
+          skipped++;
           annotate({
             action: { name: "arrival.reaction_line.skipped" },
             meta: { reason: outcome.reason, kind: job.data.kind },
@@ -692,6 +696,7 @@ export async function handleReactionLine(
           continue;
         }
 
+        generated++;
         annotate({
           action: { name: "arrival.reaction_line.generated" },
           meta: { kind: job.data.kind, local_date: job.data.localDate },
@@ -705,5 +710,9 @@ export async function handleReactionLine(
         throw err;
       }
     }
+    // A skip is a refusal the surface is designed around, not a miss: the
+    // deterministic lead already stands without a line. The pass did its work
+    // whenever it ran to the end.
+    return jobDone({ total: jobs.length, generated, skipped });
   });
 }

src/lib/jobs/reminder/backup-handlers.ts

@@ -8,8 +8,12 @@ import { type Job } from "pg-boss";
 import { recordError } from "@/lib/jobs/worker-status";
 import { buildFullBackupPayload } from "@/lib/export/full-backup-payload";
 import { encrypt } from "@/lib/crypto";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
-import { runOffhostBackup } from "@/lib/jobs/offhost-backup";
+import {
+  OffhostBackupNotConfiguredError,
+  runOffhostBackup,
+} from "@/lib/jobs/offhost-backup";
 import { getWorkerPrisma } from "./shared";
 
 export interface DataBackupPayload {
@@ -20,9 +24,11 @@ export interface OffhostBackupPayload {
   triggeredAt: string;
 }
 
-export async function handleOffhostBackup(jobs: Job<OffhostBackupPayload>[]) {
+export async function handleOffhostBackup(
+  jobs: Job<OffhostBackupPayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.offhost_backup", async (evt) => {
+  return withBackgroundEvent("job.offhost_backup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const report = await runOffhostBackup(p);
@@ -39,16 +45,34 @@ export async function handleOffhostBackup(jobs: Job<OffhostBackupPayload>[]) {
           JSON.stringify(report.failures.slice(0, 10)),
         );
       }
+      // Per-user upload failures ride out as `offhost_backup_failed`: the run
+      // itself uploaded what it could, and failing the queue over one user's
+      // object would re-upload the whole cohort on every retry.
+      return jobDone({
+        offhost_backup_uploaded: report.uploaded,
+        offhost_backup_failed: report.failed,
+        offhost_backup_total_users: report.totalUsers,
+        offhost_backup_endpoint: report.config.endpoint,
+        offhost_backup_bucket: report.config.bucket,
+      });
     } catch (err) {
-      // Not configured ⇒ skip silently with a warning, not an error.
+      // Not configured ⇒ skip silently with a warning, not an error: most
+      // self-hosts never set the S3 credentials, and a nightly failed job on
+      // every one of them buries the runs that genuinely could not upload.
       evt.addWarning(`offhost-backup skipped/failed: ${err}`);
+      if (err instanceof OffhostBackupNotConfiguredError) {
+        return jobDone({ offhost_backup_configured: false });
+      }
+      return jobFailed("offhost backup failed", err);
     }
   });
 }
 
-export async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
+export async function handleDataBackup(
+  jobs: Job<DataBackupPayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.data_backup", async (evt) => {
+  return withBackgroundEvent("job.data_backup", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const users = await prisma.user.findMany({
@@ -56,6 +80,7 @@ export async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
       });
 
       let backed = 0;
+      let usersFailed = 0;
       for (const user of users) {
         try {
           const { payload } = await buildFullBackupPayload(prisma, user.id, {
@@ -82,6 +107,10 @@ export async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
           });
           backed++;
         } catch (err) {
+          // One user's payload failing is that user's problem, not the pass's:
+          // it rides out as a count so the weekly run is not retried for the
+          // whole cohort.
+          usersFailed++;
           evt.addWarning(`Failed for user ${user.id}: ${err}`);
         }
       }
@@ -90,6 +119,11 @@ export async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
         task_name: "job.data_backup",
         result: { backed, total: users.length },
       });
+      return jobDone({
+        backed,
+        total: users.length,
+        users_failed: usersFailed,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();

src/lib/jobs/reminder/cleanup-handlers.ts

@@ -2,10 +2,27 @@
  * Retention and maintenance cleanups: rate-limit rows, idempotency keys, audit logs, OAuth states (Withings / WHOOP), mood-reminder events, push attempts, and measurement tombstones.
  *
  * Extracted from reminder-worker.ts, which owns the queue names, cron
- * schedules, and boss.work registrations.
+ * schedules, and the queue bindings.
+ *
+ * Every pass in here used to catch its own failure, write a warning nobody
+ * reads, and return — so a retention purge that had not run for weeks was
+ * indistinguishable from one with nothing to delete. They now return
+ * `jobFailed`, which fails the pg-boss job and puts the queue on the operator's
+ * failing-jobs surface.
+ *
+ * That change has a price, and the registrars pay it: these are bulk DELETEs
+ * over a retention horizon, and their failure mode is deterministic. A
+ * statement that timed out against a large trailing edge times out again
+ * immediately. Under pg-boss's default policy (retryLimit 2, no delay, no
+ * backoff) each converted queue would run the same doomed statement three
+ * times in a row every night. So every cron-driven cleanup registered from
+ * this module is scheduled with `retryLimit: 0`: the failure is recorded once
+ * and the next night's tick is the retry, which is the right backoff for
+ * retention work that has a whole day of slack.
  */
 import { type Job } from "pg-boss";
 import { withBackgroundEvent } from "@/lib/logging/background";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import {
   cleanupExpiredWhoopConnectTickets,
   cleanupExpiredWhoopOAuthStates,
@@ -78,9 +95,9 @@ export interface MoodReminderCleanupPayload {
 
 export async function handleMoodReminderCleanup(
   jobs: Job<MoodReminderCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.mood_reminder_cleanup", async (evt) => {
+  return withBackgroundEvent("job.mood_reminder_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const cutoff = new Date();
@@ -90,8 +107,10 @@ export async function handleMoodReminderCleanup(
         where: { date: { lt: cutoffIso } },
       });
       evt.addMeta("mood_reminder_cleanup_deleted", deleted.count);
+      return jobDone({ deleted: deleted.count });
     } catch (err) {
       evt.addWarning(`mood-reminder-cleanup failed: ${err}`);
+      return jobFailed("mood-reminder cleanup failed", err);
     }
   });
 }
@@ -118,9 +137,9 @@ export interface PushAttemptCleanupPayload {
 
 export async function handlePushAttemptCleanup(
   jobs: Job<PushAttemptCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.push_attempt_cleanup", async (evt) => {
+  return withBackgroundEvent("job.push_attempt_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const cutoff = new Date();
@@ -129,8 +148,10 @@ export async function handlePushAttemptCleanup(
         where: { createdAt: { lt: cutoff } },
       });
       evt.addMeta("push_attempt_cleanup_deleted", deleted.count);
+      return jobDone({ deleted: deleted.count });
     } catch (err) {
       evt.addWarning(`push-attempt-cleanup failed: ${err}`);
+      return jobFailed("push-attempt cleanup failed", err);
     }
   });
 }
@@ -152,9 +173,9 @@ export interface ArrivalReactionCleanupPayload {
 
 export async function handleArrivalReactionCleanup(
   jobs: Job<ArrivalReactionCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.arrival_reaction_cleanup", async (evt) => {
+  return withBackgroundEvent("job.arrival_reaction_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const cutoff = new Date();
@@ -163,8 +184,10 @@ export async function handleArrivalReactionCleanup(
         where: { createdAt: { lt: cutoff } },
       });
       evt.addMeta("arrival_reaction_cleanup_deleted", deleted.count);
+      return jobDone({ deleted: deleted.count });
     } catch (err) {
       evt.addWarning(`arrival-reaction-cleanup failed: ${err}`);
+      return jobFailed("arrival-reaction cleanup failed", err);
     }
   });
 }
@@ -175,9 +198,9 @@ export interface MeasurementTombstoneCleanupPayload {
 
 export async function handleMeasurementTombstoneCleanup(
   jobs: Job<MeasurementTombstoneCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent(
+  return withBackgroundEvent(
     "job.measurement_tombstone_cleanup",
     async (evt) => {
       const p = getWorkerPrisma();
@@ -208,6 +231,12 @@ export async function handleMeasurementTombstoneCleanup(
             "tombstone-cleanup stopped at the batch cap; a backlog remains for the next run",
           );
         }
+        return jobDone({
+          measurements_pruned: measurements.deleted,
+          mood_pruned: mood.deleted,
+          intakes_pruned: intakes.deleted,
+          drained,
+        });
       } catch (err) {
         // Rethrow. A retention purge that fails silently every night looks
         // exactly like one that has nothing to do; pg-boss recording a failed
@@ -221,32 +250,36 @@ export async function handleMeasurementTombstoneCleanup(
 
 export async function handleRateLimitCleanup(
   jobs: Job<RateLimitCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.rate_limit_cleanup", async (evt) => {
+  return withBackgroundEvent("job.rate_limit_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const result = await p.$executeRaw`
         DELETE FROM rate_limits WHERE reset_at < NOW()
       `;
       evt.addMeta("rate_limit_cleanup_deleted", result);
+      return jobDone({ deleted: result });
     } catch (err) {
       evt.addWarning(`rate-limit-cleanup failed: ${err}`);
+      return jobFailed("rate-limit cleanup failed", err);
     }
   });
 }
 
 export async function handleIdempotencyCleanup(
   jobs: Job<IdempotencyCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.idempotency_cleanup", async (evt) => {
+  return withBackgroundEvent("job.idempotency_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const deleted = await cleanupExpiredIdempotencyKeys(p);
       evt.addMeta("idempotency_cleanup_deleted", deleted);
+      return jobDone({ deleted });
     } catch (err) {
       evt.addWarning(`idempotency-cleanup failed: ${err}`);
+      return jobFailed("idempotency-key cleanup failed", err);
     }
   });
 }
@@ -266,26 +299,28 @@ export async function handleIdempotencyCleanup(
  */
 export async function handleStepUpElevationCleanup(
   jobs: Job<StepUpElevationCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.step_up_elevation_cleanup", async (evt) => {
+  return withBackgroundEvent("job.step_up_elevation_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const { count } = await p.stepUpElevation.deleteMany({
         where: { expiresAt: { lt: new Date() } },
       });
       evt.addMeta("step_up_elevation_cleanup_deleted", count);
+      return jobDone({ deleted: count });
     } catch (err) {
       evt.addWarning(`step-up-elevation-cleanup failed: ${err}`);
+      return jobFailed("step-up elevation cleanup failed", err);
     }
   });
 }
 
 export async function handleAuditLogCleanup(
   jobs: Job<AuditLogCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.audit_log_cleanup", async (evt) => {
+  return withBackgroundEvent("job.audit_log_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const outcome = await cleanupOldAuditLogs(p);
@@ -296,6 +331,7 @@ export async function handleAuditLogCleanup(
           "audit-log-cleanup stopped at the batch cap; a backlog remains for the next run",
         );
       }
+      return jobDone({ deleted: outcome.deleted, drained: outcome.drained });
     } catch (err) {
       // Rethrow — see the tombstone handler. Storage-limitation retention that
       // fails quietly is retention that is not happening.
@@ -307,15 +343,17 @@ export async function handleAuditLogCleanup(
 
 export async function handleCoachMessageCleanup(
   jobs: Job<CoachMessageCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.coach_message_cleanup", async (evt) => {
+  return withBackgroundEvent("job.coach_message_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const deleted = await cleanupOldCoachMessages(p);
       evt.addMeta("coach_message_cleanup_deleted", deleted);
+      return jobDone({ deleted });
     } catch (err) {
       evt.addWarning(`coach-message-cleanup failed: ${err}`);
+      return jobFailed("coach-message cleanup failed", err);
     }
   });
 }
@@ -326,9 +364,9 @@ export interface McpTokenCleanupPayload {
 
 export async function handleMcpTokenCleanup(
   jobs: Job<McpTokenCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.mcp_token_cleanup", async (evt) => {
+  return withBackgroundEvent("job.mcp_token_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const result = await cleanupExpiredMcpTokens(p);
@@ -340,27 +378,40 @@ export async function handleMcpTokenCleanup(
         "mcp_token_cleanup_connections_deleted",
         result.connectionsDeleted,
       );
+      return jobDone({
+        access_tokens_deleted: result.accessTokensDeleted,
+        connections_deleted: result.connectionsDeleted,
+      });
     } catch (err) {
       evt.addWarning(`mcp-token-cleanup failed: ${err}`);
+      return jobFailed("mcp-token cleanup failed", err);
     }
   });
 }
 
 export async function handleWithingsOAuthStateCleanup(
   jobs: Job<WithingsOAuthStateCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.withings_oauth_state_cleanup", async (evt) => {
-    const p = getWorkerPrisma();
-    try {
-      const deleted = await cleanupExpiredWithingsOAuthStates(p);
-      evt.addMeta("withings_oauth_state_cleanup_deleted", deleted);
-    } catch (err) {
-      // The OAuth flow tolerates a stale row sticking around for an
-      // extra day — log + carry on so the boss queue doesn't retry-loop.
-      evt.addWarning(`withings-oauth-state-cleanup failed: ${err}`);
-    }
-  });
+  return withBackgroundEvent(
+    "job.withings_oauth_state_cleanup",
+    async (evt) => {
+      const p = getWorkerPrisma();
+      try {
+        const deleted = await cleanupExpiredWithingsOAuthStates(p);
+        evt.addMeta("withings_oauth_state_cleanup_deleted", deleted);
+        return jobDone({ deleted });
+      } catch (err) {
+        // The OAuth flow tolerates a stale row for an extra day, which used
+        // to be the argument for swallowing this so the queue would not
+        // retry-loop. Tolerable is not the same as invisible: the failure is
+        // now reported, and the retry-loop worry is answered where it belongs,
+        // by the `retryLimit: 0` this queue is scheduled with.
+        evt.addWarning(`withings-oauth-state-cleanup failed: ${err}`);
+        return jobFailed("withings oauth-state cleanup failed", err);
+      }
+    },
+  );
 }
 
 export interface OidcNativeHandoffCleanupPayload {
@@ -369,17 +420,21 @@ export interface OidcNativeHandoffCleanupPayload {
 
 export async function handleOidcNativeHandoffCleanup(
   jobs: Job<OidcNativeHandoffCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.oidc_native_handoff_cleanup", async (evt) => {
+  return withBackgroundEvent("job.oidc_native_handoff_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const deleted = await cleanupExpiredOidcNativeHandoffs(p);
       evt.addMeta("oidc_native_handoff_cleanup_deleted", deleted);
+      return jobDone({ deleted });
     } catch (err) {
-      // The handoff flow tolerates a stale row for an extra day (expiry is
-      // enforced at read) — log + carry on so the boss queue doesn't retry-loop.
+      // Expiry is enforced at read, so a stale row for an extra day is
+      // harmless — but see the withings twin above: harmless is not a
+      // reason to hide the failure. `retryLimit: 0` on the schedule is what
+      // keeps a doomed statement from running three times a night.
       evt.addWarning(`oidc-native-handoff-cleanup failed: ${err}`);
+      return jobFailed("oidc native-handoff cleanup failed", err);
     }
   });
 }
@@ -390,17 +445,22 @@ export interface WhoopOAuthStateCleanupPayload {
 
 export async function handleWhoopOAuthStateCleanup(
   jobs: Job<WhoopOAuthStateCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.whoop_oauth_state_cleanup", async (evt) => {
+  return withBackgroundEvent("job.whoop_oauth_state_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const deleted = await cleanupExpiredWhoopOAuthStates(p);
       evt.addMeta("whoop_oauth_state_cleanup_deleted", deleted);
       const ticketsDeleted = await cleanupExpiredWhoopConnectTickets(p);
       evt.addMeta("whoop_connect_ticket_cleanup_deleted", ticketsDeleted);
+      return jobDone({
+        states_deleted: deleted,
+        connect_tickets_deleted: ticketsDeleted,
+      });
     } catch (err) {
       evt.addWarning(`whoop-oauth-state-cleanup failed: ${err}`);
+      return jobFailed("whoop oauth-state cleanup failed", err);
     }
   });
 }

src/lib/jobs/reminder/fitbit-sync.ts

@@ -7,6 +7,7 @@
 import { type Job } from "pg-boss";
 import { fireAndForget } from "@/lib/logging/fire-and-forget";
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { runFitbitPollCohort } from "@/lib/fitbit/sync";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
@@ -23,8 +24,10 @@ export interface FitbitSyncPayload {
   userId?: string;
 }
 
-export async function handleFitbitSync(jobs: Job<FitbitSyncPayload>[]) {
-  await withBackgroundEvent("job.fitbit_sync", async (evt) => {
+export async function handleFitbitSync(
+  jobs: Job<FitbitSyncPayload>[],
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.fitbit_sync", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const targets: Array<{ userId: string }> = [];
@@ -37,16 +40,26 @@ export async function handleFitbitSync(jobs: Job<FitbitSyncPayload>[]) {
         });
         targets.push(...connections);
       }
-      if (targets.length === 0) return;
+      if (targets.length === 0) {
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          measurements_imported: 0,
+        });
+      }
 
       // Fan the cohort out with bounded concurrency + per-user error isolation:
       // one slow Google response can't stall the whole cohort, and a single
       // user's failure is warned without aborting the pass.
+      let usersFailed = 0;
       const { usersSynced, measurementsImported } = await runFitbitPollCohort(
         targets.map((t) => t.userId),
         {
-          onUserError: (userId, err) =>
-            evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`),
+          onUserError: (userId, err) => {
+            usersFailed++;
+            evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`);
+          },
           // v1.18.1 — a fresh reading landed; resolve the user's Vorsorge
           // reminders eventfully. Fire-and-forget; the cron is the net.
           onUserSynced: (userId, imported) => {
@@ -67,6 +80,14 @@ export async function handleFitbitSync(jobs: Job<FitbitSyncPayload>[]) {
           measurements_imported: measurementsImported,
         },
       });
+      // The pass is the unit of success: a revoked grant belongs on that user's
+      // integration ledger, not in a queue failure that retries every account.
+      return jobDone({
+        total: targets.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        measurements_imported: measurementsImported,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -81,15 +102,17 @@ export interface FitbitOAuthStateCleanupPayload {
 
 export async function handleFitbitOAuthStateCleanup(
   jobs: Job<FitbitOAuthStateCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.fitbit_oauth_state_cleanup", async (evt) => {
+  return withBackgroundEvent("job.fitbit_oauth_state_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const deleted = await cleanupExpiredFitbitOAuthStates(p);
       evt.addMeta("fitbit_oauth_state_cleanup_deleted", deleted);
+      return jobDone({ deleted });
     } catch (err) {
       evt.addWarning(`fitbit-oauth-state-cleanup failed: ${err}`);
+      return jobFailed("fitbit oauth state cleanup failed", err);
     }
   });
 }

src/lib/jobs/reminder/google-health-sync.ts

@@ -7,6 +7,7 @@
 import { type Job } from "pg-boss";
 import { fireAndForget } from "@/lib/logging/fire-and-forget";
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { runGoogleHealthPollCohort } from "@/lib/google-health/sync";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
@@ -26,8 +27,8 @@ export interface GoogleHealthSyncPayload {
 
 export async function handleGoogleHealthSync(
   jobs: Job<GoogleHealthSyncPayload>[],
-) {
-  await withBackgroundEvent("job.google_health_sync", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.google_health_sync", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const targets: Array<{ userId: string }> = [];
@@ -40,19 +41,29 @@ export async function handleGoogleHealthSync(
         });
         targets.push(...connections);
       }
-      if (targets.length === 0) return;
+      if (targets.length === 0) {
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          measurements_imported: 0,
+        });
+      }
 
       // Fan the cohort out with bounded concurrency + per-user error isolation:
       // one slow Google response can't stall the whole cohort, and a single
       // user's failure is warned without aborting the pass.
+      let usersFailed = 0;
       const { usersSynced, measurementsImported } =
         await runGoogleHealthPollCohort(
           targets.map((t) => t.userId),
           {
-            onUserError: (userId, err) =>
+            onUserError: (userId, err) => {
+              usersFailed++;
               evt.addWarning(
                 `job.google_health_sync failed for user ${userId}: ${err}`,
-              ),
+              );
+            },
             // A fresh reading landed; resolve the user's Vorsorge reminders
             // eventfully. Fire-and-forget; the cron is the net.
             onUserSynced: (userId, imported) => {
@@ -73,6 +84,14 @@ export async function handleGoogleHealthSync(
           measurements_imported: measurementsImported,
         },
       });
+      // The pass is the unit of success: a revoked grant belongs on that user's
+      // integration ledger, not in a queue failure that retries every account.
+      return jobDone({
+        total: targets.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        measurements_imported: measurementsImported,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -87,17 +106,19 @@ export interface GoogleHealthOAuthStateCleanupPayload {
 
 export async function handleGoogleHealthOAuthStateCleanup(
   jobs: Job<GoogleHealthOAuthStateCleanupPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent(
+  return withBackgroundEvent(
     "job.google_health_oauth_state_cleanup",
     async (evt) => {
       const p = getWorkerPrisma();
       try {
         const deleted = await cleanupExpiredGoogleHealthOAuthStates(p);
         evt.addMeta("google_health_oauth_state_cleanup_deleted", deleted);
+        return jobDone({ deleted });
       } catch (err) {
         evt.addWarning(`google-health-oauth-state-cleanup failed: ${err}`);
+        return jobFailed("google health oauth state cleanup failed", err);
       }
     },
   );

src/lib/jobs/reminder/insights-handlers.ts

@@ -19,6 +19,7 @@ import {
   type InsightStatusGeneratePayload,
   runInsightStatusGenerate,
 } from "@/lib/jobs/insight-status-generate";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status";
 import { generateWeightStatusForUser } from "@/lib/insights/weight-status";
@@ -69,6 +70,10 @@ export interface MedicationComplianceStatusPayload {
  * `status-cron-candidates.ts` for the full division of nightly labour).
  * The generators normalise `locale` themselves (de stays de, everything
  * else gets English prose).
+ *
+ * The pass is what succeeds or fails. One user whose generation threw is that
+ * user's cold card, reported as the `failed` count; failing the job would
+ * re-run the whole cohort over it.
  */
 export async function runStatusCronGenerate(
   taskName: string,
@@ -76,13 +81,15 @@ export async function runStatusCronGenerate(
     userId: string,
     options: { locale: string | null; force: boolean },
   ) => Promise<unknown>,
-): Promise<void> {
-  await withBackgroundEvent(taskName, async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent(taskName, async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const users = await findStatusCronCandidates(prisma);
 
-      if (users.length === 0) return;
+      if (users.length === 0) {
+        return jobDone({ total: 0, generated: 0, failed: 0 });
+      }
 
       let generated = 0;
       let failed = 0;
@@ -103,6 +110,7 @@ export async function runStatusCronGenerate(
         task_name: taskName,
         result: { generated, failed, total: users.length },
       });
+      return jobDone({ total: users.length, generated, failed });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -129,13 +137,15 @@ export async function runStatusCronGenerate(
  * inside the batch itself). Net effect: the nightly ladder pays one call per
  * user instead of seven, with no queue/registry churn.
  */
-async function runStatusBatchCron(taskName: string): Promise<void> {
-  await withBackgroundEvent(taskName, async (evt) => {
+async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
+  return withBackgroundEvent(taskName, async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       recordInsightsRun();
       const users = await findStatusCronCandidates(prisma);
-      if (users.length === 0) return;
+      if (users.length === 0) {
+        return jobDone({ total: 0, generated: 0, served: 0, failed: 0 });
+      }
 
       let generated = 0;
       let served = 0;
@@ -167,6 +177,10 @@ async function runStatusBatchCron(taskName: string): Promise<void> {
         task_name: taskName,
         result: { generated, served, failed, total: users.length },
       });
+      // A user whose batch call threw already fell back to the single-card
+      // path inside the batch, and the six later per-metric crons cover what
+      // is still cold. The batch pass ran, so it reports the count.
+      return jobDone({ total: users.length, generated, served, failed });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -176,14 +190,16 @@ async function runStatusBatchCron(taskName: string): Promise<void> {
   });
 }
 
-export function handleGeneralStatusGenerate(jobs: Job<GeneralStatusPayload>[]) {
+export function handleGeneralStatusGenerate(
+  jobs: Job<GeneralStatusPayload>[],
+): Promise<JobOutcome> {
   void jobs;
   return runStatusBatchCron("job.insights.batch");
 }
 
 export function handleBloodPressureStatusGenerate(
   jobs: Job<BloodPressureStatusPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
   return runStatusCronGenerate(
     "job.insights.blood_pressure",
@@ -191,7 +207,9 @@ export function handleBloodPressureStatusGenerate(
   );
 }
 
-export function handleWeightStatusGenerate(jobs: Job<WeightStatusPayload>[]) {
+export function handleWeightStatusGenerate(
+  jobs: Job<WeightStatusPayload>[],
+): Promise<JobOutcome> {
   void jobs;
   return runStatusCronGenerate(
     "job.insights.weight",
@@ -199,7 +217,9 @@ export function handleWeightStatusGenerate(jobs: Job<WeightStatusPayload>[]) {
   );
 }
 
-export function handlePulseStatusGenerate(jobs: Job<PulseStatusPayload>[]) {
+export function handlePulseStatusGenerate(
+  jobs: Job<PulseStatusPayload>[],
+): Promise<JobOutcome> {
   void jobs;
   return runStatusCronGenerate(
     "job.insights.pulse",
@@ -207,19 +227,23 @@ export function handlePulseStatusGenerate(jobs: Job<PulseStatusPayload>[]) {
   );
 }
 
-export function handleBmiStatusGenerate(jobs: Job<BmiStatusPayload>[]) {
+export function handleBmiStatusGenerate(
+  jobs: Job<BmiStatusPayload>[],
+): Promise<JobOutcome> {
   void jobs;
   return runStatusCronGenerate("job.insights.bmi", generateBmiStatusForUser);
 }
 
-export function handleMoodStatusGenerate(jobs: Job<MoodStatusPayload>[]) {
+export function handleMoodStatusGenerate(
+  jobs: Job<MoodStatusPayload>[],
+): Promise<JobOutcome> {
   void jobs;
   return runStatusCronGenerate("job.insights.mood", generateMoodStatusForUser);
 }
 
 export function handleMedicationComplianceStatusGenerate(
   jobs: Job<MedicationComplianceStatusPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
   return runStatusCronGenerate(
     "job.insights.medication_compliance",
@@ -229,8 +253,8 @@ export function handleMedicationComplianceStatusGenerate(
 
 export async function handleInsightPregenerateJob(
   jobs: Job<InsightPregeneratePayload>[],
-) {
-  await withBackgroundEvent("job.insight_pregenerate", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.insight_pregenerate", async (evt) => {
     // v1.8.7.1 — a forced single-user warm carries `{ userId, force }`;
     // the scheduled tick carries neither. Route each job individually so a
     // batch that mixes a cron tick with on-demand warms (it never does in
@@ -265,13 +289,30 @@ export async function handleInsightPregenerateJob(
       }
     }
 
-    if (scheduled.length === 0) return;
+    // A batch of forced warms only is a complete run: every warm either landed
+    // or threw out of the loop above.
+    if (scheduled.length === 0) {
+      return jobDone({ forced: forced.length, scheduled: 0 });
+    }
     try {
       const summary = await runInsightPregenerate(getWorkerPrisma());
       evt.setBackground({
         task_name: "job.insight_pregenerate",
         result: { ...summary },
       });
+      return jobDone({
+        forced: forced.length,
+        scheduled: scheduled.length,
+        total: summary.total,
+        generated: summary.generated,
+        cached: summary.cached,
+        unchanged: summary.unchanged,
+        skipped: summary.skipped,
+        failed: summary.failed,
+        budget_blocked: summary.budgetBlocked,
+        assessments_warmed: summary.assessmentsWarmed,
+        metric_assessments_warmed: summary.metricAssessmentsWarmed,
+      });
     } catch (err) {
       evt.addWarning(
         `insight-pregenerate failed: ${
@@ -296,12 +337,18 @@ export async function handleInsightPregenerateJob(
  */
 export async function handleInsightStatusGenerate(
   jobs: Job<InsightStatusGeneratePayload>[],
-) {
-  await withBackgroundEvent("job.insight_status_generate", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.insight_status_generate", async (evt) => {
+    let generated = 0;
+    let malformed = 0;
     for (const job of jobs) {
-      if (!job.data?.userId || !job.data?.metric) continue;
+      if (!job.data?.userId || !job.data?.metric) {
+        malformed++;
+        continue;
+      }
       try {
         await runInsightStatusGenerate(job.data);
+        generated++;
         evt.addMeta(
           "status_generated",
           `${job.data.metric}:${job.data.locale}`,
@@ -321,5 +368,8 @@ export async function handleInsightStatusGenerate(
         throw err;
       }
     }
+    // Every job in the batch either generated a card or carried no metric to
+    // generate; a generation that failed left through the rethrow above.
+    return jobDone({ generated, malformed });
   });
 }

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

@@ -10,6 +10,7 @@ import {
   type IntakeAutoSkipPayload,
   runIntakeAutoSkipPass,
 } from "@/lib/jobs/intake-auto-skip";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { expireStaleInUseItems } from "@/lib/medications/inventory/service";
 import { getWorkerPrisma } from "./shared";
@@ -21,18 +22,21 @@ import { getWorkerPrisma } from "./shared";
  */
 export async function handleMedicationInventoryExpire(
   jobs: Job<MedicationInventoryExpirePayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.medication_inventory_expire", async (evt) => {
+  return withBackgroundEvent("job.medication_inventory_expire", async (evt) => {
     try {
       const count = await expireStaleInUseItems({ nowMs: Date.now() });
       evt.addMeta("inventory_expired_count", count);
+      // A pass that found no lapsed pen did its work; zero is a result.
+      return jobDone({ inventory_expired_count: count });
     } catch (err) {
       evt.addWarning(
         `medication-inventory-expire failed: ${
           err instanceof Error ? err.message : String(err)
         }`,
       );
+      return jobFailed("inventory expire pass failed", err);
     }
   });
 }
@@ -46,9 +50,11 @@ export async function handleMedicationInventoryExpire(
  * unit test can drive it with an in-memory fake Prisma; this wrapper
  * threads the worker's pg-boss + background-event plumbing.
  */
-export async function handleIntakeAutoSkip(jobs: Job<IntakeAutoSkipPayload>[]) {
+export async function handleIntakeAutoSkip(
+  jobs: Job<IntakeAutoSkipPayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.intake_auto_skip", async (evt) => {
+  return withBackgroundEvent("job.intake_auto_skip", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const result = await runIntakeAutoSkipPass(prisma, {
@@ -56,12 +62,17 @@ export async function handleIntakeAutoSkip(jobs: Job<IntakeAutoSkipPayload>[]) {
       });
       evt.addMeta("intake_auto_skip_count", result.skippedCount);
       evt.addMeta("intake_auto_skip_cutoff", result.cutoff.toISOString());
+      return jobDone({
+        intake_auto_skip_count: result.skippedCount,
+        intake_auto_skip_cutoff: result.cutoff.toISOString(),
+      });
     } catch (err) {
       evt.addWarning(
         `intake-auto-skip failed: ${
           err instanceof Error ? err.message : String(err)
         }`,
       );
+      return jobFailed("intake auto-skip pass failed", err);
     }
   });
 }

src/lib/jobs/reminder/medication-reminder-check.ts

@@ -8,6 +8,7 @@ import { type Job } from "pg-boss";
 import { recordError, recordReminderCheck } from "@/lib/jobs/worker-status";
 import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
 import { decrypt } from "@/lib/crypto";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { dispatchNotification } from "@/lib/notifications/dispatcher";
 import { medicationDoseTag } from "@/lib/notifications/dose-tag";
@@ -137,9 +138,11 @@ export async function cleanupScheduledTelegramDeletions(): Promise<void> {
  * Uses phase-based logic (GREEN/YELLOW/ORANGE/RED) to send one notification
  * per phase transition rather than every 15 minutes.
  */
-export async function handleReminderCheck(jobs: Job<ReminderCheckPayload>[]) {
+export async function handleReminderCheck(
+  jobs: Job<ReminderCheckPayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.medication_reminder", async (evt) => {
+  return withBackgroundEvent("job.medication_reminder", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       recordReminderCheck();
@@ -176,6 +179,12 @@ export async function handleReminderCheck(jobs: Job<ReminderCheckPayload>[]) {
         },
       });
 
+      // The per-slot facts the tick already emits as wide-event meta, counted
+      // so the run's outcome carries them as scalars.
+      let missedDosesMinted = 0;
+      let notificationsDispatched = 0;
+      let notificationsFailed = 0;
+
       for (const med of medications) {
         const userTz = med.user.timezone || "Europe/Berlin";
         const { start: todayStart, end: todayEnd } = getUserTodayBounds(
@@ -440,6 +449,7 @@ export async function handleReminderCheck(jobs: Job<ReminderCheckPayload>[]) {
                   },
                 });
 
+                missedDosesMinted++;
                 evt.addMeta("missed_dose", `${med.name}:${slotTime}`);
 
                 // v1.4.39 W-MED — refresh the compliance rollup so the
@@ -521,7 +531,12 @@ export async function handleReminderCheck(jobs: Job<ReminderCheckPayload>[]) {
                     url: `/medications/${med.id}`,
                   },
                 });
+                notificationsDispatched++;
               } catch (notifErr) {
+                // One user's channel failing must not fail the tick — the
+                // dedup anchor is already stamped, so the count is the only
+                // trace this slot's notice never left the host.
+                notificationsFailed++;
                 evt.addWarning(
                   `Notification dispatch failed for ${currentPhase} phase ${med.name}: ${notifErr}`,
                 );
@@ -530,6 +545,13 @@ export async function handleReminderCheck(jobs: Job<ReminderCheckPayload>[]) {
           }
         }
       }
+
+      return jobDone({
+        medications: medications.length,
+        missed_doses_minted: missedDosesMinted,
+        notifications_dispatched: notificationsDispatched,
+        notifications_failed: notificationsFailed,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();

src/lib/jobs/reminder/mood-cycle-checks.ts

@@ -6,6 +6,7 @@
  */
 import { type Job } from "pg-boss";
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { runMoodReminderTick } from "@/lib/jobs/mood-reminder";
 import { runCycleReminderTick } from "@/lib/jobs/cycle-reminder";
@@ -30,9 +31,9 @@ export interface MeasurementReminderPayload {
 
 export async function handleMoodReminderCheck(
   jobs: Job<MoodReminderPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.mood_reminder", async (evt) => {
+  return withBackgroundEvent("job.mood_reminder", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const summary = await runMoodReminderTick(prisma, new Date());
@@ -47,6 +48,16 @@ export async function handleMoodReminderCheck(
           skipped_outside_window: summary.skippedOutsideWindow,
         },
       });
+      // A tick that found nobody in their local reminder hour still ran the
+      // sweep, so it is a success with zero counts.
+      return jobDone({
+        candidates_scanned: summary.candidatesScanned,
+        in_window: summary.inWindow,
+        dispatched: summary.dispatched,
+        skipped_already_logged: summary.skippedAlreadyLogged,
+        skipped_already_dispatched: summary.skippedAlreadyDispatched,
+        skipped_outside_window: summary.skippedOutsideWindow,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -63,9 +74,9 @@ export async function handleMoodReminderCheck(
  */
 export async function handleCycleReminderCheck(
   jobs: Job<CycleReminderPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.cycle_reminder", async (evt) => {
+  return withBackgroundEvent("job.cycle_reminder", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const summary = await runCycleReminderTick(prisma, new Date());
@@ -82,6 +93,16 @@ export async function handleCycleReminderCheck(
           skipped_outside_window: summary.skippedOutsideWindow,
         },
       });
+      return jobDone({
+        candidates_scanned: summary.candidatesScanned,
+        in_window: summary.inWindow,
+        dispatched_period_soon: summary.dispatchedPeriodSoon,
+        dispatched_period_confirm: summary.dispatchedPeriodConfirm,
+        suppressed_client_managed: summary.suppressedClientManaged,
+        suppressed_discreet: summary.suppressedDiscreet,
+        skipped_already_notified: summary.skippedAlreadyNotified,
+        skipped_outside_window: summary.skippedOutsideWindow,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -97,9 +118,9 @@ export async function handleCycleReminderCheck(
  */
 export async function handleMeasurementReminderCheck(
   jobs: Job<MeasurementReminderPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.measurement_reminder", async (evt) => {
+  return withBackgroundEvent("job.measurement_reminder", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const summary = await runMeasurementReminderTick(prisma, new Date());
@@ -116,6 +137,19 @@ export async function handleMeasurementReminderCheck(
           failed: summary.failed,
         },
       });
+      // `failed` counts per-reminder dispatch failures the tick isolated. The
+      // sweep itself ran, so those ride out as a count rather than failing the
+      // job and retrying the whole cohort over one user's channel.
+      return jobDone({
+        candidates_scanned: summary.candidatesScanned,
+        in_window: summary.inWindow,
+        dispatched: summary.dispatched,
+        auto_resolved: summary.autoResolved,
+        skipped_not_due: summary.skippedNotDue,
+        skipped_outside_window: summary.skippedOutsideWindow,
+        skipped_no_channel: summary.skippedNoChannel,
+        failed: summary.failed,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -133,8 +167,8 @@ export async function handleMeasurementReminderCheck(
  */
 export async function handleReminderSatisfy(
   jobs: Job<ReminderSatisfyPayload>[],
-) {
-  await withBackgroundEvent("job.reminder_satisfy", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.reminder_satisfy", async (evt) => {
     const prisma = getWorkerPrisma();
     const now = new Date();
     // Distinct users in this batch — a multi-source ingest spike can enqueue
@@ -160,6 +194,13 @@ export async function handleReminderSatisfy(
           satisfied,
         },
       });
+      // A batch whose jobs carried no usable userId resolved nothing and is
+      // still a success with zero counts.
+      return jobDone({
+        users: userIds.length,
+        candidates_scanned: candidates,
+        satisfied,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();

src/lib/jobs/reminder/ops-handlers.ts

@@ -8,6 +8,7 @@ import { type Job } from "pg-boss";
 import { runGeoBackfill } from "@/lib/jobs/geo-backfill";
 import { runTlsPinMonitor } from "@/lib/jobs/tls-pin-monitor";
 import { type PrDetectionPayload } from "@/lib/jobs/pr-detection";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { runHostMetricTick } from "@/lib/jobs/host-metric-sampler";
 import { aggregateRecommendationFeedback } from "@/lib/jobs/feedback-aggregator";
@@ -34,40 +35,45 @@ export interface TlsPinMonitorPayload {
 
 export async function handleHostMetricSample(
   jobs: Job<HostMetricSamplePayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.host_metric_sample", async (evt) => {
+  return withBackgroundEvent("job.host_metric_sample", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const { pruned } = await runHostMetricTick(p);
       evt.addMeta("host_metric_pruned", pruned);
+      return jobDone({ host_metric_pruned: pruned });
     } catch (err) {
-      // The chart degrades gracefully when samples are missing — log
-      // and move on rather than poisoning the boss queue with retries.
+      // The chart degrades gracefully when samples are missing, so the
+      // warning stays — but a tick that took no sample did not do its work.
       evt.addWarning(`host-metric-sample failed: ${err}`);
+      return jobFailed("host metric sample failed", err);
     }
   });
 }
 
 export async function handleFeedbackAggregator(
   jobs: Job<FeedbackAggregatorPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.feedback_aggregator", async (evt) => {
+  return withBackgroundEvent("job.feedback_aggregator", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const summary = await aggregateRecommendationFeedback(p);
+      const totalRows = summary.buckets.reduce((acc, b) => acc + b.total, 0);
       evt.addMeta("feedback_buckets", summary.buckets.length);
-      evt.addMeta(
-        "feedback_total_rows",
-        summary.buckets.reduce((acc, b) => acc + b.total, 0),
-      );
+      evt.addMeta("feedback_total_rows", totalRows);
       evt.addMeta("feedback_window_days", summary.windowDays);
+      return jobDone({
+        feedback_buckets: summary.buckets.length,
+        feedback_total_rows: totalRows,
+        feedback_window_days: summary.windowDays,
+      });
     } catch (err) {
-      // The admin dashboard tolerates a stale summary — log and move
-      // on rather than poisoning the boss queue with retries that
-      // would block the next cleanup window.
+      // The admin dashboard tolerates a stale summary, so the warning stays —
+      // but a pass that aggregated nothing did not do its work.
       evt.addWarning(`feedback-aggregator failed: ${err}`);
+      return jobFailed("feedback aggregation failed", err);
     }
   });
 }
@@ -91,18 +97,21 @@ export async function handleFeedbackAggregator(
  */
 let geoBackfillRunning = false;
 
-export async function handleGeoBackfill(jobs: Job<GeoBackfillPayload>[]) {
+export async function handleGeoBackfill(
+  jobs: Job<GeoBackfillPayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.geo_backfill", async (evt) => {
+  return withBackgroundEvent("job.geo_backfill", async (evt) => {
     if (geoBackfillRunning) {
       // Earlier pass still in flight; skip this tick. Idempotent + the
       // next tick after the in-flight pass completes will pick up
-      // anything we miss here.
+      // anything we miss here. Deliberately ceding to the in-flight pass
+      // is not a failure, so the tick reports the skip and succeeds.
       evt.addWarning(
         "geo-backfill skipped — earlier pass still in flight inside the same worker process",
       );
       evt.addMeta("geo_backfill_skipped", true);
-      return;
+      return jobDone({ geo_backfill_skipped: true });
     }
     geoBackfillRunning = true;
     const p = getWorkerPrisma();
@@ -112,38 +121,60 @@ export async function handleGeoBackfill(jobs: Job<GeoBackfillPayload>[]) {
       evt.addMeta("geo_backfill_located", summary.located);
       evt.addMeta("geo_backfill_carrier_resolved", summary.carrierResolved);
       evt.addMeta("geo_backfill_still_unresolved", summary.stillUnresolved);
+      return jobDone({
+        geo_backfill_scanned: summary.scanned,
+        geo_backfill_located: summary.located,
+        geo_backfill_carrier_resolved: summary.carrierResolved,
+        geo_backfill_still_unresolved: summary.stillUnresolved,
+      });
     } catch (err) {
-      // The admin sign-in overview tolerates a stale Standort cell —
-      // log and move on so a one-off resolver hiccup does not poison
-      // the queue and block the next pass.
+      // The admin sign-in overview tolerates a stale Standort cell, so the
+      // warning stays — but a pass that resolved nothing did not do its work.
       evt.addWarning(`geo-backfill failed: ${err}`);
+      return jobFailed("geo backfill pass failed", err);
     } finally {
       geoBackfillRunning = false;
     }
   });
 }
 
-export async function handleTlsPinMonitor(jobs: Job<TlsPinMonitorPayload>[]) {
+export async function handleTlsPinMonitor(
+  jobs: Job<TlsPinMonitorPayload>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.tls_pin_monitor", async (evt) => {
+  return withBackgroundEvent("job.tls_pin_monitor", async (evt) => {
     const p = getWorkerPrisma();
     try {
       const summary = await runTlsPinMonitor(p);
       evt.addMeta("tls_pin_monitor_outcome", summary.outcome);
       evt.addMeta("tls_pin_monitor_host", summary.host);
       evt.addMeta("tls_pin_monitor_known_count", summary.knownPinCount);
+      // No configured target leaves `host` null; the empty string keeps the
+      // fact shape scalar, and `outcome: "skipped"` already names that case.
+      return jobDone({
+        tls_pin_monitor_outcome: summary.outcome,
+        tls_pin_monitor_host: summary.host ?? "",
+        tls_pin_monitor_known_count: summary.knownPinCount,
+      });
     } catch (err) {
       // runTlsPinMonitor swallows probe failures internally; anything that
-      // escapes is unexpected. Log and move on so a one-off failure does not
-      // poison the queue and block the next tick.
+      // escapes is unexpected, and a tick that never compared a pin is
+      // exactly the silence this monitor exists to break.
       evt.addWarning(`tls-pin-monitor failed: ${err}`);
+      return jobFailed("tls pin probe failed", err);
     }
   });
 }
 
 export async function handlePrDetection(
   jobs: Job<PrDetectionPayload | { userId?: undefined }>[],
-) {
+): Promise<JobOutcome> {
+  // Each job in the batch opens its own wide event; the outcome is reported
+  // once for the batch, so the per-job facts accumulate out here.
+  let usersScanned = 0;
+  let insertedAcrossJobs = 0;
+  let tiesAcrossJobs = 0;
+  let usersFailed = 0;
   for (const job of jobs) {
     await withBackgroundEvent("job.pr_detection", async (evt) => {
       const p = getWorkerPrisma();
@@ -170,6 +201,10 @@ export async function handlePrDetection(
           insertedTotal += result.inserted;
           tiesTotal += result.ties;
         } catch (err) {
+          // One user's detection failing is that user's problem, not the
+          // pass's: it rides out as a count so the cron is not retried for
+          // the whole cohort.
+          usersFailed++;
           evt.addWarning(`pr-detection failed for user ${userId}: ${err}`);
         }
       }
@@ -178,6 +213,18 @@ export async function handlePrDetection(
       evt.addMeta("pr_detection_ties", tiesTotal);
       evt.addMeta("pr_detection_silent", silent);
       evt.addMeta("pr_detection_mode", payloadUserId ? "ingest" : "cron");
+
+      usersScanned += userIds.length;
+      insertedAcrossJobs += insertedTotal;
+      tiesAcrossJobs += tiesTotal;
     });
   }
+
+  return jobDone({
+    pr_detection_jobs: jobs.length,
+    pr_detection_users: usersScanned,
+    pr_detection_inserted: insertedAcrossJobs,
+    pr_detection_ties: tiesAcrossJobs,
+    pr_detection_users_failed: usersFailed,
+  });
 }

src/lib/jobs/reminder/poll-cohort.ts

@@ -21,6 +21,7 @@ import { type Job } from "pg-boss";
 import { fireAndForget } from "@/lib/logging/fire-and-forget";
 
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { isSyncFailureRecorded } from "@/lib/integrations/status";
@@ -58,6 +59,13 @@ export interface PollCohortConfig {
  * handler resolves its targets from the batch payloads (single-user enqueue) or
  * the cohort query (cron tick), then re-syncs each sequentially with per-user
  * error isolation so one revoked grant / outage never starves the cohort.
+ *
+ * Outcome semantics for a fan-out pass: the JOB succeeded if the pass ran.
+ * One user's revoked grant is that user's problem — it belongs on their
+ * integration ledger, not in a queue-wide failure that retries the whole
+ * cohort — so per-user failures ride out as the `users_failed` count. Only a
+ * failure of the pass itself (cohort discovery, the wide-event scope) fails
+ * the job, and that path already throws.
  */
 export function makePollCohortHandler({
   taskName,
@@ -65,8 +73,10 @@ export function makePollCohortHandler({
   syncUser,
   recordFailure,
 }: PollCohortConfig) {
-  return async function handlePollCohort(jobs: Job<PollCohortPayload>[]) {
-    await withBackgroundEvent(taskName, async (evt) => {
+  return async function handlePollCohort(
+    jobs: Job<PollCohortPayload>[],
+  ): Promise<JobOutcome> {
+    return withBackgroundEvent(taskName, async (evt) => {
       const prisma = getWorkerPrisma();
       try {
         const targets: Array<{ userId: string }> = [];
@@ -77,9 +87,12 @@ export function makePollCohortHandler({
           const ids = await findCohort(prisma);
           targets.push(...ids.map((userId) => ({ userId })));
         }
-        if (targets.length === 0) return;
+        if (targets.length === 0) {
+          return jobDone({ total: 0, users_synced: 0, users_failed: 0 });
+        }
 
         let usersSynced = 0;
+        let usersFailed = 0;
         let measurementsImported = 0;
         for (const { userId } of targets) {
           try {
@@ -106,6 +119,7 @@ export function makePollCohortHandler({
                 );
               });
             }
+            usersFailed++;
             evt.addWarning(`${taskName} failed for user ${userId}: ${err}`);
           }
         }
@@ -114,10 +128,17 @@ export function makePollCohortHandler({
           task_name: taskName,
           result: {
             users_synced: usersSynced,
+            users_failed: usersFailed,
             total: targets.length,
             measurements_imported: measurementsImported,
           },
         });
+        return jobDone({
+          total: targets.length,
+          users_synced: usersSynced,
+          users_failed: usersFailed,
+          measurements_imported: measurementsImported,
+        });
       } catch (err) {
         evt.setError(err);
         recordError();

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

@@ -7,9 +7,15 @@
  *
  * 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
  * `cycle-reminder-queue` and `measurement-reminder-queue` guards read THIS
  * module.
+ *
+ * Retry policy: every queue here keeps the pg-boss default. These ticks are
+ * fan-out passes whose per-user failures never fail the job, so a job-level
+ * failure means the pass itself could not run — a transient the default two
+ * retries are the right answer to. The satisfy queue carries its own
+ * `retryLimit` at send time (see `reminder-satisfy.ts`).
  */
 import { PgBoss } from "pg-boss";
 import {
@@ -17,7 +23,11 @@ import {
   REMINDER_SATISFY_CONCURRENCY,
   type ReminderSatisfyPayload,
 } from "@/lib/jobs/reminder-satisfy";
-import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  createAndSchedule,
+  createAndWork,
+  type ScheduleEntry,
+} from "./registrar-shared";
 import {
   ReminderCheckPayload,
   handleReminderCheck,
@@ -121,7 +131,8 @@ export async function registerReminderQueues(
 ): Promise<readonly string[]> {
   await createAndSchedule(boss, allQueues, schedules);
 
-  await boss.work<ReminderCheckPayload>(
+  await createAndWork<ReminderCheckPayload>(
+    boss,
     QUEUE_NAME,
     { localConcurrency: 1 },
     handleReminderCheck,
@@ -130,7 +141,8 @@ export async function registerReminderQueues(
   // two reminder ticks from interleaving against the same user row;
   // the dedup ledger would still save us, but skipping the race here
   // avoids spurious P2002 errors in the wide-event log.
-  await boss.work<MoodReminderPayload>(
+  await createAndWork<MoodReminderPayload>(
+    boss,
     MOOD_REMINDER_QUEUE,
     { localConcurrency: 1 },
     handleMoodReminderCheck,
@@ -138,7 +150,8 @@ export async function registerReminderQueues(
   // v1.15 — single-flight cycle-reminder worker. localConcurrency=1 keeps
   // two ticks from racing the fire-and-forget `push_attempts` ledger that
   // anchors the per-day idempotency, exactly like the mood-reminder worker.
-  await boss.work<CycleReminderPayload>(
+  await createAndWork<CycleReminderPayload>(
+    boss,
     CYCLE_REMINDER_QUEUE,
     { localConcurrency: 1 },
     handleCycleReminderCheck,
@@ -147,7 +160,8 @@ export async function registerReminderQueues(
   // localConcurrency=1 keeps two ticks from racing the `nextDueAt`
   // advance that anchors the per-cycle idempotency, exactly like the
   // mood / cycle reminder workers.
-  await boss.work<MeasurementReminderPayload>(
+  await createAndWork<MeasurementReminderPayload>(
+    boss,
     MEASUREMENT_REMINDER_QUEUE,
     { localConcurrency: 1 },
     handleMeasurementReminderCheck,
@@ -155,7 +169,8 @@ export async function registerReminderQueues(
   // v1.18.1 — eventful Vorsorge satisfaction. Resolves a user's reminders
   // against their just-landed measurement / lab. Read-heavy on the user's
   // own data; the same small concurrency budget as PR detection.
-  await boss.work<ReminderSatisfyPayload>(
+  await createAndWork<ReminderSatisfyPayload>(
+    boss,
     REMINDER_SATISFY_QUEUE,
     { localConcurrency: REMINDER_SATISFY_CONCURRENCY },
     handleReminderSatisfy,

src/lib/jobs/reminder/whoop-sync.ts

@@ -7,6 +7,7 @@
 import { type Job } from "pg-boss";
 import { fireAndForget } from "@/lib/logging/fire-and-forget";
 import { recordError } from "@/lib/jobs/worker-status";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import {
   syncUserRecovery,
@@ -66,6 +67,10 @@ type ByIdSync = (userId: string, resourceId: string) => Promise<number>;
  *   - A cron tick (no `userId` on any job) walks every connection's collection.
  *
  * One user's parked-at-reauth state never starves the rest of the cohort.
+ *
+ * The pass is what the outcome judges: a per-user failure is that user's
+ * problem, recorded on their integration ledger and counted as `users_failed`,
+ * not a queue-wide failure that would retry the whole cohort.
  */
 export async function runWhoopResourceSync(
   taskName: string,
@@ -73,8 +78,8 @@ export async function runWhoopResourceSync(
   jobs: Job<WhoopSyncPayload>[],
   syncFn: CollectionSync,
   byIdFn?: ByIdSync,
-): Promise<void> {
-  await withBackgroundEvent(taskName, async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent(taskName, async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       // (userId, optional resourceId) work items from the webhook payloads.
@@ -94,9 +99,17 @@ export async function runWhoopResourceSync(
         });
         targets.push(...connections);
       }
-      if (targets.length === 0) return;
+      if (targets.length === 0) {
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          measurements_imported: 0,
+        });
+      }
 
       let usersSynced = 0;
+      let usersFailed = 0;
       let measurementsImported = 0;
       for (const { userId, resourceId } of targets) {
         try {
@@ -123,6 +136,7 @@ export async function runWhoopResourceSync(
           if (!isSyncFailureRecorded(err)) {
             await recordWhoopSyncFailure(userId, err, leg);
           }
+          usersFailed++;
           evt.addWarning(`${taskName} failed for user ${userId}: ${err}`);
         }
       }
@@ -135,6 +149,12 @@ export async function runWhoopResourceSync(
           measurements_imported: measurementsImported,
         },
       });
+      return jobDone({
+        total: targets.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        measurements_imported: measurementsImported,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -143,7 +163,9 @@ export async function runWhoopResourceSync(
   });
 }
 
-export function handleWhoopRecoverySync(jobs: Job<WhoopSyncPayload>[]) {
+export function handleWhoopRecoverySync(
+  jobs: Job<WhoopSyncPayload>[],
+): Promise<JobOutcome> {
   return runWhoopResourceSync(
     "job.whoop_recovery_sync",
     "recovery",
@@ -153,7 +175,9 @@ export function handleWhoopRecoverySync(jobs: Job<WhoopSyncPayload>[]) {
   );
 }
 
-export function handleWhoopSleepSync(jobs: Job<WhoopSyncPayload>[]) {
+export function handleWhoopSleepSync(
+  jobs: Job<WhoopSyncPayload>[],
+): Promise<JobOutcome> {
   return runWhoopResourceSync(
     "job.whoop_sleep_sync",
     "sleep",
@@ -163,7 +187,9 @@ export function handleWhoopSleepSync(jobs: Job<WhoopSyncPayload>[]) {
   );
 }
 
-export function handleWhoopWorkoutSync(jobs: Job<WhoopSyncPayload>[]) {
+export function handleWhoopWorkoutSync(
+  jobs: Job<WhoopSyncPayload>[],
+): Promise<JobOutcome> {
   return runWhoopResourceSync(
     "job.whoop_workout_sync",
     "workout",
@@ -173,7 +199,9 @@ export function handleWhoopWorkoutSync(jobs: Job<WhoopSyncPayload>[]) {
   );
 }
 
-export function handleWhoopCycleSync(jobs: Job<WhoopSyncPayload>[]) {
+export function handleWhoopCycleSync(
+  jobs: Job<WhoopSyncPayload>[],
+): Promise<JobOutcome> {
   // Cycle has no webhook (poll-only), so no fetch-by-id path.
   return runWhoopResourceSync(
     "job.whoop_cycle_sync",

src/lib/jobs/reminder/withings-sync.ts

@@ -7,6 +7,7 @@
 import { type Job } from "pg-boss";
 import { fireAndForget } from "@/lib/logging/fire-and-forget";
 import { recordError, recordWithingsSync } from "@/lib/jobs/worker-status";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import {
   retryDueWithingsWebhookSubscriptions,
@@ -50,15 +51,20 @@ export interface WithingsSleepSyncPayload {
  */
 export async function handleWithingsFallbackSync(
   jobs: Job<WithingsSyncPayload>[],
-) {
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.withings_sync", async (evt) => {
+  return withBackgroundEvent("job.withings_sync", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       recordWithingsSync();
+      // The subscription repair is an auxiliary leg, not the pass this queue is
+      // named for. Failing the job on it would skip the fallback sync that every
+      // user relies on when a webhook goes missing, so it rides out as a fact.
+      let subscriptionRepairFailed = false;
       try {
         await retryDueWithingsWebhookSubscriptions();
       } catch {
+        subscriptionRepairFailed = true;
         evt.addWarning(
           "Withings subscription repair failed; continuing fallback sync",
         );
@@ -68,10 +74,17 @@ export async function handleWithingsFallbackSync(
       });
 
       if (connections.length === 0) {
-        return;
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          measurements_imported: 0,
+          subscription_repair_failed: subscriptionRepairFailed,
+        });
       }
 
       let usersSynced = 0;
+      let usersFailed = 0;
       let measurementsImported = 0;
 
       for (const connection of connections) {
@@ -87,6 +100,7 @@ export async function handleWithingsFallbackSync(
             });
           }
         } catch {
+          usersFailed++;
           evt.addWarning(`Fallback sync failed for user ${connection.userId}`);
         }
       }
@@ -99,6 +113,13 @@ export async function handleWithingsFallbackSync(
           measurements_imported: measurementsImported,
         },
       });
+      return jobDone({
+        total: connections.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        measurements_imported: measurementsImported,
+        subscription_repair_failed: subscriptionRepairFailed,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -127,8 +148,8 @@ export async function handleWithingsFallbackSync(
  */
 export async function handleWithingsEcgSync(
   jobs: Job<WithingsEcgSyncPayload>[],
-) {
-  await withBackgroundEvent("job.withings_ecg_sync", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.withings_ecg_sync", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const targets: Array<{
@@ -152,9 +173,17 @@ export async function handleWithingsEcgSync(
         });
         targets.push(...connections);
       }
-      if (targets.length === 0) return;
+      if (targets.length === 0) {
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          recordings_imported: 0,
+        });
+      }
 
       let usersSynced = 0;
+      let usersFailed = 0;
       let recordingsImported = 0;
       for (const { userId, startdate, enddate } of targets) {
         try {
@@ -165,6 +194,7 @@ export async function handleWithingsEcgSync(
           recordingsImported += await syncUserEcg(userId, options);
           usersSynced++;
         } catch (err) {
+          usersFailed++;
           evt.addWarning(`Withings ECG sync failed for user ${userId}: ${err}`);
         }
       }
@@ -177,6 +207,12 @@ export async function handleWithingsEcgSync(
           recordings_imported: recordingsImported,
         },
       });
+      return jobDone({
+        total: targets.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        recordings_imported: recordingsImported,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -203,8 +239,8 @@ export async function handleWithingsEcgSync(
  */
 export async function handleWithingsActivitySync(
   jobs: Job<WithingsActivitySyncPayload>[],
-) {
-  await withBackgroundEvent("job.withings_activity_sync", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.withings_activity_sync", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const targets: Array<{ userId: string }> = [];
@@ -220,9 +256,17 @@ export async function handleWithingsActivitySync(
         });
         targets.push(...connections);
       }
-      if (targets.length === 0) return;
+      if (targets.length === 0) {
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          measurements_imported: 0,
+        });
+      }
 
       let usersSynced = 0;
+      let usersFailed = 0;
       let measurementsImported = 0;
       for (const { userId } of targets) {
         try {
@@ -235,6 +279,7 @@ export async function handleWithingsActivitySync(
             });
           }
         } catch {
+          usersFailed++;
           evt.addWarning(`Withings activity sync failed for user ${userId}`);
         }
       }
@@ -247,6 +292,12 @@ export async function handleWithingsActivitySync(
           measurements_imported: measurementsImported,
         },
       });
+      return jobDone({
+        total: targets.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        measurements_imported: measurementsImported,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -262,8 +313,8 @@ export async function handleWithingsActivitySync(
  */
 export async function handleWithingsSleepSync(
   jobs: Job<WithingsSleepSyncPayload>[],
-) {
-  await withBackgroundEvent("job.withings_sleep_sync", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.withings_sleep_sync", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
       const targets: Array<{ userId: string }> = [];
@@ -278,9 +329,17 @@ export async function handleWithingsSleepSync(
         });
         targets.push(...connections);
       }
-      if (targets.length === 0) return;
+      if (targets.length === 0) {
+        return jobDone({
+          total: 0,
+          users_synced: 0,
+          users_failed: 0,
+          measurements_imported: 0,
+        });
+      }
 
       let usersSynced = 0;
+      let usersFailed = 0;
       let measurementsImported = 0;
       for (const { userId } of targets) {
         try {
@@ -293,6 +352,7 @@ export async function handleWithingsSleepSync(
             });
           }
         } catch {
+          usersFailed++;
           evt.addWarning(`Withings sleep sync failed for user ${userId}`);
         }
       }
@@ -305,6 +365,12 @@ export async function handleWithingsSleepSync(
           measurements_imported: measurementsImported,
         },
       });
+      return jobDone({
+        total: targets.length,
+        users_synced: usersSynced,
+        users_failed: usersFailed,
+        measurements_imported: measurementsImported,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();

src/lib/jobs/restore-drill.ts

@@ -30,6 +30,7 @@ import {
   type S3Like,
 } from "@/lib/jobs/offhost-backup";
 import { reportWorkerError } from "@/lib/jobs/report-worker-error";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 
 export const RESTORE_DRILL_QUEUE = "data-restore-drill";
@@ -149,9 +150,11 @@ export async function runRestoreDrill(
  * page); every other failure goes through `reportWorkerError` so the
  * operator hears about a rotting backup chain.
  */
-export async function handleRestoreDrill(jobs: Job<object>[]): Promise<void> {
+export async function handleRestoreDrill(
+  jobs: Job<object>[],
+): Promise<JobOutcome> {
   void jobs;
-  await withBackgroundEvent("job.restore_drill", async (evt) => {
+  return withBackgroundEvent("job.restore_drill", async (evt) => {
     try {
       const report = await runRestoreDrill();
       evt.addMeta("restore_drill_object_key", report.objectKey);
@@ -181,13 +184,31 @@ export async function handleRestoreDrill(jobs: Job<object>[]): Promise<void> {
           { objectKey: report.objectKey, ageDays: report.ageDays },
         );
       }
+      // A stale chain is a verdict about the uploader, not a failure of the
+      // drill: the artefact was fetched, decrypted and parsed, and the page
+      // above already reached the operator. Failing the job here would retry a
+      // reading that cannot change until the next upload lands.
+      return jobDone({
+        restore_drill_object_key: report.objectKey,
+        restore_drill_age_days: report.ageDays,
+        restore_drill_ciphertext_bytes: report.ciphertextBytes,
+        restore_drill_plaintext_bytes: report.plaintextBytes,
+        restore_drill_measurements: report.recordCounts.measurements,
+        restore_drill_medications: report.recordCounts.medications,
+        restore_drill_intake_events: report.recordCounts.intakeEvents,
+        restore_drill_mood_entries: report.recordCounts.moodEntries,
+        restore_drill_stale: report.stale,
+      });
     } catch (err) {
       if (err instanceof OffhostBackupNotConfiguredError) {
         evt.addWarning(`restore-drill skipped: ${err.message}`);
-        return;
+        // Most self-hosters never set the S3 vars. Nothing to verify is not a
+        // failed verification, and a monthly failed job would be noise.
+        return jobDone({ skipped: "offhost backup not configured" });
       }
       evt.addWarning(`restore-drill failed: ${err}`);
       await reportWorkerError(RESTORE_DRILL_QUEUE, err);
+      return jobFailed("restore drill failed", err);
     }
   });
 }

src/lib/jobs/run-job.ts

@@ -45,6 +45,12 @@ export class JobFailure extends Error {
     this.name = "JobFailure";
     this.queue = queue;
     this.reason = reason;
+    // Carry the thrown error's stack. Without this the only stack the
+    // operator ever sees points at `runJob`, which is the same for all 101
+    // queues and says nothing about where the work actually broke.
+    if (cause instanceof Error && cause.stack) {
+      this.stack = `${this.name}: ${this.message}\ncaused by ${cause.stack}`;
+    }
   }
 }
 

src/lib/jobs/workout-insight-generate.ts

@@ -48,6 +48,7 @@ import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-work
 import { defaultLocale, locales, type Locale } from "@/lib/i18n/config";
 import { finalizeStatusSummary } from "@/lib/insights/status-shared";
 import { runStatusCompletion } from "@/lib/insights/status-provider";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { annotate } from "@/lib/logging/context";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { resolveModuleMap } from "@/lib/modules/gate";
@@ -524,10 +525,14 @@ export async function runWorkoutInsightGenerate(
 
 export async function handleWorkoutInsightGenerate(
   jobs: Job<WorkoutInsightGeneratePayload>[],
-): Promise<void> {
-  await withBackgroundEvent("job.workout_insight_generate", async (evt) => {
+): Promise<JobOutcome> {
+  return withBackgroundEvent("job.workout_insight_generate", async (evt) => {
+    let generated = 0;
+    let skipped = 0;
     for (const job of jobs) {
       const outcome = await runWorkoutInsightGenerate(job.data);
+      if (outcome.status === "skipped") skipped++;
+      else generated++;
       evt.addMeta(
         "workout_insight",
         outcome.status === "skipped"
@@ -535,5 +540,9 @@ export async function handleWorkoutInsightGenerate(
           : `generated:${outcome.providerType}`,
       );
     }
+    // Every gate in front of the provider returns `skipped`, so a pass that
+    // wrote no paragraph at all is still a pass that ran. Only a transient
+    // fault escapes `runWorkoutInsightGenerate`, and that one throws.
+    return jobDone({ total: jobs.length, generated, skipped });
   });
 }