fix(personal-records): re-derive the cumulative-type records the pre-fix SUM inflated

Before the source-collapse fix, a day with both an Apple Health AND a
Withings write for the same cumulative metric (steps, active energy,
flights, distance, daylight, falls) summed both sources into one
inflated total and locked it in as the all-time PersonalRecord. The
worker's improvement gate only ever compares a freshly-computed day
against the stored best, so the false record silently outranked every
honest day forever and the affected metric could never set a genuine
record again.

Adds a boot-time one-shot repair (queue: cumulative-pr-rederive),
mirroring the existing step-consolidation-repair pattern: deletes the
suspect measurement-driven records created before the fix landed, then
re-runs detection with silent:true so the honest re-derived best
replaces it without firing a "new record" push for what is data
hygiene, not an achievement. Self-converging by a fixed cutoff instant
rather than a marker column or reused SQL migration number.

src/lib/jobs/__tests__/cumulative-pr-rederive-queue.test.ts

@@ -0,0 +1,48 @@
+/**
+ * v1.30.3 (QA F4) — cumulative-PersonalRecord rederivation queue
+ * registration guard.
+ *
+ * Same source-text-grep approach as the step-consolidation-repair guard:
+ * assert the queue is registered in `allQueues`, a `boss.work` handler is
+ * wired against it, and the boot-discovery enqueue helper fires — without
+ * booting pg-boss + Prisma. An unregistered queue silently never drains
+ * (the v1.4.37 dead-queue class).
+ */
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+
+import { describe, expect, it } from "vitest";
+
+const REGISTRAR_PATH = join(__dirname, "..", "reminder", "register-rollup.ts");
+const source = readFileSync(REGISTRAR_PATH, "utf8");
+
+describe("reminder-worker — cumulative-pr-rederive wiring", () => {
+  it("imports the queue symbols from the rederivation module", () => {
+    expect(source).toMatch(
+      /from\s*["']@\/lib\/personal-records\/cumulative-pr-rederivation["']/,
+    );
+    expect(source).toMatch(/\bCUMULATIVE_PR_REDERIVE_QUEUE\b/);
+    expect(source).toMatch(/\brunCumulativePrRederivationForUser\b/);
+    expect(source).toMatch(/\benqueueBootTimeCumulativePrRederivation\b/);
+  });
+
+  it("registers the queue in the allQueues loop", () => {
+    const allQueuesMatch = source.match(
+      /const allQueues\s*=\s*\[([\s\S]*?)\];/,
+    );
+    expect(allQueuesMatch).not.toBeNull();
+    expect(allQueuesMatch![1]).toMatch(/\bCUMULATIVE_PR_REDERIVE_QUEUE\b/);
+  });
+
+  it("registers a boss.work handler against the queue", () => {
+    expect(source).toMatch(
+      /boss\.work[\s\S]{0,200}CUMULATIVE_PR_REDERIVE_QUEUE[\s\S]{0,400}runCumulativePrRederivationForUser/,
+    );
+  });
+
+  it("fires the boot-discovery enqueue helper", () => {
+    expect(source).toMatch(
+      /await enqueueBootTimeCumulativePrRederivation\([^)]*\)/,
+    );
+  });
+});

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

@@ -67,6 +67,13 @@ import {
   enqueueBootTimeStepConsolidationRepair,
   type StepConsolidationRepairPayload,
 } from "@/lib/jobs/step-consolidation-repair";
+import {
+  CUMULATIVE_PR_REDERIVE_QUEUE,
+  CUMULATIVE_PR_REDERIVE_CONCURRENCY,
+  runCumulativePrRederivationForUser,
+  enqueueBootTimeCumulativePrRederivation,
+  type CumulativePrRederivePayload,
+} from "@/lib/personal-records/cumulative-pr-rederivation";
 import {
   MEAN_CONSOLIDATION_QUEUE,
   MEAN_CONSOLIDATION_CONCURRENCY,
@@ -132,6 +139,10 @@ const allQueues = [
   // v1.28.37 — one-shot repair for the provider step rows the pre-fix
   // consolidation swept. Boot-discovery driven, self-converging.
   STEP_CONSOLIDATION_REPAIR_QUEUE,
+  // v1.30.3 — one-shot repair for the cumulative-type PersonalRecord rows
+  // the pre-fix multi-source SUM inflated. Boot-discovery driven,
+  // self-converging (QA F4).
+  CUMULATIVE_PR_REDERIVE_QUEUE,
   // v1.7.0 — daily-mean consolidation for high-frequency spot HealthKit
   // metrics (walking speed/step length, respiratory rate, audio exposure).
   MEAN_CONSOLIDATION_QUEUE,
@@ -248,6 +259,27 @@ export async function registerRollupQueues(
     },
   );
 
+  // v1.30.3 — cumulative-PersonalRecord rederivation worker (QA F4). The
+  // boot enqueue helper below sends one job per user still holding a
+  // suspect pre-fix inflated cumulative-type record; this handler deletes
+  // it and re-runs detection silently so the honest re-derived best takes
+  // its place. Serial concurrency so the repair never crowds the
+  // dashboard request pool.
+  await boss.work<CumulativePrRederivePayload>(
+    CUMULATIVE_PR_REDERIVE_QUEUE,
+    { localConcurrency: CUMULATIVE_PR_REDERIVE_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const summary = await runCumulativePrRederivationForUser(userId);
+        workerLog(
+          "info",
+          `[cumulative-pr-rederive] user=${userId} rowsDeleted=${summary.rowsDeleted} rowsReinserted=${summary.rowsReinserted}`,
+        );
+      }
+    },
+  );
+
   // v1.7.0 — daily-mean consolidation worker. The boot enqueue helper
   // below sends one job per user holding live per-sample high-frequency
   // mean-type rows; this handler collapses each completed day to its
@@ -771,4 +803,35 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
       err,
     );
   }
+
+  // v1.30.3 (QA F4) — cumulative-PersonalRecord rederivation. Finds every
+  // account still holding a suspect measurement-driven cumulative-type
+  // record created before the source-collapse fix landed and enqueues one
+  // repair job per account. Self-converging: a deleted-and-redetected
+  // record's `createdAt` sits after the fix cutoff, so the account drops
+  // out of the discovery predicate for good.
+  try {
+    // Stage 6.
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeCumulativePrRederivation(
+        BOOT_BACKFILL_STAGGER_SECONDS * 6,
+      );
+    if (error) {
+      workerLog(
+        "error",
+        `[cumulative-pr-rederive] boot discovery failed: ${error}`,
+      );
+    } else {
+      workerLog(
+        "info",
+        `[cumulative-pr-rederive] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[cumulative-pr-rederive] boot discovery threw an unexpected error",
+      err,
+    );
+  }
 }

src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts

@@ -0,0 +1,166 @@
+/**
+ * v1.30.3 (QA F4) — the one-shot repair for the pre-fix inflated
+ * cumulative-type PersonalRecord rows: deletes the suspect row(s) for a
+ * user and re-runs detection silently so the honest re-derived best takes
+ * its place, plus the boot-discovery enqueue that finds affected accounts.
+ */
+import { describe, expect, it, vi, beforeEach } from "vitest";
+
+const deleteMany = vi.fn();
+const findMany = vi.fn();
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    personalRecord: {
+      deleteMany: (a: unknown) => deleteMany(a),
+      findMany: (a: unknown) => findMany(a),
+    },
+  },
+}));
+
+const detectPersonalRecordsForUser = vi.fn();
+vi.mock("@/lib/personal-records/pr-detection-worker", () => ({
+  detectPersonalRecordsForUser: (...args: unknown[]) =>
+    detectPersonalRecordsForUser(...args),
+}));
+
+const send = vi.fn();
+const getGlobalBoss = vi.fn();
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: () => getGlobalBoss(),
+}));
+
+const annotate = vi.fn();
+vi.mock("@/lib/logging/context", () => ({
+  annotate: (...a: unknown[]) => annotate(...a),
+}));
+
+import {
+  runCumulativePrRederivationForUser,
+  enqueueBootTimeCumulativePrRederivation,
+  CUMULATIVE_PR_FIX_CUTOFF,
+} from "@/lib/personal-records/cumulative-pr-rederivation";
+
+describe("runCumulativePrRederivationForUser", () => {
+  beforeEach(() => {
+    deleteMany.mockReset();
+    detectPersonalRecordsForUser.mockReset();
+    annotate.mockReset();
+  });
+
+  it("deletes suspect rows created before the fix cutoff and re-derives silently", async () => {
+    deleteMany.mockResolvedValue({ count: 1 });
+    detectPersonalRecordsForUser.mockResolvedValue({
+      inserted: 1,
+      ties: 0,
+      scanned: 6,
+      silent: true,
+    });
+
+    const summary = await runCumulativePrRederivationForUser("u1");
+
+    expect(deleteMany).toHaveBeenCalledWith({
+      where: {
+        userId: "u1",
+        metricSlot: null,
+        metricType: { in: expect.arrayContaining(["ACTIVITY_STEPS"]) },
+        createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF },
+      },
+    });
+    // Re-derivation always runs silent — this is data hygiene, not a new
+    // achievement, so no push should fire for it.
+    expect(detectPersonalRecordsForUser).toHaveBeenCalledWith("u1", {
+      silent: true,
+    });
+    expect(summary).toEqual({ rowsDeleted: 1, rowsReinserted: 1 });
+  });
+
+  it("skips re-derivation entirely when nothing was deleted", async () => {
+    deleteMany.mockResolvedValue({ count: 0 });
+
+    const summary = await runCumulativePrRederivationForUser("u2");
+
+    expect(detectPersonalRecordsForUser).not.toHaveBeenCalled();
+    expect(summary).toEqual({ rowsDeleted: 0, rowsReinserted: 0 });
+  });
+
+  it("counts a tie as a reinserted row (the honest re-derived value can equal a still-live sibling metric's best)", async () => {
+    deleteMany.mockResolvedValue({ count: 2 });
+    detectPersonalRecordsForUser.mockResolvedValue({
+      inserted: 0,
+      ties: 1,
+      scanned: 6,
+      silent: true,
+    });
+
+    const summary = await runCumulativePrRederivationForUser("u3");
+    expect(summary).toEqual({ rowsDeleted: 2, rowsReinserted: 1 });
+  });
+});
+
+describe("enqueueBootTimeCumulativePrRederivation", () => {
+  beforeEach(() => {
+    findMany.mockReset();
+    send.mockReset();
+    getGlobalBoss.mockReset();
+  });
+
+  it("no-ops when no boss is attached (test / scriptless context)", async () => {
+    getGlobalBoss.mockReturnValue(null);
+    const result = await enqueueBootTimeCumulativePrRederivation();
+    expect(result).toEqual({ enqueued: 0, skipped: 0, error: null });
+    expect(findMany).not.toHaveBeenCalled();
+  });
+
+  it("finds zero affected users and enqueues nothing", async () => {
+    getGlobalBoss.mockReturnValue({ send });
+    findMany.mockResolvedValue([]);
+    const result = await enqueueBootTimeCumulativePrRederivation();
+    expect(result).toEqual({ enqueued: 0, skipped: 0, error: null });
+    expect(send).not.toHaveBeenCalled();
+  });
+
+  it("enqueues one job per distinct affected user with a coalescing singletonKey", async () => {
+    getGlobalBoss.mockReturnValue({ send });
+    findMany.mockResolvedValue([{ userId: "u1" }, { userId: "u2" }]);
+    send.mockResolvedValue("job-id");
+
+    const result = await enqueueBootTimeCumulativePrRederivation(30);
+
+    expect(result).toEqual({ enqueued: 2, skipped: 0, error: null });
+    expect(findMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({
+          metricSlot: null,
+          createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF },
+        }),
+        distinct: ["userId"],
+      }),
+    );
+    expect(send).toHaveBeenCalledWith(
+      "cumulative-pr-rederive",
+      expect.objectContaining({ userId: "u1" }),
+      expect.objectContaining({
+        singletonKey: "cumulative-pr-rederive|u1",
+        startAfter: 30,
+      }),
+    );
+  });
+
+  it("counts a coalesced send (pg-boss returns null) as skipped, not an error", async () => {
+    getGlobalBoss.mockReturnValue({ send });
+    findMany.mockResolvedValue([{ userId: "u1" }]);
+    send.mockResolvedValue(null);
+
+    const result = await enqueueBootTimeCumulativePrRederivation();
+    expect(result).toEqual({ enqueued: 0, skipped: 1, error: null });
+  });
+
+  it("returns the error message instead of throwing on a discovery failure", async () => {
+    getGlobalBoss.mockReturnValue({ send });
+    findMany.mockRejectedValue(new Error("db down"));
+
+    const result = await enqueueBootTimeCumulativePrRederivation();
+    expect(result.enqueued).toBe(0);
+    expect(result.error).toBe("db down");
+  });
+});

src/lib/personal-records/cumulative-pr-rederivation.ts

@@ -0,0 +1,184 @@
+/**
+ * v1.30.3 (QA F4) — one-shot repair for the `PersonalRecord` rows the
+ * pre-fix cumulative-day picker double-counted.
+ *
+ * Background: before `8902d3985` (v1.29.6), `findBestCumulativeDay`
+ * (`pr-detection-worker.ts`) summed EVERY row on a calendar day regardless
+ * of source — a day with both an Apple Health AND a Withings write for the
+ * same cumulative metric (steps, active energy, flights, distance,
+ * daylight, falls) summed both sources into one inflated total and wrote
+ * it as the all-time "best". The fix (source-collapse via
+ * `pickCanonicalSourceRows` before summing) stops the inflation going
+ * forward, but the worker's improvement gate only ever compares a
+ * freshly-computed day against the STORED best — an already-inflated
+ * stored row can never be beaten again, so the false record silently
+ * outranks every honest day forever (PR list, digest milestone, doctor
+ * report), and the affected user can never set a genuine record on that
+ * metric again.
+ *
+ * This job deletes the suspect rows (measurement-driven — `metricSlot`
+ * null — one of `CUMULATIVE_HK_TYPES`, created BEFORE the fix landed) and
+ * immediately re-runs detection for the affected user with `silent: true`
+ * so the (already-fixed) worker inserts the honest re-derived best without
+ * firing a "new record" push for what is data hygiene, not an achievement.
+ *
+ * Converges to zero without a marker column: the cutoff is the fixed
+ * instant the source-collapse fix went live upstream (safely after the
+ * `8902d3985` merge). Any row created before it predates the bug fix and
+ * is suspect; any row this job (or the ordinary worker) writes going
+ * forward is always created after that instant, so it can never re-match
+ * the discovery predicate. A re-run (reboot mid-walk, or an install that
+ * upgrades straight past this release) simply finds fewer/no affected
+ * users each time.
+ *
+ * The queue name MUST be registered in `allQueues` in
+ * `src/lib/jobs/reminder/register-rollup.ts` or pg-boss never provisions
+ * it and the boot enqueue silently never drains (the v1.4.37 dead-queue
+ * class).
+ */
+import { prisma } from "@/lib/db";
+import { annotate } from "@/lib/logging/context";
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
+import { detectPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker";
+import type { MeasurementType } from "@/generated/prisma/client";
+
+export const CUMULATIVE_PR_REDERIVE_QUEUE = "cumulative-pr-rederive";
+
+/**
+ * Serial concurrency — a repair walks one account's PersonalRecord rows
+ * and re-runs full detection; matches the other one-shot repair jobs
+ * (`STEP_CONSOLIDATION_REPAIR_CONCURRENCY`).
+ */
+export const CUMULATIVE_PR_REDERIVE_CONCURRENCY = 1;
+
+/**
+ * The instant the cumulative-day source-collapse fix (`8902d3985`,
+ * v1.29.6) went live. Any measurement-driven `PersonalRecord` row for a
+ * cumulative type `createdAt` STRICTLY BEFORE this instant was written by
+ * the pre-fix multi-source SUM and is suspect. Hardcoded (not derived from
+ * git/commit metadata at runtime) so the predicate is a compile-time
+ * constant every self-hosted install evaluates identically regardless of
+ * when it upgrades to this release.
+ */
+export const CUMULATIVE_PR_FIX_CUTOFF = new Date("2026-07-18T00:00:00.000Z");
+
+const CUMULATIVE_TYPES_ARRAY: MeasurementType[] = [
+  ...CUMULATIVE_HK_TYPES,
+] as MeasurementType[];
+
+export interface CumulativePrRederivePayload {
+  userId: string;
+  enqueuedAt: string;
+}
+
+export interface CumulativePrRederiveSummary {
+  /** Suspect rows deleted for this user. */
+  rowsDeleted: number;
+  /** Records the (already-fixed) worker re-inserted after the delete. */
+  rowsReinserted: number;
+}
+
+/**
+ * Per-user repair handler. Deletes the suspect measurement-driven
+ * cumulative-type `PersonalRecord` rows for `userId`, then re-runs
+ * detection silently so the honest re-derived best (if any — the warm-up
+ * gate can still apply) replaces it. Safe to re-run: a second pass finds
+ * zero suspect rows once the honest one has replaced them (its
+ * `createdAt` is always after the cutoff).
+ */
+export async function runCumulativePrRederivationForUser(
+  userId: string,
+): Promise<CumulativePrRederiveSummary> {
+  const { count: rowsDeleted } = await prisma.personalRecord.deleteMany({
+    where: {
+      userId,
+      metricSlot: null,
+      metricType: { in: CUMULATIVE_TYPES_ARRAY },
+      createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF },
+    },
+  });
+
+  let rowsReinserted = 0;
+  if (rowsDeleted > 0) {
+    const result = await detectPersonalRecordsForUser(userId, {
+      silent: true,
+    });
+    rowsReinserted = result.inserted + result.ties;
+  }
+
+  annotate({
+    action: {
+      name: "personal-record.cumulative.rederive",
+      details: { rows_deleted: rowsDeleted, rows_reinserted: rowsReinserted },
+    },
+  });
+
+  return { rowsDeleted, rowsReinserted };
+}
+
+/**
+ * Boot-time discovery. Finds every account still holding a suspect
+ * measurement-driven cumulative-type `PersonalRecord` row (created before
+ * the fix cutoff) and enqueues one repair job per account.
+ *
+ * Self-converging: once a suspect row is deleted and (when the warm-up
+ * gate clears) replaced by an honestly re-derived one, its `createdAt`
+ * sits after the cutoff and the account drops off the discovery list for
+ * good. Best-effort: errors return through the result so worker boot never
+ * fails on a repair miss.
+ */
+export async function enqueueBootTimeCumulativePrRederivation(
+  startAfterSeconds: number = 0,
+): Promise<{ enqueued: number; skipped: number; error: string | null }> {
+  const boss = getGlobalBoss();
+  if (!boss) {
+    return { enqueued: 0, skipped: 0, error: null };
+  }
+
+  try {
+    const affected = await prisma.personalRecord.findMany({
+      where: {
+        metricSlot: null,
+        metricType: { in: CUMULATIVE_TYPES_ARRAY },
+        createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF },
+      },
+      distinct: ["userId"],
+      select: { userId: true },
+    });
+
+    if (affected.length === 0) {
+      return { enqueued: 0, skipped: 0, error: null };
+    }
+
+    let enqueued = 0;
+    let skipped = 0;
+    for (const { userId } of affected) {
+      const payload: CumulativePrRederivePayload = {
+        userId,
+        enqueuedAt: new Date().toISOString(),
+      };
+      const jobId = await boss.send(CUMULATIVE_PR_REDERIVE_QUEUE, payload, {
+        retryLimit: 3,
+        retryDelay: 60,
+        retryBackoff: true,
+        // Coalesce: if a repair for this user is already queued and we
+        // restart, pg-boss returns null instead of duplicating.
+        singletonKey: `cumulative-pr-rederive|${userId}`,
+        ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
+      });
+      if (jobId) {
+        enqueued += 1;
+      } else {
+        skipped += 1;
+      }
+    }
+    return { enqueued, skipped, error: null };
+  } catch (err) {
+    return {
+      enqueued: 0,
+      skipped: 0,
+      error: err instanceof Error ? err.message : String(err),
+    };
+  }
+}