feat(01-22): publish truthful cohort outcomes

- fold complete partial failed parked and skipped user verdicts
- bound serialized job facts to a privacy-safe scalar allowlist
- preserve provider retry isolation and reminder queue completion

src/lib/jobs/job-outcome.ts

@@ -25,7 +25,7 @@
  * 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
+ * stable codes, booleans. It rides into the failure report and, once the
  * failure ledger lands, into the operator surface. Keep it to scalars that
  * a dashboard can key on; it is not a place for payload data or free text.
  */
@@ -36,6 +36,45 @@ export type JobFact = string | number | boolean;
 /** The pinned-shape bag of facts a handler reports about its run. */
 export type JobFacts = Readonly<Record<string, JobFact>>;
 
+/**
+ * Facts cross a persistence/observability boundary. Keep their vocabulary
+ * explicit so a caller cannot attach an identifier, credential, URL,
+ * free-form error, or health value to a durable job result.
+ */
+export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
+  "provider",
+  "outcome",
+  "total",
+  "users_synced",
+  "users_complete",
+  "users_partial",
+  "users_failed",
+  "users_parked",
+  "users_skipped",
+  "users_useful",
+  "users_clean_zero",
+  "users_retryable",
+  "downstream_failed",
+  "measurements_imported",
+  "deleted",
+  "generated",
+  "jobs",
+  "skipped",
+  "processed",
+  "created",
+  "updated",
+  "failed",
+  "retried",
+  "queued",
+  "sent",
+  "subscription_repair_failed",
+]);
+
+export const MAX_JOB_FACTS = 32;
+export const MAX_JOB_OUTCOME_BYTES = 2_048;
+
+const STABLE_CODE_PATTERN = /^[a-z][a-z0-9_]{0,63}$/;
+
 /**
  * What a handler did. `ok: true` means the work named by the queue actually
  * happened; `ok: false` means it did not, and `runJob` will fail the pg-boss
@@ -79,3 +118,78 @@ export function jobFailed(
 ): JobOutcome {
   return { ok: false, reason, cause, did };
 }
+
+export type SerializedJobOutcome =
+  | { readonly ok: true; readonly did: JobFacts }
+  | {
+      readonly ok: false;
+      readonly reason_code: string;
+      readonly did?: JobFacts;
+    };
+
+function assertStableCode(value: string, label: string): void {
+  if (!STABLE_CODE_PATTERN.test(value)) {
+    throw new TypeError(`${label} must be a stable lowercase code`);
+  }
+}
+
+function validateFacts(facts: JobFacts): JobFacts {
+  const entries = Object.entries(facts);
+
+  if (entries.length > MAX_JOB_FACTS) {
+    throw new RangeError(`job outcome exceeds ${MAX_JOB_FACTS} facts`);
+  }
+
+  for (const [key, value] of entries) {
+    if (!JOB_FACT_ALLOWLIST.has(key)) {
+      throw new TypeError(`job fact key is not allowed: ${key}`);
+    }
+
+    if (typeof value === "number") {
+      if (!Number.isSafeInteger(value) || value < 0) {
+        throw new TypeError(
+          `job fact ${key} must be a non-negative safe integer`,
+        );
+      }
+      continue;
+    }
+
+    if (typeof value === "string") {
+      assertStableCode(value, `job fact ${key}`);
+      continue;
+    }
+
+    if (typeof value !== "boolean") {
+      throw new TypeError(`job fact ${key} must be a bounded scalar`);
+    }
+  }
+
+  return Object.freeze(Object.fromEntries(entries)) as JobFacts;
+}
+
+/**
+ * Produce the only representation suitable for persistence or structured
+ * logging. The raw `cause` deliberately remains in-process for retry handling.
+ */
+export function serializeJobOutcome(outcome: JobOutcome): SerializedJobOutcome {
+  const did = validateFacts(outcome.did ?? {});
+  const serialized: SerializedJobOutcome = outcome.ok
+    ? { ok: true, did }
+    : (() => {
+        assertStableCode(outcome.reason, "job failure reason");
+        return outcome.did === undefined
+          ? { ok: false, reason_code: outcome.reason }
+          : { ok: false, reason_code: outcome.reason, did };
+      })();
+
+  const byteLength = new TextEncoder().encode(
+    JSON.stringify(serialized),
+  ).length;
+  if (byteLength > MAX_JOB_OUTCOME_BYTES) {
+    throw new RangeError(
+      `serialized job outcome exceeds ${MAX_JOB_OUTCOME_BYTES} bytes`,
+    );
+  }
+
+  return serialized;
+}

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

@@ -5,13 +5,19 @@
  * queue names, cron schedules, and boss.work registrations.
  */
 import { type Job } from "pg-boss";
+import pLimit from "p-limit";
 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 { syncUserGoogleHealth } from "@/lib/google-health/sync";
+import { isReauthRequired } from "@/lib/integrations/status";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup";
+import {
+  foldIntegrationCohortOutcomes,
+  type IntegrationUserVerdict,
+} from "./poll-cohort";
 import { getWorkerPrisma } from "./shared";
 
 /**
@@ -42,56 +48,69 @@ export async function handleGoogleHealthSync(
         targets.push(...connections);
       }
       if (targets.length === 0) {
-        return jobDone({
-          total: 0,
-          users_synced: 0,
-          users_failed: 0,
-          measurements_imported: 0,
+        return foldIntegrationCohortOutcomes({
+          provider: "google_health",
+          verdicts: [],
         });
       }
 
       // 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) => {
-              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) => {
-              if (imported > 0) {
+      const limit = pLimit(4);
+      const verdicts = await Promise.all(
+        targets.map(({ userId }) =>
+          limit(async (): Promise<IntegrationUserVerdict> => {
+            if (await isReauthRequired(userId, "google-health")) {
+              return { status: "parked", imported: 0 };
+            }
+
+            try {
+              const result = await syncUserGoogleHealth(userId);
+              const parked =
+                result.failed &&
+                result.imported === 0 &&
+                (await isReauthRequired(userId, "google-health"));
+              const status = parked
+                ? "parked"
+                : result.failed
+                  ? result.imported > 0
+                    ? "partial"
+                    : "failed"
+                  : "complete";
+
+              if (result.imported > 0) {
                 fireAndForget(enqueueReminderSatisfy(userId), {
                   action: "reminder.satisfy.enqueue",
                 });
               }
-            },
-          },
-        );
+
+              return {
+                status,
+                imported: result.imported,
+                retryable: result.failed && !parked,
+              };
+            } catch {
+              evt.addWarning(
+                "job.google_health_sync failed for one cohort member",
+              );
+              return { status: "failed", imported: 0, retryable: true };
+            }
+          }),
+        ),
+      );
+      const outcome = foldIntegrationCohortOutcomes({
+        provider: "google_health",
+        verdicts,
+      });
 
       evt.setBackground({
         task_name: "job.google_health_sync",
-        result: {
-          users_synced: usersSynced,
-          total: targets.length,
-          measurements_imported: measurementsImported,
-        },
+        result: outcome.did,
       });
       // 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,
-      });
+      return outcome;
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -117,8 +136,8 @@ export async function handleGoogleHealthOAuthStateCleanup(
         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);
+        evt.addWarning("google-health-oauth-state-cleanup failed");
+        return jobFailed("google_health_oauth_state_cleanup_failed", err);
       }
     },
   );

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

@@ -27,6 +27,129 @@ import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { isSyncFailureRecorded } from "@/lib/integrations/status";
 import { getWorkerPrisma } from "./shared";
 
+export type IntegrationCohortProvider = "google_health" | "withings";
+
+export type IntegrationUserVerdict = Readonly<{
+  status: "complete" | "partial" | "failed" | "parked" | "skipped";
+  imported: number;
+  downstreamFailed?: boolean;
+  retryable?: boolean;
+  reasonCode?: string;
+}>;
+
+export type IntegrationCohortFacts = Readonly<{
+  provider: IntegrationCohortProvider;
+  outcome: "useful" | "clean_zero";
+  total: number;
+  users_synced: number;
+  users_complete: number;
+  users_partial: number;
+  users_failed: number;
+  users_parked: number;
+  users_skipped: number;
+  users_useful: number;
+  users_clean_zero: number;
+  users_retryable: number;
+  downstream_failed: number;
+  measurements_imported: number;
+}>;
+
+export function foldIntegrationCohortOutcomes(input: {
+  provider: IntegrationCohortProvider;
+  verdicts: readonly IntegrationUserVerdict[];
+  downstreamFailures?: number;
+}): { readonly ok: true; readonly did: IntegrationCohortFacts } {
+  let usersComplete = 0;
+  let usersPartial = 0;
+  let usersFailed = 0;
+  let usersParked = 0;
+  let usersSkipped = 0;
+  let usersUseful = 0;
+  let usersCleanZero = 0;
+  let usersRetryable = 0;
+  if (
+    input.downstreamFailures !== undefined &&
+    (!Number.isSafeInteger(input.downstreamFailures) ||
+      input.downstreamFailures < 0)
+  ) {
+    throw new TypeError(
+      "cohort downstream failure count must be a non-negative safe integer",
+    );
+  }
+  let downstreamFailed = input.downstreamFailures ?? 0;
+  let measurementsImported = 0;
+
+  for (const verdict of input.verdicts) {
+    if (!Number.isSafeInteger(verdict.imported) || verdict.imported < 0) {
+      throw new TypeError(
+        "cohort imported count must be a non-negative safe integer",
+      );
+    }
+
+    switch (verdict.status) {
+      case "complete":
+        usersComplete++;
+        if (verdict.imported === 0) usersCleanZero++;
+        break;
+      case "partial":
+        usersPartial++;
+        break;
+      case "failed":
+        usersFailed++;
+        break;
+      case "parked":
+        usersParked++;
+        break;
+      case "skipped":
+        usersSkipped++;
+        break;
+      default: {
+        const exhaustive: never = verdict.status;
+        throw new TypeError(`unsupported cohort status: ${String(exhaustive)}`);
+      }
+    }
+
+    if (
+      (verdict.status === "complete" || verdict.status === "partial") &&
+      verdict.imported > 0
+    ) {
+      usersUseful++;
+    }
+    if (verdict.retryable === true) usersRetryable++;
+    if (verdict.downstreamFailed === true) {
+      if (!Number.isSafeInteger(downstreamFailed + 1)) {
+        throw new RangeError(
+          "cohort downstream failure total exceeds safe integer range",
+        );
+      }
+      downstreamFailed++;
+    }
+
+    const nextImported = measurementsImported + verdict.imported;
+    if (!Number.isSafeInteger(nextImported)) {
+      throw new RangeError("cohort imported total exceeds safe integer range");
+    }
+    measurementsImported = nextImported;
+  }
+
+  return jobDone({
+    provider: input.provider,
+    outcome: usersUseful > 0 ? "useful" : "clean_zero",
+    total: input.verdicts.length,
+    users_synced: usersComplete,
+    users_complete: usersComplete,
+    users_partial: usersPartial,
+    users_failed: usersFailed,
+    users_parked: usersParked,
+    users_skipped: usersSkipped,
+    users_useful: usersUseful,
+    users_clean_zero: usersCleanZero,
+    users_retryable: usersRetryable,
+    downstream_failed: downstreamFailed,
+    measurements_imported: measurementsImported,
+  }) as { readonly ok: true; readonly did: IntegrationCohortFacts };
+}
+
 /** Every poll-cohort handler accepts an optional single-user payload. */
 export interface PollCohortPayload {
   userId?: string;
@@ -113,14 +236,12 @@ export function makePollCohortHandler({
             // Only an UNMARKED escape — today's Oura / Polar write-path throw,
             // or any future provider path — is recorded here.
             if (!isSyncFailureRecorded(err)) {
-              await recordFailure(userId, err).catch((recErr) => {
-                evt.addWarning(
-                  `${taskName} recordFailure failed for user ${userId}: ${recErr}`,
-                );
+              await recordFailure(userId, err).catch(() => {
+                evt.addWarning(`${taskName} recordFailure failed`);
               });
             }
             usersFailed++;
-            evt.addWarning(`${taskName} failed for user ${userId}: ${err}`);
+            evt.addWarning(`${taskName} failed for one cohort member`);
           }
         }
 

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

@@ -7,8 +7,9 @@
 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 { type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
+import { isReauthRequired } from "@/lib/integrations/status";
 import {
   retryDueWithingsWebhookSubscriptions,
   syncUserMeasurements,
@@ -18,6 +19,10 @@ import { syncUserSleep } from "@/lib/withings/sync-sleep";
 import { syncUserEcg } from "@/lib/withings/sync-ecg";
 import type { WithingsEcgSyncPayload } from "@/lib/jobs/withings-ecg-queue";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import {
+  foldIntegrationCohortOutcomes,
+  type IntegrationUserVerdict,
+} from "./poll-cohort";
 import { getWorkerPrisma } from "./shared";
 export type { WithingsEcgSyncPayload } from "@/lib/jobs/withings-ecg-queue";
 
@@ -74,52 +79,61 @@ export async function handleWithingsFallbackSync(
       });
 
       if (connections.length === 0) {
-        return jobDone({
-          total: 0,
-          users_synced: 0,
-          users_failed: 0,
-          measurements_imported: 0,
-          subscription_repair_failed: subscriptionRepairFailed,
+        return foldIntegrationCohortOutcomes({
+          provider: "withings",
+          verdicts: [],
+          downstreamFailures: subscriptionRepairFailed ? 1 : 0,
         });
       }
 
-      let usersSynced = 0;
-      let usersFailed = 0;
-      let measurementsImported = 0;
+      const verdicts: IntegrationUserVerdict[] = [];
 
       for (const connection of connections) {
+        if (await isReauthRequired(connection.userId, "withings")) {
+          verdicts.push({ status: "parked", imported: 0 });
+          continue;
+        }
+
         try {
-          const { imported } = await syncUserMeasurements(connection.userId);
-          usersSynced++;
-          measurementsImported += imported;
+          const result = await syncUserMeasurements(connection.userId);
+          const parked =
+            result.failed &&
+            result.imported === 0 &&
+            (await isReauthRequired(connection.userId, "withings"));
+          verdicts.push({
+            status: parked
+              ? "parked"
+              : result.failed
+                ? result.imported > 0
+                  ? "partial"
+                  : "failed"
+                : "complete",
+            imported: result.imported,
+            retryable: result.failed && !parked,
+          });
           // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge
           // reminders eventfully. Fire-and-forget; the cron is the net.
-          if (imported > 0) {
+          if (result.imported > 0) {
             fireAndForget(enqueueReminderSatisfy(connection.userId), {
               action: "reminder.satisfy.enqueue",
             });
           }
         } catch {
-          usersFailed++;
-          evt.addWarning(`Fallback sync failed for user ${connection.userId}`);
+          verdicts.push({ status: "failed", imported: 0, retryable: true });
+          evt.addWarning("Fallback sync failed for one cohort member");
         }
       }
 
+      const outcome = foldIntegrationCohortOutcomes({
+        provider: "withings",
+        verdicts,
+        downstreamFailures: subscriptionRepairFailed ? 1 : 0,
+      });
       evt.setBackground({
         task_name: "job.withings_sync",
-        result: {
-          users_synced: usersSynced,
-          total: connections.length,
-          measurements_imported: measurementsImported,
-        },
-      });
-      return jobDone({
-        total: connections.length,
-        users_synced: usersSynced,
-        users_failed: usersFailed,
-        measurements_imported: measurementsImported,
-        subscription_repair_failed: subscriptionRepairFailed,
+        result: outcome.did,
       });
+      return outcome;
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -174,45 +188,46 @@ export async function handleWithingsEcgSync(
         targets.push(...connections);
       }
       if (targets.length === 0) {
-        return jobDone({
-          total: 0,
-          users_synced: 0,
-          users_failed: 0,
-          recordings_imported: 0,
+        return foldIntegrationCohortOutcomes({
+          provider: "withings",
+          verdicts: [],
         });
       }
 
-      let usersSynced = 0;
-      let usersFailed = 0;
-      let recordingsImported = 0;
+      const verdicts: IntegrationUserVerdict[] = [];
       for (const { userId, startdate, enddate } of targets) {
+        if (await isReauthRequired(userId, "withings")) {
+          verdicts.push({ status: "parked", imported: 0 });
+          continue;
+        }
+
         try {
           const options =
             startdate === undefined && enddate === undefined
               ? {}
               : { startdate, enddate };
-          recordingsImported += await syncUserEcg(userId, options);
-          usersSynced++;
-        } catch (err) {
-          usersFailed++;
-          evt.addWarning(`Withings ECG sync failed for user ${userId}: ${err}`);
+          const imported = await syncUserEcg(userId, options);
+          const parked =
+            imported === 0 && (await isReauthRequired(userId, "withings"));
+          verdicts.push({
+            status: parked ? "parked" : "complete",
+            imported,
+          });
+        } catch {
+          verdicts.push({ status: "failed", imported: 0, retryable: true });
+          evt.addWarning("Withings ECG sync failed for one cohort member");
         }
       }
 
+      const outcome = foldIntegrationCohortOutcomes({
+        provider: "withings",
+        verdicts,
+      });
       evt.setBackground({
         task_name: "job.withings_ecg_sync",
-        result: {
-          users_synced: usersSynced,
-          total: targets.length,
-          recordings_imported: recordingsImported,
-        },
-      });
-      return jobDone({
-        total: targets.length,
-        users_synced: usersSynced,
-        users_failed: usersFailed,
-        recordings_imported: recordingsImported,
+        result: outcome.did,
       });
+      return outcome;
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -257,47 +272,47 @@ export async function handleWithingsActivitySync(
         targets.push(...connections);
       }
       if (targets.length === 0) {
-        return jobDone({
-          total: 0,
-          users_synced: 0,
-          users_failed: 0,
-          measurements_imported: 0,
+        return foldIntegrationCohortOutcomes({
+          provider: "withings",
+          verdicts: [],
         });
       }
 
-      let usersSynced = 0;
-      let usersFailed = 0;
-      let measurementsImported = 0;
+      const verdicts: IntegrationUserVerdict[] = [];
       for (const { userId } of targets) {
+        if (await isReauthRequired(userId, "withings")) {
+          verdicts.push({ status: "parked", imported: 0 });
+          continue;
+        }
+
         try {
           const imported = await syncUserActivity(userId);
-          usersSynced++;
-          measurementsImported += imported;
+          const parked =
+            imported === 0 && (await isReauthRequired(userId, "withings"));
+          verdicts.push({
+            status: parked ? "parked" : "complete",
+            imported,
+          });
           if (imported > 0) {
             fireAndForget(enqueueReminderSatisfy(userId), {
               action: "reminder.satisfy.enqueue",
             });
           }
         } catch {
-          usersFailed++;
-          evt.addWarning(`Withings activity sync failed for user ${userId}`);
+          verdicts.push({ status: "failed", imported: 0, retryable: true });
+          evt.addWarning("Withings activity sync failed for one cohort member");
         }
       }
 
+      const outcome = foldIntegrationCohortOutcomes({
+        provider: "withings",
+        verdicts,
+      });
       evt.setBackground({
         task_name: "job.withings_activity_sync",
-        result: {
-          users_synced: usersSynced,
-          total: targets.length,
-          measurements_imported: measurementsImported,
-        },
-      });
-      return jobDone({
-        total: targets.length,
-        users_synced: usersSynced,
-        users_failed: usersFailed,
-        measurements_imported: measurementsImported,
+        result: outcome.did,
       });
+      return outcome;
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -330,47 +345,47 @@ export async function handleWithingsSleepSync(
         targets.push(...connections);
       }
       if (targets.length === 0) {
-        return jobDone({
-          total: 0,
-          users_synced: 0,
-          users_failed: 0,
-          measurements_imported: 0,
+        return foldIntegrationCohortOutcomes({
+          provider: "withings",
+          verdicts: [],
         });
       }
 
-      let usersSynced = 0;
-      let usersFailed = 0;
-      let measurementsImported = 0;
+      const verdicts: IntegrationUserVerdict[] = [];
       for (const { userId } of targets) {
+        if (await isReauthRequired(userId, "withings")) {
+          verdicts.push({ status: "parked", imported: 0 });
+          continue;
+        }
+
         try {
           const imported = await syncUserSleep(userId);
-          usersSynced++;
-          measurementsImported += imported;
+          const parked =
+            imported === 0 && (await isReauthRequired(userId, "withings"));
+          verdicts.push({
+            status: parked ? "parked" : "complete",
+            imported,
+          });
           if (imported > 0) {
             fireAndForget(enqueueReminderSatisfy(userId), {
               action: "reminder.satisfy.enqueue",
             });
           }
         } catch {
-          usersFailed++;
-          evt.addWarning(`Withings sleep sync failed for user ${userId}`);
+          verdicts.push({ status: "failed", imported: 0, retryable: true });
+          evt.addWarning("Withings sleep sync failed for one cohort member");
         }
       }
 
+      const outcome = foldIntegrationCohortOutcomes({
+        provider: "withings",
+        verdicts,
+      });
       evt.setBackground({
         task_name: "job.withings_sleep_sync",
-        result: {
-          users_synced: usersSynced,
-          total: targets.length,
-          measurements_imported: measurementsImported,
-        },
-      });
-      return jobDone({
-        total: targets.length,
-        users_synced: usersSynced,
-        users_failed: usersFailed,
-        measurements_imported: measurementsImported,
+        result: outcome.did,
       });
+      return outcome;
     } catch (err) {
       evt.setError(err);
       recordError();