Merge remote-tracking branch 'origin/main' into fix/v1397-derive

src/lib/analytics/resting-pulse.ts

@@ -139,6 +139,26 @@ export function resolveRestingPulseSeries(input: {
   pulseSamples: ReadonlyArray<PulseSample>;
   /** Day key for the proxy buckets: the user's own day. */
   dayKeyOf: (d: Date) => string;
+}): { series: PulseSample[]; which: "resting" | "proxy" | "none" } {
+  return mergeRestingWithProxy({
+    restingSamples: input.restingSamples,
+    proxySamples: deriveRestingProxyFromPulse(
+      input.pulseSamples,
+      input.dayKeyOf,
+    ),
+    dayKeyOf: input.dayKeyOf,
+  });
+}
+
+/**
+ * {@link resolveRestingPulseSeries} for a caller that already holds the
+ * proxy, one point per day, as `readRestingPulseProxy` folds it in Postgres.
+ * The per-day merge and the label are the same.
+ */
+export function mergeRestingWithProxy(input: {
+  restingSamples: ReadonlyArray<PulseSample>;
+  proxySamples: ReadonlyArray<PulseSample>;
+  dayKeyOf: (d: Date) => string;
 }): { series: PulseSample[]; which: "resting" | "proxy" | "none" } {
   const dayKeyOf = input.dayKeyOf;
   const byTime = (a: PulseSample, b: PulseSample) =>
@@ -148,14 +168,13 @@ export function resolveRestingPulseSeries(input: {
   // Days already covered by a real resting row — the proxy must not
   // second-guess them.
   const restingDays = new Set(resting.map((s) => dayKeyOf(s.measuredAt)));
-  const proxy = deriveRestingProxyFromPulse(
-    input.pulseSamples,
-    dayKeyOf,
-  ).filter((p) => !restingDays.has(dayKeyOf(p.measuredAt)));
+  const proxy = input.proxySamples.filter(
+    (p) => !restingDays.has(dayKeyOf(p.measuredAt)),
+  );
 
   if (resting.length === 0) {
     return proxy.length > 0
-      ? { series: proxy, which: "proxy" }
+      ? { series: [...proxy], which: "proxy" }
       : { series: [], which: "none" };
   }
   if (proxy.length === 0) return { series: resting, which: "resting" };

src/lib/crypto/encryption-corpus.ts

@@ -390,19 +390,29 @@ export interface RotationResult {
   dropped: number;
 }
 
-/** Re-encrypt one column's stale rows to the active key. */
+/** Thrown out of a column walk when the caller's time budget is spent. */
+class RotationStopped extends Error {}
+
+/**
+ * Re-encrypt one column's stale rows to the active key. `shouldStop` is the
+ * caller's time budget, checked between rows; a stopped column reports
+ * `stopped` and the rows it did not reach stay on their old key for the next
+ * run, which skips every row already rotated.
+ */
 export async function rotateColumn(
   client: CorpusClient,
   col: EncryptedColumn,
-): Promise<RotationResult> {
+  shouldStop: () => boolean = () => false,
+): Promise<RotationResult & { stopped: boolean }> {
   const delegate = getDelegate(client, col.model);
-  const result: RotationResult = {
+  const result: RotationResult & { stopped: boolean } = {
     model: col.model,
     field: col.field,
     scanned: 0,
     rotated: 0,
     errors: 0,
     dropped: 0,
+    stopped: false,
   };
 
   /**
@@ -432,6 +442,10 @@ export async function rotateColumn(
     // (`convertSingleValueBackup`). A row converted, replaced or deleted since
     // it was listed is gone, not an error.
     for await (const id of singleValueIds(delegate)) {
+      if (shouldStop()) {
+        result.stopped = true;
+        break;
+      }
       result.scanned += 1;
       try {
         const outcome = await convertSingleValueBackup(
@@ -452,31 +466,37 @@ export async function rotateColumn(
   // Idempotent — rows already on the active key are skipped, so an
   // interrupted run resumes cleanly on the next invocation. `scanned` counts
   // the rows that hold ciphertext, whatever the column.
-  await walkColumn(delegate, col, async ({ id, value, codec }) => {
-    result.scanned += 1;
-    if (walkedKeyId(value, codec, col.kind) === getActiveKeyId()) return;
-    try {
-      // The plaintext is never inspected, only re-sealed, so every envelope
-      // a stored value can legitimately carry survives rotation untouched.
-      const next =
-        codec !== null
-          ? reencryptBlob(value as Uint8Array, codec, col.aad)
-          : fromCiphertext(
-              encrypt(decrypt(toCiphertext(value, col.kind)!)),
-              col.kind,
-            );
-      await delegate.update({
-        where: { [pkField(col)]: id },
-        data: { [col.field]: next },
-      });
-      result.rotated += 1;
-    } catch (err) {
-      // Deleted between the read and the write (a backup replacing its
-      // pieces, an account going): nothing is left to rotate.
-      if (isRowGone(err)) return;
-      await onUnreadable(id);
-    }
-  });
+  try {
+    await walkColumn(delegate, col, async ({ id, value, codec }) => {
+      if (shouldStop()) throw new RotationStopped();
+      result.scanned += 1;
+      if (walkedKeyId(value, codec, col.kind) === getActiveKeyId()) return;
+      try {
+        // The plaintext is never inspected, only re-sealed, so every envelope
+        // a stored value can legitimately carry survives rotation untouched.
+        const next =
+          codec !== null
+            ? reencryptBlob(value as Uint8Array, codec, col.aad)
+            : fromCiphertext(
+                encrypt(decrypt(toCiphertext(value, col.kind)!)),
+                col.kind,
+              );
+        await delegate.update({
+          where: { [pkField(col)]: id },
+          data: { [col.field]: next },
+        });
+        result.rotated += 1;
+      } catch (err) {
+        // Deleted between the read and the write (a backup replacing its
+        // pieces, an account going): nothing is left to rotate.
+        if (isRowGone(err)) return;
+        await onUnreadable(id);
+      }
+    });
+  } catch (err) {
+    if (!(err instanceof RotationStopped)) throw err;
+    result.stopped = true;
+  }
   return result;
 }
 
@@ -488,16 +508,31 @@ export interface CorpusRotation {
   totalErrors: number;
   /** Unreadable rows deleted from `disposable` columns. */
   totalDropped: number;
+  /**
+   * The time budget ran out before every column was walked. Running again
+   * finishes the job: rows already on the active key are skipped.
+   */
+  stoppedEarly: boolean;
 }
 
-/** Re-encrypt the whole corpus to the active key. Idempotent + active-key-only. */
+/**
+ * Re-encrypt the whole corpus to the active key. Idempotent + active-key-only.
+ * `shouldStop` is the caller's time budget (`jobBudget` in the worker).
+ */
 export async function rotateCorpus(
   client: CorpusClient,
+  shouldStop: () => boolean = () => false,
 ): Promise<CorpusRotation> {
   const activeKeyId = getActiveKeyId();
   const results: RotationResult[] = [];
+  let stoppedEarly = false;
   for (const col of ENCRYPTED_COLUMNS) {
-    results.push(await rotateColumn(client, col));
+    const { stopped, ...result } = await rotateColumn(client, col, shouldStop);
+    results.push(result);
+    if (stopped) {
+      stoppedEarly = true;
+      break;
+    }
   }
   let totalScanned = 0;
   let totalRotated = 0;
@@ -516,5 +551,6 @@ export async function rotateCorpus(
     totalRotated,
     totalErrors,
     totalDropped,
+    stoppedEarly,
   };
 }

src/lib/doctor-report/__tests__/leaf-gating-sweep.test.ts

@@ -49,6 +49,7 @@ const { calls, rows, prismaMock } = vi.hoisted(() => {
       return Promise.resolve(rows[name] ?? []);
     },
     count: () => Promise.resolve(0),
+    groupBy: () => Promise.resolve([]),
     findUnique: () => {
       calls.push(name);
       return Promise.resolve(rows[`${name}Unique`]?.[0] ?? null);

src/lib/doctor-report/collect.ts

@@ -77,8 +77,21 @@ import {
 const DENSE_REPORT_RAW_WINDOW_DAYS = 90;
 
 /**
- * #1023 — inside the raw window, pulse takes the per-day path anyway once the
- * window holds more readings than this.
+ * The types a device writes at sampling rate whose report sections are
+ * statistics and a series and nothing else: the figures the day path keeps
+ * exactly (count, mean, minimum, maximum, latest) plus one point per local
+ * day. Glucose is dense too but has its own rule below, because its clinical
+ * panel needs the per-reading counts the bucket query carries.
+ */
+const DENSE_STAT_TYPES = [
+  "PULSE",
+  "HEART_RATE_VARIABILITY",
+  "OXYGEN_SATURATION",
+] as const satisfies readonly MeasurementType[];
+
+/**
+ * #1023 — inside the raw window, a dense statistics type takes the per-day
+ * path anyway once the window holds more of its readings than this.
  *
  * A watch that records heart rate once a minute puts 130 000 readings in a
  * 90-day report. Read raw, every one of them became an object, went through
@@ -170,22 +183,38 @@ export async function collectDoctorReportData(
   const lastDay = new Date(`${userDayKey(end, reportTz)}T00:00:00.000Z`);
 
   const aggregateDenseTypes = days > DENSE_REPORT_RAW_WINDOW_DAYS;
-  const densePulse =
-    !excluded.includes("PULSE") &&
-    (aggregateDenseTypes ||
-      (await prisma.measurement.count({
-        where: {
-          userId,
-          type: "PULSE",
-          measuredAt: { gte: start, lte: end },
-          deletedAt: null,
-        },
-      })) > DENSE_REPORT_RAW_ROW_CAP);
+  const includedDenseStatTypes = DENSE_STAT_TYPES.filter(
+    (type) => !excluded.includes(type),
+  );
+  // Inside the raw window, one grouped count decides each type on its own:
+  // a watch's minute-by-minute heart rate folds while a few blood-oxygen
+  // readings a night beside it stay raw.
+  const denseStatCounts =
+    aggregateDenseTypes || includedDenseStatTypes.length === 0
+      ? null
+      : new Map(
+          (
+            await prisma.measurement.groupBy({
+              by: ["type"],
+              where: {
+                userId,
+                type: { in: [...includedDenseStatTypes] },
+                measuredAt: { gte: start, lte: end },
+                deletedAt: null,
+              },
+              _count: { _all: true },
+            })
+          ).map((row) => [row.type, row._count._all]),
+        );
+  const denseTypes: MeasurementType[] = includedDenseStatTypes.filter(
+    (type) =>
+      denseStatCounts === null ||
+      (denseStatCounts.get(type) ?? 0) > DENSE_REPORT_RAW_ROW_CAP,
+  );
   const denseGlucose =
     aggregateDenseTypes && !excluded.includes("BLOOD_GLUCOSE");
-  const rawExcluded = [...excluded];
-  if (densePulse) rawExcluded.push("PULSE");
-  if (denseGlucose) rawExcluded.push("BLOOD_GLUCOSE");
+  if (denseGlucose) denseTypes.push("BLOOD_GLUCOSE");
+  const rawExcluded = [...excluded, ...denseTypes];
 
   const wantsMedications = MEDICATION_LEAVES.some((leaf) => gate.admits(leaf));
   const wantsIntakeEvents =
@@ -318,8 +347,7 @@ export async function collectDoctorReportData(
         start,
         end,
         reportTz,
-        densePulse,
-        denseGlucose,
+        denseTypes,
       }),
     ]);
 

src/lib/doctor-report/dense-buckets.ts

@@ -1,16 +1,17 @@
 /**
- * The per-local-day pre-aggregation for the two sample-dense measurement
- * types.
+ * The per-local-day pre-aggregation for the sample-dense measurement types.
  *
  * On a long window a CGM or a per-sample heart-rate account produces six
- * figures of rows, so pulse and glucose collapse to one row per local
- * day/source/device/context in SQL and the canonical-source picker then works
- * over that bounded intermediate set. Every other type keeps the raw-row path.
+ * figures of rows, so the dense types (glucose, pulse, heart-rate variability,
+ * blood oxygen) collapse to one row per local day/source/device/context in
+ * SQL and the canonical-source picker then works over that bounded
+ * intermediate set. Every other type keeps the raw-row path.
  *
  * Extracted from the aggregator with the selection rework: the query is a job
  * of its own, it is the one raw-SQL surface on this path, and keeping it in a
  * thousand-line function is how a parameter stops being reviewed.
  */
+import { Prisma } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
 import type {
   GlucoseContext,
@@ -46,8 +47,8 @@ export interface DenseMeasurementBucket {
 }
 
 /**
- * Load the dense buckets for whichever of the two types the caller wants. Both
- * flags false is not a call: the caller skips it entirely.
+ * Load the dense buckets for the types the caller names. An empty list is not
+ * a call: nothing is read.
  *
  * Every value is bound as a tagged-template parameter — no splice, no
  * interpolation of anything that came off the wire.
@@ -57,11 +58,13 @@ export async function loadDenseMeasurementBuckets(params: {
   start: Date;
   end: Date;
   reportTz: string;
-  densePulse: boolean;
-  denseGlucose: boolean;
+  denseTypes: readonly MeasurementType[];
 }): Promise<DenseMeasurementBucket[]> {
-  const { userId, start, end, reportTz, densePulse, denseGlucose } = params;
-  if (!densePulse && !denseGlucose) return [];
+  const { userId, start, end, reportTz, denseTypes } = params;
+  if (denseTypes.length === 0) return [];
+  const typeList = Prisma.join(
+    denseTypes.map((type) => Prisma.sql`${type}::"measurement_type"`),
+  );
   return prisma.$queryRaw<DenseMeasurementBucket[]>`
         WITH localized AS (
           SELECT
@@ -81,11 +84,7 @@ export async function loadDenseMeasurementBuckets(params: {
             AND m."measured_at" >= ${start}
             AND m."measured_at" <= ${end}
             AND m."deleted_at" IS NULL
-            AND (
-              (${densePulse} AND m."type" = 'PULSE'::"measurement_type")
-              OR
-              (${denseGlucose} AND m."type" = 'BLOOD_GLUCOSE'::"measurement_type")
-            )
+            AND m."type" IN (${typeList})
         ),
         scored AS (
           SELECT

src/lib/jobs/__tests__/insight-pregenerate.test.ts

@@ -207,6 +207,31 @@ describe("findPregenerateCandidates", () => {
   });
 });
 
+describe("runInsightPregenerate — the job's time budget", () => {
+  it("stops between candidates once the budget is spent and says so", async () => {
+    capabilities = {
+      briefing: unavailable("user_disabled"),
+      statusText: unavailable("user_disabled"),
+    };
+    const { prisma } = makePrisma([
+      { id: "u1", locale: "de" },
+      { id: "u2", locale: "de" },
+      { id: "u3", locale: "de" },
+    ]);
+    let checks = 0;
+    const result = await runInsightPregenerate(prisma as never, {
+      generate: vi.fn(),
+      statusGenerators: [],
+      warmGenericMetrics: vi.fn().mockResolvedValue(0),
+      // Spent after the first candidate.
+      shouldStop: () => checks++ >= 1,
+    });
+
+    expect(result).toMatchObject({ total: 3, skipped: 1, stoppedEarly: true });
+    expect(aiCapabilityForJob).toHaveBeenCalledTimes(2);
+  });
+});
+
 describe("runInsightPregenerate — per-user capabilities", () => {
   it("skips a user with neither capability before any budget write or generation", async () => {
     capabilities = {
@@ -1337,7 +1362,7 @@ describe("queue registration", () => {
 
   it("schedules the cron in the schedules table (with retry policy)", () => {
     expect(workerSrc).toMatch(
-      /\[\s*INSIGHT_PREGENERATE_QUEUE\s*,\s*INSIGHT_PREGENERATE_CRON\s*,\s*insightRetryOptions\s*\]/,
+      /\[\s*INSIGHT_PREGENERATE_QUEUE\s*,\s*INSIGHT_PREGENERATE_CRON\s*,\s*\{\s*\.\.\.insightRetryOptions\s*,\s*expireInSeconds:\s*INSIGHT_PREGENERATE_EXPIRE_SECONDS\s*,?\s*\}\s*,?\s*\]/,
     );
   });
 

src/lib/jobs/__tests__/long-pass.test.ts

@@ -0,0 +1,74 @@
+/**
+ * `lockedPass`: each job runs under the lock `job:<queue>:<identity>`, and a
+ * delivery that finds the lock held does no work and succeeds.
+ *
+ * The lock itself (a session advisory lock on a connection of its own) is
+ * exercised against Postgres in
+ * `tests/integration/integration-backfill-lane-expiry.test.ts`; this pins the
+ * wrapper around it.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { Job } from "pg-boss";
+
+const held = vi.hoisted(() => new Set<string>());
+const keys = vi.hoisted(() => [] as string[]);
+
+vi.mock("@/lib/jobs/job-lock", () => ({
+  withJobLock: async <T>(key: string, run: () => Promise<T>) => {
+    keys.push(key);
+    if (held.has(key)) return { ran: false };
+    held.add(key);
+    try {
+      return { ran: true, result: await run() };
+    } finally {
+      held.delete(key);
+    }
+  },
+}));
+
+import { jobDone, jobFailed } from "@/lib/jobs/job-outcome";
+import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+
+type Payload = { userId?: string };
+const job = (data: Payload) => ({ id: "j", data }) as Job<Payload>;
+
+beforeEach(() => {
+  held.clear();
+  keys.length = 0;
+});
+
+describe("lockedPass", () => {
+  it("runs each job under its queue and identity", async () => {
+    const handler = vi.fn(async () => jobDone({ generated: 1 }));
+    const bound = lockedPass(
+      "q",
+      (j: Job<Payload>) =>
+        j.data.userId ? `user:${j.data.userId}` : WHOLE_PASS,
+      handler,
+    );
+    const outcome = await bound([job({ userId: "a" })]);
+    expect(outcome).toEqual(jobDone({ generated: 1 }));
+    await bound([job({})]);
+    expect(keys).toEqual(["job:q:user:a", "job:q:pass"]);
+  });
+
+  it("does nothing beside a run that still holds the lock", async () => {
+    held.add("job:q:pass");
+    const handler = vi.fn(async () => jobDone());
+    const outcome = await lockedPass("q", () => WHOLE_PASS, handler)([job({})]);
+    expect(handler).not.toHaveBeenCalled();
+    expect(outcome).toEqual(jobDone({ already_running: 1 }));
+  });
+
+  it("passes a failed outcome through", async () => {
+    const failed = jobFailed("stopped early", undefined, {
+      stopped_early: true,
+    });
+    const outcome = await lockedPass(
+      "q",
+      () => WHOLE_PASS,
+      async () => failed,
+    )([job({})]);
+    expect(outcome).toBe(failed);
+  });
+});

src/lib/jobs/__tests__/period-narrative-warm.test.ts

@@ -250,7 +250,7 @@ describe("queue registration", () => {
 
   it("schedules the cron in the schedules table (with retry policy)", () => {
     expect(workerSrc).toMatch(
-      /\[\s*PERIOD_NARRATIVE_QUEUE\s*,\s*PERIOD_NARRATIVE_CRON\s*,\s*insightRetryOptions\s*\]/,
+      /\[\s*PERIOD_NARRATIVE_QUEUE\s*,\s*PERIOD_NARRATIVE_CRON\s*,\s*\{\s*\.\.\.insightRetryOptions\s*,\s*expireInSeconds:\s*PERIOD_NARRATIVE_EXPIRE_SECONDS\s*,?\s*\}\s*,?\s*\]/,
     );
   });
 

src/lib/jobs/__tests__/status-cron-candidates.test.ts

@@ -190,9 +190,9 @@ describe("mood-status queue registration (dead-queue guard)", () => {
     expect(match![1]).toMatch(/\bMOOD_STATUS_QUEUE\b/);
   });
 
-  it("schedules the 02:30 cron with the insight retry policy", () => {
+  it("schedules the 02:30 cron with the nightly pass options (retries and expiry)", () => {
     expect(workerSrc).toMatch(
-      /\[\s*MOOD_STATUS_QUEUE\s*,\s*MOOD_STATUS_CRON\s*,\s*insightRetryOptions\s*\]/,
+      /\[\s*MOOD_STATUS_QUEUE\s*,\s*MOOD_STATUS_CRON\s*,\s*nightlyInsightPassOptions\s*\]/,
     );
     expect(workerSrc).toMatch(/MOOD_STATUS_CRON\s*=\s*"30 2 \* \* \*"/);
   });

src/lib/jobs/backup-restore.ts

@@ -811,7 +811,12 @@ export async function enqueueBackupRestoreSweepAtBoot(
     await boss.send(
       BACKUP_RESTORE_QUEUE,
       { sweep: true } satisfies BackupRestorePayload,
-      { startAfter: Math.ceil(BACKUP_RESTORE_STALE_AFTER_MS / 1000) + 15 },
+      {
+        startAfter: Math.ceil(BACKUP_RESTORE_STALE_AFTER_MS / 1000) + 15,
+        // The sweep is short, but it rides a queue whose every send carries
+        // the restore's expiry, so no job on it is cut at the default.
+        expireInSeconds: BACKUP_RESTORE_EXPIRE_SECONDS,
+      },
     );
   } catch {
     // The admission check fails an abandoned job the next time the account

src/lib/jobs/document-content-index-backfill.ts

@@ -64,6 +64,14 @@ export const CONTENT_INDEX_BACKFILL_CONCURRENCY = 1;
 /** Discovery page size (id-only; blobs are loaded one at a time). */
 const PAGE_SIZE = 50;
 
+/**
+ * The job's expiry: three hours. One job transcribes up to 200 documents with
+ * a provider call each, which outlasted pg-boss's fifteen-minute default and
+ * was retried beside itself. The run stops at three quarters of this
+ * (`jobBudget`) and holds a lock per account (`lockedPass`).
+ */
+export const CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS = 3 * 60 * 60;
+
 /** Max documents one job indexes before yielding (bounds spend + runtime). */
 const MAX_DOCS_PER_RUN = 200;
 
@@ -97,6 +105,11 @@ export interface ContentIndexBackfillSummary {
   failed: number;
   reason:
     "ok" | "no-provider" | "no-consent" | "budget-reached" | "unavailable";
+  /**
+   * The job's time budget ran out before the walk did. The documents not
+   * reached stay un-indexed for the next run.
+   */
+  stoppedEarly?: boolean;
 }
 
 /**
@@ -158,6 +171,11 @@ async function retokeniseStaleRows(
  */
 export async function runContentIndexBackfillForUser(
   userId: string,
+  /**
+   * The job's time budget (`jobBudget`), checked before each document: up to
+   * 200 transcriptions, one provider call each, can outlast the job.
+   */
+  shouldStop: () => boolean = () => false,
 ): Promise<ContentIndexBackfillSummary> {
   // Stale-tokeniser rows first: local work that owes nothing to the
   // provider, consent, or budget gates below, so a user without a usable
@@ -222,6 +240,7 @@ export async function runContentIndexBackfillForUser(
   let failed = stale.failed;
   let reason: ContentIndexBackfillSummary["reason"] = "ok";
   let cursor: string | null = null;
+  let stoppedEarly = false;
 
   outer: for (;;) {
     if (indexed >= MAX_DOCS_PER_RUN) break;
@@ -244,6 +263,10 @@ export async function runContentIndexBackfillForUser(
 
     for (const { id } of rows) {
       if (indexed >= MAX_DOCS_PER_RUN) break outer;
+      if (shouldStop()) {
+        stoppedEarly = true;
+        break outer;
+      }
 
       const dateKey = buildDateKey();
       const reservation = await reserveBudget(
@@ -325,9 +348,23 @@ export async function runContentIndexBackfillForUser(
 
   annotate({
     action: { name: "documents.contentIndex.backfill" },
-    meta: { indexed, retokenised: stale.retokenised, skipped, failed, reason },
+    meta: {
+      indexed,
+      retokenised: stale.retokenised,
+      skipped,
+      failed,
+      reason,
+      stopped_early: stoppedEarly,
+    },
   });
-  return { indexed, retokenised: stale.retokenised, skipped, failed, reason };
+  return {
+    indexed,
+    retokenised: stale.retokenised,
+    skipped,
+    failed,
+    reason,
+    ...(stoppedEarly ? { stoppedEarly } : {}),
+  };
 }
 
 /**
@@ -344,6 +381,7 @@ export async function enqueueContentIndexBackfill(
     enqueuedAt: new Date().toISOString(),
   };
   const jobId = await boss.send(CONTENT_INDEX_BACKFILL_QUEUE, payload, {
+    expireInSeconds: CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS,
     retryLimit: 3,
     retryDelay: 60,
     retryBackoff: true,

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

@@ -25,6 +25,7 @@
 import { type Job } from "pg-boss";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { auditLog } from "@/lib/auth/audit";
 import {
   rotateCorpus,
@@ -37,27 +38,40 @@ export const ENCRYPTION_KEY_ROTATE_CONCURRENCY = 1;
 /** Fixed key so duplicate admin triggers coalesce into one queued run. */
 export const ENCRYPTION_KEY_ROTATE_SINGLETON = "encryption-key-rotate";
 
+/**
+ * The job's expiry: twelve hours. The pass re-encrypts every encrypted column
+ * on the instance and outlasted pg-boss's fifteen-minute default, which
+ * retried it beside itself. The pass stops at three quarters of this
+ * (`jobBudget`) and holds a lock (`lockedPass`); the admin route's send
+ * carries it.
+ */
+export const ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS = 12 * 60 * 60;
+
 export interface EncryptionKeyRotatePayload {
   /** The admin who triggered the run (for the audit trail). */
   requestedByUserId?: string;
   enqueuedAt?: string;
 }
 
-export async function runEncryptionKeyRotation(): Promise<{
+export async function runEncryptionKeyRotation(
+  shouldStop: () => boolean = () => false,
+): Promise<{
   activeKeyId: string;
   totalScanned: number;
   totalRotated: number;
   totalErrors: number;
   totalDropped: number;
+  stoppedEarly: boolean;
 }> {
   const prisma = getWorkerPrisma();
-  const out = await rotateCorpus(prisma as unknown as CorpusClient);
+  const out = await rotateCorpus(prisma as unknown as CorpusClient, shouldStop);
   return {
     activeKeyId: out.activeKeyId,
     totalScanned: out.totalScanned,
     totalRotated: out.totalRotated,
     totalErrors: out.totalErrors,
     totalDropped: out.totalDropped,
+    stoppedEarly: out.stoppedEarly,
   };
 }
 
@@ -67,7 +81,10 @@ export async function handleEncryptionKeyRotate(
   return withBackgroundEvent("job.encryption_key_rotate", async (evt) => {
     const requestedBy = jobs[0]?.data?.requestedByUserId ?? null;
     try {
-      const result = await runEncryptionKeyRotation();
+      // A whole-corpus pass: it stops between rows once the job's budget is
+      // spent, and says so in the audit row, so the operator runs it again
+      // to finish (rows already rotated are skipped).
+      const result = await runEncryptionKeyRotation(jobBudget(jobs));
       evt.addMeta("rotate_active_key_id", result.activeKeyId);
       evt.addMeta("rotate_scanned", result.totalScanned);
       evt.addMeta("rotate_rotated", result.totalRotated);

src/lib/jobs/insight-pregenerate-shared.ts

@@ -19,6 +19,15 @@ import { annotate } from "@/lib/logging/context";
 
 export const INSIGHT_PREGENERATE_QUEUE = "insight-pregenerate";
 
+/**
+ * The job's expiry: three hours. The scheduled pass walks up to 200 accounts
+ * with a comprehensive generation each, which can outlast pg-boss's fifteen
+ * minute default; past it pg-boss retried the job beside the pass still
+ * running. The pass now stops itself at three quarters of this (`jobBudget`)
+ * and holds a lock (`lockedPass`), and every send and the schedule carry it.
+ */
+export const INSIGHT_PREGENERATE_EXPIRE_SECONDS = 3 * 60 * 60;
+
 export interface InsightPregeneratePayload {
   /** Set on the nightly scheduled tick (informational; the handler keys
    * its batch behaviour off the ABSENCE of `userId`/`force`). */
@@ -73,6 +82,7 @@ export async function enqueueForceWarm(payload: {
       } satisfies InsightPregeneratePayload,
       {
         singletonKey: `force:${payload.userId}`,
+        expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
         // De-dupe within a 6-minute window. v1.16.8 — widened from 120 s:
         // the client's revalidation poll runs up to ~250 s after a stale
         // page-open, so a 2-minute singleton let the tail of one poll
@@ -148,6 +158,7 @@ export async function enqueuePregenerateFailureRetry(payload: {
         // warm can never swallow the scheduled retry (and vice versa);
         // one retry per user per window.
         singletonKey: `retry:${payload.userId}`,
+        expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
         singletonSeconds: 3600,
         // A worker crash mid-warm re-runs the job; a warm that completes
         // with a failed comprehensive does NOT throw, so this cannot loop.

src/lib/jobs/insight-pregenerate.ts

@@ -277,6 +277,8 @@ export interface PregenerateRunResult {
   skipped: number;
   failed: number;
   budgetBlocked: number;
+  /** The job's time budget ran out before every candidate was reached. */
+  stoppedEarly: boolean;
   /**
    * Count of per-metric assessment caches written warm across the whole
    * run. The comprehensive generator evicts the seven `*-status`
@@ -575,6 +577,12 @@ export async function runInsightPregenerate(
       userId: string;
       locale: SupportedLocale;
     }) => Promise<void>;
+    /**
+     * The job's time budget (`jobBudget`). Checked between candidates: a pass
+     * of 200 provider calls can outlast its job, and the candidates it does
+     * not reach are the stalest again tomorrow, so they lead the next pass.
+     */
+    shouldStop?: () => boolean;
   } = {},
 ): Promise<PregenerateRunResult> {
   const now = options.now ?? new Date();
@@ -606,6 +614,7 @@ export async function runInsightPregenerate(
     skipped: 0,
     failed: 0,
     budgetBlocked: 0,
+    stoppedEarly: false,
     assessmentsWarmed: 0,
     metricAssessmentsWarmed: 0,
   };
@@ -619,6 +628,10 @@ export async function runInsightPregenerate(
   result.total = candidates.length;
 
   for (const candidate of candidates) {
+    if (options.shouldStop?.()) {
+      result.stoppedEarly = true;
+      break;
+    }
     // v1.18.11 P3 — one feature-cache scope per candidate, so the
     // comprehensive briefing's bounded feature read is computed once and any
     // sibling consumer in this candidate's warm reuses it. Scoped per user so
@@ -819,6 +832,7 @@ export async function runInsightPregenerate(
       skipped: result.skipped,
       failed: result.failed,
       budget_blocked: result.budgetBlocked,
+      stopped_early: result.stoppedEarly,
       assessments_warmed: result.assessmentsWarmed,
       metric_assessments_warmed: result.metricAssessmentsWarmed,
     },

src/lib/jobs/integration-backfill-lock.ts

@@ -11,8 +11,9 @@
  * started beside it in the same worker, so the lane's "one at a time" did not
  * hold for the one case it exists for, an import too long for its expiry.
  *
- * So each run holds a session-level advisory lock keyed on the job's identity
- * (kind, provider where there is one, account) for as long as it runs. A
+ * So each run holds a session-level advisory lock (`withJobLock`) keyed on the
+ * job's identity (kind, provider where there is one, account) for as long as
+ * it runs. A
  * delivery that finds the lock taken does no work: the run that holds it is
  * still importing, and it either stamps its completion marker or leaves the
  * account for the next boot's discovery. The lock lives on a connection of its
@@ -24,47 +25,26 @@
  * One extra connection per running import; the lane admits one import per
  * process at a time, so that is one connection.
  */
-import { Client } from "pg";
+import { withJobLock, type GuardedRun } from "./job-lock";
 
 /** The lock key for one admission job's identity. */
 function integrationBackfillLockKey(identity: string): string {
   return `integration-backfill:${identity}`;
 }
 
-/** What a guarded run came to. */
-type GuardedRun<T> = { ran: true; result: T } | { ran: false };
-
 /**
  * Run `run` while holding the import lock for `identity`. Resolves
  * `{ ran: false }` at once, without calling `run`, when another run holds it.
  * Rejects exactly as `run` does otherwise.
  */
-export async function withIntegrationBackfillLock<T>(
+export function withIntegrationBackfillLock<T>(
   identity: string,
   run: () => Promise<T>,
   connectionString: string | undefined = process.env.DATABASE_URL,
 ): Promise<GuardedRun<T>> {
-  const client = new Client({ connectionString, keepAlive: true });
-  // An idle client whose server goes away emits `error`; unhandled, that
-  // takes the whole worker down. The run itself notices the database is
-  // gone through its own queries.
-  client.on("error", () => {});
-  await client.connect();
-  try {
-    const key = integrationBackfillLockKey(identity);
-    const { rows } = await client.query<{ held: boolean }>(
-      "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
-      [key],
-    );
-    if (!rows[0]?.held) return { ran: false };
-    try {
-      return { ran: true, result: await run() };
-    } finally {
-      await client
-        .query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key])
-        .catch(() => {});
-    }
-  } finally {
-    await client.end().catch(() => {});
-  }
+  return withJobLock(
+    integrationBackfillLockKey(identity),
+    run,
+    connectionString,
+  );
 }

src/lib/jobs/job-lock.ts

@@ -0,0 +1,58 @@
+/**
+ * A session-level advisory lock held for the length of one run, across
+ * processes.
+ *
+ * pg-boss holds a job's "one at a time" only while the job is inside its
+ * expiry. Once the expiry passes it fails the job, fires its abort signal and
+ * retries it, and it does not stop the handler: measured against pg-boss 12.34
+ * (`tests/integration/integration-backfill-lane-expiry.test.ts`), the retry of
+ * the same job started beside the run that was still going. No queue policy
+ * helps, because the policies constrain job rows and the expired row is no
+ * longer active. Only something the running handler holds can, so each long
+ * run holds an advisory lock keyed on its identity, and a delivery that finds
+ * the lock taken does no work.
+ *
+ * The lock lives on a connection of its own, outside the Prisma pool, because
+ * a session lock has to stay on one connection for the whole run and a pooled
+ * one is handed back between statements. Ending the connection releases the
+ * lock, and so does the process dying, so a crash never leaves a key locked.
+ * One extra connection per running locked job.
+ */
+import { Client } from "pg";
+
+/** What a guarded run came to. */
+export type GuardedRun<T> = { ran: true; result: T } | { ran: false };
+
+/**
+ * Run `run` while holding the advisory lock for `key`. Resolves
+ * `{ ran: false }` at once, without calling `run`, when another run holds it.
+ * Rejects exactly as `run` does otherwise.
+ */
+export async function withJobLock<T>(
+  key: string,
+  run: () => Promise<T>,
+  connectionString: string | undefined = process.env.DATABASE_URL,
+): Promise<GuardedRun<T>> {
+  const client = new Client({ connectionString, keepAlive: true });
+  // An idle client whose server goes away emits `error`; unhandled, that
+  // takes the whole worker down. The run itself notices the database is
+  // gone through its own queries.
+  client.on("error", () => {});
+  await client.connect();
+  try {
+    const { rows } = await client.query<{ held: boolean }>(
+      "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
+      [key],
+    );
+    if (!rows[0]?.held) return { ran: false };
+    try {
+      return { ran: true, result: await run() };
+    } finally {
+      await client
+        .query("SELECT pg_advisory_unlock(hashtextextended($1, 0))", [key])
+        .catch(() => {});
+    }
+  } finally {
+    await client.end().catch(() => {});
+  }
+}

src/lib/jobs/long-pass.ts

@@ -0,0 +1,58 @@
+/**
+ * The binding for a queue whose job can outlast pg-boss's default expiry.
+ *
+ * Three things together make a long job safe, and `queue-runtime.ts` names
+ * every queue that needs them:
+ *
+ *   1. An explicit `expireInSeconds` on every send and schedule, so pg-boss
+ *      does not declare the job dead at fifteen minutes while it is working.
+ *   2. A time budget (`jobBudget` / `jobDeadline` in `job-budget.ts`) the
+ *      handler checks between units of work, so the pass stops on its own
+ *      before that expiry and the next run picks up what is left.
+ *   3. A lock the running handler holds (`withJobLock`), so the retry pg-boss
+ *      starts once the expiry has passed does nothing beside a run that is
+ *      still going. Queue policies cannot do this: they constrain job rows,
+ *      and an expired job's row is no longer active.
+ *
+ * `lockedPass` is the third. It wraps a handler so each job runs under the
+ * lock `<queue>:<identity>`, where the identity is what two runs must not
+ * share: the account for a per-user job, a constant for a pass over everyone.
+ * A delivery that finds the lock held reports `already_running` and succeeds,
+ * because the run that holds it is doing the work.
+ */
+import type { Job } from "pg-boss";
+
+import { withJobLock } from "@/lib/jobs/job-lock";
+import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { annotate } from "@/lib/logging/context";
+
+/** The identity of a pass that walks every account: one run at a time. */
+export const WHOLE_PASS = "pass";
+
+export function lockedPass<J extends Job<unknown>>(
+  queue: string,
+  identityOf: (job: J) => string,
+  handler: (jobs: J[]) => Promise<JobOutcome>,
+): (jobs: J[]) => Promise<JobOutcome> {
+  return async (jobs) => {
+    let outcome: JobOutcome = jobDone({ jobs: 0 });
+    for (const job of jobs) {
+      const run = await withJobLock(`job:${queue}:${identityOf(job)}`, () =>
+        handler([job]),
+      );
+      if (!run.ran) {
+        annotate({
+          action: { name: "job.long_pass.already_running" },
+          meta: { queue },
+        });
+        outcome = jobDone({ already_running: 1 });
+        continue;
+      }
+      // The first failed job fails the batch; pg-boss hands one job per
+      // batch here, so this is that job's own outcome.
+      if (!run.result.ok) return run.result;
+      outcome = run.result;
+    }
+    return outcome;
+  };
+}

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

@@ -4,7 +4,8 @@ import { Prisma, type PrismaClient } from "@/generated/prisma/client";
 import { invalidateUserMedications } from "@/lib/cache/invalidate";
 import { auditLog } from "@/lib/auth/audit";
 import { prisma, toJson } from "@/lib/db";
-import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { consumeImportedIntakesBatch } from "@/lib/medications/inventory/consumption";
 import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
 import {
@@ -825,13 +826,25 @@ async function processNextChunk(
   );
 }
 
-export async function processMedicationIntakeImportJob(
+/**
+ * Work the import chunk by chunk until it is finished. `shouldStop` is the
+ * job's time budget: between chunks, once it is spent, the run returns
+ * `"stopped"` and leaves the job row as it is. Every chunk commits its own
+ * progress under the row lock, so the retry that follows picks up at the next
+ * chunk, and a second delivery running beside this one waits on that lock and
+ * continues from the progress this one wrote.
+ */
+async function workMedicationIntakeImport(
   jobId: string,
-): Promise<MedicationImportResult | null> {
+  shouldStop: () => boolean,
+): Promise<MedicationImportResult | null | "stopped"> {
   const client = getWorkerPrisma();
   for (;;) {
     const outcome = await processNextChunk(client, jobId);
-    if (!outcome.terminal) continue;
+    if (!outcome.terminal) {
+      if (shouldStop()) return "stopped";
+      continue;
+    }
     if (outcome.finalized && outcome.userId && outcome.result) {
       if (outcome.result.imported > 0) {
         invalidateUserMedications(outcome.userId, { evict: true });
@@ -843,6 +856,14 @@ export async function processMedicationIntakeImportJob(
   }
 }
 
+/** The whole import in one go, with no time budget. */
+export async function processMedicationIntakeImportJob(
+  jobId: string,
+): Promise<MedicationImportResult | null> {
+  const result = await workMedicationIntakeImport(jobId, () => false);
+  return result === "stopped" ? null : result;
+}
+
 async function recordWorkerFailure(
   job: Job<MedicationIntakeImportQueuePayload>,
   error: unknown,
@@ -879,7 +900,19 @@ export async function handleMedicationIntakeImport(
   job: Job<MedicationIntakeImportQueuePayload>,
 ): Promise<JobOutcome> {
   try {
-    const result = await processMedicationIntakeImportJob(job.data.jobId);
+    const result = await workMedicationIntakeImport(
+      job.data.jobId,
+      jobBudget([job]),
+    );
+    // Out of time before the last chunk: fail the delivery so pg-boss
+    // retries it, and the retry resumes from the committed progress.
+    if (result === "stopped") {
+      return jobFailed(
+        "stopped before the last chunk; the retry resumes",
+        undefined,
+        { stopped_early: true },
+      );
+    }
     // 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.

src/lib/jobs/offhost-backup.ts

@@ -42,6 +42,15 @@ const ALGORITHM = "aes-256-gcm";
 const IV_LENGTH = 12;
 const TAG_LENGTH = 16;
 
+/**
+ * The off-host pass's expiry: four hours. It uploads every account and
+ * already stopped between accounts at three quarters of the default fifteen
+ * minutes, which on an instance of any size meant a retry per night; one
+ * large account past the default was retried beside itself. Four hours with
+ * the same budget and a lock (`lockedPass`) leaves room for the night's pass.
+ */
+export const OFFHOST_BACKUP_EXPIRE_SECONDS = 4 * 60 * 60;
+
 export interface OffhostBackupConfig {
   endpoint: string;
   bucket: string;

src/lib/jobs/period-narrative-shared.ts

@@ -23,6 +23,15 @@ import type { Locale } from "@/lib/i18n/config";
 
 export const PERIOD_NARRATIVE_QUEUE = "period-narrative-warm";
 
+/**
+ * The job's expiry: three hours. On a boundary night the scheduled pass
+ * writes up to two narratives for each of 200 accounts, which can outlast
+ * pg-boss's fifteen-minute default. The pass stops itself at three quarters
+ * of this (`jobBudget`) and holds a lock (`lockedPass`); every send and the
+ * schedule carry it.
+ */
+export const PERIOD_NARRATIVE_EXPIRE_SECONDS = 3 * 60 * 60;
+
 /**
  * Nightly at 05:05 Europe/Berlin — inside the existing maintenance window,
  * after the comprehensive insight pre-generation (04:30) and the computed
@@ -79,6 +88,7 @@ export async function enqueueNarrativeWarm(payload: {
       {
         singletonKey: `warm:${payload.userId}:${payload.period}:${payload.locale}`,
         singletonSeconds: 120,
+        expireInSeconds: PERIOD_NARRATIVE_EXPIRE_SECONDS,
         // Transient provider / pool failures retry with backoff instead
         // of leaving the narrative cold until the next boundary night.
         retryLimit: 3,

src/lib/jobs/period-narrative-warm.ts

@@ -63,6 +63,8 @@ export interface NarrativeWarmRunResult {
   insufficient: number;
   failed: number;
   budgetBlocked: number;
+  /** The job's time budget ran out before every candidate was reached. */
+  stoppedEarly: boolean;
 }
 
 /**
@@ -152,6 +154,12 @@ export async function runPeriodNarrativeWarm(
     cap?: number;
     /** Injected for the test — defaults to the real generator. */
     generate?: typeof generatePeriodNarrative;
+    /**
+     * The job's time budget (`jobBudget`), checked between candidates. The
+     * candidates are ordered least recently narrated first, so the ones a
+     * stopped pass did not reach lead the next one.
+     */
+    shouldStop?: () => boolean;
   } = {},
 ): Promise<NarrativeWarmRunResult> {
   const now = options.now ?? new Date();
@@ -167,6 +175,7 @@ export async function runPeriodNarrativeWarm(
     insufficient: 0,
     failed: 0,
     budgetBlocked: 0,
+    stoppedEarly: false,
   };
 
   const flags = await getAssistantFlags();
@@ -180,6 +189,10 @@ export async function runPeriodNarrativeWarm(
   result.total = candidates.length;
 
   for (const candidate of candidates) {
+    if (options.shouldStop?.()) {
+      result.stoppedEarly = true;
+      break;
+    }
     // The nightly pass exists for the model-written narrative. Unavailable,
     // it skips the user before the context build and the budget write; the
     // deterministic narrative is written on the next read instead (the
@@ -233,6 +246,7 @@ export async function runPeriodNarrativeWarm(
       insufficient: result.insufficient,
       failed: result.failed,
       budget_blocked: result.budgetBlocked,
+      stopped_early: result.stoppedEarly,
     },
   });
 

src/lib/jobs/pr-detection.ts

@@ -15,6 +15,16 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 
 export const PR_DETECTION_QUEUE = "pr-detection";
 
+/**
+ * The job's expiry: twenty-five minutes. The fallback cron walks every
+ * account's whole history every thirty minutes, which on a large instance
+ * outlasts pg-boss's fifteen-minute default; past it pg-boss retried the pass
+ * beside itself. The pass stops at three quarters of this (`jobBudget`) and
+ * holds a lock (`lockedPass`), so it ends before the next tick and a retry or
+ * a late tick never runs beside it. The ingest sends carry the same expiry.
+ */
+export const PR_DETECTION_EXPIRE_SECONDS = 25 * 60;
+
 /** Concurrency budget for the worker process — five jobs in flight
  *  is enough to drain a multi-user backfill spike without crowding
  *  the reminder check or the daily insights workload that runs on the
@@ -62,6 +72,7 @@ export async function enqueuePrDetection(
     triggeredAt: new Date().toISOString(),
   };
   await boss.send(PR_DETECTION_QUEUE, payload, {
+    expireInSeconds: PR_DETECTION_EXPIRE_SECONDS,
     // Stagger retries gently — the detector is purely read-heavy on
     // the user's own data, but a stampede of failed jobs against a
     // briefly-flaky DB would compound the problem.

src/lib/jobs/queue-runtime.ts

@@ -0,0 +1,398 @@
+/**
+ * How long one job of each queue can run, and what keeps a long one safe.
+ *
+ * pg-boss gives every job an expiry, fifteen minutes unless the send says
+ * otherwise. Past it pg-boss fails the job, fires its abort signal and retries
+ * it, and it does not stop the handler, so a job that legitimately needs longer
+ * ends up running twice beside itself. No queue policy prevents that: the
+ * policies constrain job rows, and an expired row is no longer active. Every
+ * queue bound with `createAndWork` is therefore declared here as one of:
+ *
+ *   - `short`: a job cannot come near fifteen minutes (one account's bounded
+ *     window, one document, a forwarder, a bounded delete). It relies on the
+ *     default expiry, and the reason says why that is enough.
+ *   - `budgeted`: a pass over a long history that stops itself at three
+ *     quarters of the default expiry (`jobBudget`) and hands the rest to its
+ *     next run, so it ends before fifteen minutes by construction.
+ *   - `long`: a job that may run for hours. It carries an explicit
+ *     `expireInSeconds` on every send and schedule, stops itself before that
+ *     expiry (`jobBudget` / `jobDeadline`), and holds something per identity
+ *     that keeps a second run from working beside it: the lock `lockedPass`
+ *     takes, a lock of its own, or a row it claims. The one exception is a
+ *     one-shot job with `retryLimit: 0`, which pg-boss never retries and which
+ *     cannot stop half-way without losing its input.
+ *
+ * `src/__tests__/job-queue-runtime-guard.test.ts` holds every binding, send
+ * and schedule in the tree to this table.
+ */
+import { INTEGRATION_BACKFILL_EXPIRE_SECONDS } from "@/lib/jobs/integration-backfill-admission";
+import { NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS } from "@/lib/jobs/reminder/registrar-shared";
+import { INSIGHT_PREGENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-pregenerate-shared";
+import { PERIOD_NARRATIVE_EXPIRE_SECONDS } from "@/lib/jobs/period-narrative-shared";
+import { PR_DETECTION_EXPIRE_SECONDS } from "@/lib/jobs/pr-detection";
+import { ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS } from "@/lib/jobs/encryption-key-rotate";
+import { CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS } from "@/lib/jobs/document-content-index-backfill";
+import { DATA_BACKUP_SEND_OPTIONS } from "@/lib/jobs/data-backup-policy";
+import { OFFHOST_BACKUP_EXPIRE_SECONDS } from "@/lib/jobs/offhost-backup";
+import { BACKUP_RESTORE_EXPIRE_SECONDS } from "@/lib/jobs/backup-restore";
+import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "@/lib/jobs/apple-health-import-worker";
+import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-intake-import";
+
+/** pg-boss's default job expiry, in seconds. */
+export const DEFAULT_JOB_EXPIRE_SECONDS = 15 * 60;
+
+/** Where a check lives: `binding` is the `createAndWork` call itself. */
+export type SourceSite = "binding" | { file: string; fn: string };
+
+export type QueueRuntime =
+  | { runtime: "short"; why: string }
+  | { runtime: "budgeted"; budget: SourceSite; why: string }
+  | {
+      runtime: "long";
+      expireInSeconds: number;
+      /**
+       * Identifiers that carry the expiry into a send or a schedule. A site
+       * that names none of them must set `expireInSeconds` itself.
+       */
+      expiryVia: readonly string[];
+      /** Where the handler stops on the job's budget, or `oneShot`. */
+      stop: SourceSite | { oneShot: string };
+      /**
+       * What keeps a second run from working beside this one: `lockedPass`
+       * at the binding, or a marker in the named function.
+       */
+      exclusive: "lockedPass" | { marker: string; at: SourceSite };
+      why: string;
+    };
+
+const short = (why: string): QueueRuntime => ({ runtime: "short", why });
+
+const nightlyStatusPass = (fn: string): QueueRuntime => ({
+  runtime: "long",
+  expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS,
+  expiryVia: ["nightlyInsightPassOptions"],
+  stop: { file: "lib/jobs/reminder/insights-handlers.ts", fn },
+  exclusive: "lockedPass",
+  why: "Nightly status pass: one provider call per candidate account, serially, over the whole instance.",
+});
+
+const cohortSync = (provider: string): QueueRuntime =>
+  short(
+    `${provider} sync: a recent window per connection (incremental since the last sync); a webhook job is one connection.`,
+  );
+
+const oauthStateCleanup = short("One DELETE of expired OAuth state rows.");
+
+const forwarder = short(
+  "Forwards each job to integration-backfill-admission; the import itself runs there.",
+);
+
+export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
+  // ── Integration sync ───────────────────────────────────────────────
+  "integration-backfill-admission": {
+    runtime: "long",
+    expireInSeconds: INTEGRATION_BACKFILL_EXPIRE_SECONDS,
+    expiryVia: ["integrationBackfillAdmissionSendOptions"],
+    stop: {
+      file: "lib/jobs/integration-backfill-drain.ts",
+      fn: "drainIntegrationBackfillAdmission",
+    },
+    exclusive: {
+      marker: "withIntegrationBackfillLock",
+      at: {
+        file: "lib/jobs/integration-backfill-drain.ts",
+        fn: "drainIntegrationBackfillAdmission",
+      },
+    },
+    why: "Full-history provider imports: thousands of pages on a dense account.",
+  },
+  "withings-fallback-sync": cohortSync("Withings"),
+  "withings-activity-sync": cohortSync("Withings activity"),
+  "withings-sleep-sync": cohortSync("Withings sleep"),
+  "withings-ecg-sync": cohortSync("Withings ECG"),
+  "withings-oauth-state-cleanup": oauthStateCleanup,
+  "whoop-recovery-sync": cohortSync("WHOOP recovery"),
+  "whoop-sleep-sync": cohortSync("WHOOP sleep"),
+  "whoop-workout-sync": cohortSync("WHOOP workout"),
+  "whoop-cycle-sync": cohortSync("WHOOP cycle"),
+  "whoop-backfill": forwarder,
+  "whoop-oauth-state-cleanup": oauthStateCleanup,
+  "oidc-native-handoff-cleanup": short("One DELETE of expired handoff rows."),
+  "fitbit-sync": cohortSync("Fitbit"),
+  "fitbit-backfill": forwarder,
+  "fitbit-sleep-repair": forwarder,
+  "fitbit-oauth-state-cleanup": oauthStateCleanup,
+  "google-health-sync": cohortSync("Google Health"),
+  "google-health-backfill": forwarder,
+  "google-health-sleep-repair": forwarder,
+  "google-health-oauth-state-cleanup": oauthStateCleanup,
+  "sleep-timeline-backfill": forwarder,
+  "lab-biomarker-backfill": forwarder,
+  "nightscout-sync": cohortSync("Nightscout"),
+  "polar-sync": cohortSync("Polar"),
+  "oura-sync": cohortSync("Oura"),
+  "strava-sync": cohortSync("Strava"),
+  "strava-backfill": forwarder,
+
+  // ── Status and insights ────────────────────────────────────────────
+  "insights-general-status": nightlyStatusPass("handleGeneralStatusGenerate"),
+  "insights-blood-pressure-status": nightlyStatusPass(
+    "handleBloodPressureStatusGenerate",
+  ),
+  "insights-weight-status": nightlyStatusPass("handleWeightStatusGenerate"),
+  "insights-pulse-status": nightlyStatusPass("handlePulseStatusGenerate"),
+  "insights-bmi-status": nightlyStatusPass("handleBmiStatusGenerate"),
+  "insights-mood-status": nightlyStatusPass("handleMoodStatusGenerate"),
+  "insights-medication-compliance-status": nightlyStatusPass(
+    "handleMedicationComplianceStatusGenerate",
+  ),
+  "insight-pregenerate": {
+    runtime: "long",
+    expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
+    expiryVia: ["INSIGHT_PREGENERATE_EXPIRE_SECONDS"],
+    stop: {
+      file: "lib/jobs/reminder/insights-handlers.ts",
+      fn: "handleInsightPregenerateJob",
+    },
+    exclusive: "lockedPass",
+    why: "Up to 200 accounts, a comprehensive generation each (a provider call bounded by the account's timeout, up to ten minutes) plus the status warm.",
+  },
+  "morning-digest-refresh": short(
+    "One account, one comprehensive generation, bounded by the provider timeout.",
+  ),
+  "insight-status-generate": short(
+    "One account, one metric, one provider call.",
+  ),
+  "data-arrival": short("One arrival's reactions."),
+  "reaction-line-generate": short("One provider call for one reaction line."),
+  "workout-insight-generate": short("One workout's insight."),
+  "recovery-score-compute": short(
+    "Arithmetic over a recent window, capped at 500 accounts; no provider call.",
+  ),
+  "stress-score-compute": short(
+    "Arithmetic over a recent window, capped at 500 accounts; no provider call.",
+  ),
+  "strain-score-compute": short(
+    "Arithmetic over a recent window, capped at 500 accounts; no provider call.",
+  ),
+  "period-narrative-warm": {
+    runtime: "long",
+    expireInSeconds: PERIOD_NARRATIVE_EXPIRE_SECONDS,
+    expiryVia: ["PERIOD_NARRATIVE_EXPIRE_SECONDS"],
+    stop: "binding",
+    exclusive: "lockedPass",
+    why: "On a boundary night up to 200 accounts with a provider call per period each; a single-account warm is short.",
+  },
+  "coach-memory-refresh": short("One conversation's summary and facts."),
+  "coach-nudge": short(
+    "Light deterministic checks per account; the provider part is capped at 25 calls and 90 seconds a tick.",
+  ),
+  "coach-reminder-sweep": short("Batches of 200 to 500 rows, no provider."),
+  "coach-plan-review": short("A batch of 100 plans, folded in SQL."),
+  "medication-low-stock": short("A light inventory check per account."),
+  "daily-briefing": short("A light cohort scan every fifteen minutes."),
+
+  // ── Rollups ────────────────────────────────────────────────────────
+  "rollup-recompute": short("One bucket."),
+  "rollup-full-backfill": short(
+    "One account's rollup tiers: four aggregate statements, each held to the 60-second statement timeout.",
+  ),
+  "mood-rollup-recompute": short("One bucket."),
+  "mood-rollup-full-backfill": short(
+    "One account's mood entries, a few a day.",
+  ),
+  "medication-compliance-full-backfill": short(
+    "One account's 90-day compliance window.",
+  ),
+  "step-consolidation": short(
+    "One account's legacy step rows, one transaction per day.",
+  ),
+  "step-consolidation-repair": short(
+    "One account's tombstoned step rows inside the retention window.",
+  ),
+  "cumulative-pr-rederive": short(
+    "One account's record detection; measured at 20 seconds for 1.25 million rows.",
+  ),
+  "mean-consolidation": {
+    runtime: "budgeted",
+    budget: "binding",
+    why: "One account's whole history, a day at a time; stops at three quarters of the default expiry and resumes.",
+  },
+  "dense-intraday-retention": {
+    runtime: "budgeted",
+    budget: "binding",
+    why: "A dense account's history, a day at a time; stops at three quarters of the default expiry and resumes.",
+  },
+  "dense-intraday-hourly-rebuild": {
+    runtime: "budgeted",
+    budget: "binding",
+    why: "A dense account's hourly rows, a day at a time; stops at three quarters of the default expiry and resumes.",
+  },
+  "drain-per-sample-cumulative": {
+    runtime: "budgeted",
+    budget: "binding",
+    why: "Every account's per-sample drains; stops at three quarters of the default expiry and hands the rest to a continuation job.",
+  },
+
+  // ── Reminders ──────────────────────────────────────────────────────
+  "medication-reminder-check": short("The due slots of the current minute."),
+  "mood-reminder-check": short("Accounts whose local reminder hour is now."),
+  "cycle-reminder-check": short("Accounts whose local reminder hour is now."),
+  "measurement-reminder-check": short(
+    "Accounts whose local reminder hour is now.",
+  ),
+  "reminder-satisfy": short("One account's open reminders."),
+
+  // ── Maintenance ────────────────────────────────────────────────────
+  "data-backup": {
+    runtime: "long",
+    expireInSeconds: DATA_BACKUP_SEND_OPTIONS.expireInSeconds,
+    expiryVia: ["DATA_BACKUP_SEND_OPTIONS"],
+    stop: {
+      file: "lib/jobs/reminder/backup-handlers.ts",
+      fn: "handleDataBackup",
+    },
+    exclusive: "lockedPass",
+    why: "Serialises every account's whole record.",
+  },
+  "data-backup-offhost": {
+    runtime: "long",
+    expireInSeconds: OFFHOST_BACKUP_EXPIRE_SECONDS,
+    expiryVia: ["OFFHOST_BACKUP_EXPIRE_SECONDS"],
+    stop: {
+      file: "lib/jobs/reminder/backup-handlers.ts",
+      fn: "handleOffhostBackup",
+    },
+    exclusive: "lockedPass",
+    why: "Uploads every account's record off-host.",
+  },
+  "data-restore-drill": short("Fetches and stream-parses one off-host object."),
+  "offhost-backup-purge": short("At most 200 delete requests per run."),
+  "backup-restore": {
+    runtime: "long",
+    expireInSeconds: BACKUP_RESTORE_EXPIRE_SECONDS,
+    expiryVia: ["BACKUP_RESTORE_SEND_OPTIONS", "BACKUP_RESTORE_EXPIRE_SECONDS"],
+    stop: { file: "lib/jobs/backup-restore.ts", fn: "handleBackupRestore" },
+    exclusive: {
+      marker: "restore_claimed",
+      at: { file: "lib/jobs/backup-restore.ts", fn: "runBackupRestoreJob" },
+    },
+    why: "Restores a whole account in one transaction.",
+  },
+  "host-metric-sample": short("One host sample."),
+  "feedback-aggregator": short("One aggregate over the feedback table."),
+  "geo-backfill": short("At most 500 audit rows per pass."),
+  "geolite2-fetch": short("One database download."),
+  "tls-pin-monitor": short("One TLS handshake."),
+  "mood-reminder-cleanup": short("One retention DELETE."),
+  "push-attempt-cleanup": short("One retention DELETE."),
+  "arrival-reaction-cleanup": short("One retention DELETE."),
+  "rate-limit-cleanup": short("One retention DELETE."),
+  "idempotency-cleanup": short("One retention DELETE."),
+  "step-up-elevation-cleanup": short("One retention DELETE."),
+  "coach-message-cleanup": short("One retention DELETE."),
+  "mcp-token-cleanup": short("One retention DELETE."),
+  "audit-log-cleanup": short("At most 40 batches of 5 000 rows per leg."),
+  "measurement-tombstone-cleanup": short(
+    "At most 40 batches of 5 000 rows per leg.",
+  ),
+  "document-tombstone-purge": short("One deleteMany."),
+  "document-summary-reaper": short("One updateMany."),
+  "cycle-prediction-refresh": short(
+    "Arithmetic per eligible account, no provider call.",
+  ),
+  "mood-prognosis-refresh": short(
+    "Arithmetic per eligible account, no provider call.",
+  ),
+  "achievement-unlock-sweep": short(
+    "A bounded achievement read per account, no provider call.",
+  ),
+  "pr-detection": {
+    runtime: "long",
+    expireInSeconds: PR_DETECTION_EXPIRE_SECONDS,
+    expiryVia: ["PR_DETECTION_EXPIRE_SECONDS"],
+    stop: {
+      file: "lib/jobs/reminder/ops-handlers.ts",
+      fn: "handlePrDetection",
+    },
+    exclusive: "lockedPass",
+    why: "The fallback cron scans every account's all-time history every thirty minutes; an ingest job is one account.",
+  },
+  "medication-inventory-expire": short("One updateMany."),
+  "intake-auto-skip": short("One updateMany."),
+  "apple-health-import-v2": {
+    runtime: "long",
+    expireInSeconds: APPLE_HEALTH_IMPORT_SEND_OPTIONS.expireInSeconds,
+    expiryVia: ["APPLE_HEALTH_IMPORT_SEND_OPTIONS"],
+    stop: {
+      oneShot:
+        "The run consumes and unlinks its staged upload, so it can neither stop half-way nor be retried; the send sets retryLimit 0.",
+    },
+    exclusive: {
+      marker: "retryLimit: 0",
+      at: {
+        file: "lib/jobs/apple-health-import-worker.ts",
+        fn: "APPLE_HEALTH_IMPORT_SEND_OPTIONS",
+      },
+    },
+    why: "Parses an export that can be several gigabytes.",
+  },
+  "apple-health-import": short(
+    "The legacy queue: moves each job onto apple-health-import-v2.",
+  ),
+  "apple-health-import-reconcile": short("One updateMany."),
+  "medication-intake-import": {
+    runtime: "long",
+    expireInSeconds: MEDICATION_INTAKE_IMPORT_SEND_OPTIONS.expireInSeconds,
+    expiryVia: ["MEDICATION_INTAKE_IMPORT_SEND_OPTIONS"],
+    stop: {
+      file: "lib/jobs/medication-intake-import.ts",
+      fn: "handleMedicationIntakeImport",
+    },
+    exclusive: {
+      marker: "FOR UPDATE",
+      at: {
+        file: "lib/jobs/medication-intake-import.ts",
+        fn: "processNextChunk",
+      },
+    },
+    why: "Imports a dose-history file chunk by chunk; its length follows the file.",
+  },
+  "intake-slot-dedup": short("One account's duplicate dose slots."),
+  "note-encryption-backfill": short(
+    "Pages of 200 rows of one account's plaintext notes, light per row.",
+  ),
+  "med-notes-encryption-backfill": short(
+    "Pages of 200 rows of one account's plaintext notes, light per row.",
+  ),
+  "free-text-encryption-backfill": short(
+    "Pages of 200 rows of one account's plaintext fields, light per row.",
+  ),
+  "encryption-key-rotate": {
+    runtime: "long",
+    expireInSeconds: ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS,
+    expiryVia: ["ENCRYPTION_KEY_ROTATE_EXPIRE_SECONDS"],
+    stop: {
+      file: "lib/jobs/encryption-key-rotate.ts",
+      fn: "handleEncryptionKeyRotate",
+    },
+    exclusive: "lockedPass",
+    why: "Re-encrypts every encrypted column on the instance.",
+  },
+  "document-content-index-backfill": {
+    runtime: "long",
+    expireInSeconds: CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS,
+    expiryVia: ["CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS"],
+    stop: "binding",
+    exclusive: "lockedPass",
+    why: "Up to 200 documents of one account, a provider transcription each.",
+  },
+  "document-index": short("One document."),
+  "document-thumbnail": short("One document."),
+  "document-summary": short("One document."),
+  "document-summary-catchup": short("Enqueues at most 200 documents."),
+  "document-thumbnail-backfill": short("Enqueues at most 1 000 documents."),
+  "environment-fetch": short(
+    "A fan-out of sends, or one account's bounded archive range.",
+  ),
+};

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

@@ -70,6 +70,7 @@ function buildPrismaMock(
 ) {
   return {
     user: { findMany: vi.fn().mockResolvedValue(users) },
+    dataBackup: { findMany: vi.fn().mockResolvedValue([]) },
   };
 }
 
@@ -236,3 +237,44 @@ describe("handleDataBackup outcome", () => {
     expect(outcome).toMatchObject({ ok: true, did: { backed: 0, total: 0 } });
   });
 });
+
+describe("handleDataBackup on a large instance", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mocks.store.mockResolvedValue(undefined);
+  });
+
+  it("backs up the accounts with the oldest copies first and stops on the budget", async () => {
+    const prisma = buildPrismaMock([
+      { id: "fresh", username: "fresh" },
+      { id: "never", username: "never" },
+      { id: "stale", username: "stale" },
+    ]);
+    prisma.dataBackup.findMany.mockResolvedValue([
+      { userId: "fresh", createdAt: new Date("2026-09-30T00:00:00Z") },
+      { userId: "stale", createdAt: new Date("2026-09-01T00:00:00Z") },
+    ]);
+    mocks.getWorkerPrisma.mockReturnValue(prisma);
+    const controller = new AbortController();
+    // Two accounts in, pg-boss gives up on the job.
+    mocks.store.mockImplementation(async () => {
+      if (mocks.store.mock.calls.length === 2) controller.abort();
+    });
+
+    const outcome = await handleDataBackup([
+      {
+        expireInSeconds: 7200,
+        signal: controller.signal,
+      } as unknown as Parameters<typeof handleDataBackup>[0][number],
+    ]);
+
+    const order = mocks.store.mock.calls.map(
+      ([input]) => (input as { userId: string }).userId,
+    );
+    expect(order).toEqual(["never", "stale"]);
+    expect(outcome).toMatchObject({
+      ok: true,
+      did: { backed: 2, stopped_early: true },
+    });
+  });
+});

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

@@ -121,16 +121,39 @@ export async function handleOffhostBackup(
   });
 }
 
+/**
+ * The weekly on-host copy of every account.
+ *
+ * The pass serialises each account's whole record, so on a large instance it
+ * can outlast its job. It walks the accounts whose weekly copy is oldest
+ * first and stops between accounts once the job's budget is spent
+ * (`jobBudget`), reporting `stopped_early`. The accounts it did not reach have
+ * the oldest copies, so they lead the next pass (the weekly cron or the
+ * admin's "run now"), and every pass makes progress where it matters most.
+ * The binding holds a lock (`lockedPass`), so a retry never runs beside it.
+ */
 export async function handleDataBackup(
   jobs: Job<DataBackupPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
+  const shouldStop = jobBudget(jobs);
   return withBackgroundEvent("job.data_backup", async (evt) => {
     const prisma = getWorkerPrisma();
     try {
-      const users = await prisma.user.findMany({
-        select: { id: true, username: true },
-      });
+      const [accounts, copies] = await Promise.all([
+        prisma.user.findMany({ select: { id: true, username: true } }),
+        prisma.dataBackup.findMany({
+          where: { type: "WEEKLY_AUTO" },
+          select: { userId: true, createdAt: true },
+        }),
+      ]);
+      const copiedAt = new Map(
+        copies.map((copy) => [copy.userId, copy.createdAt.getTime()]),
+      );
+      // No copy yet sorts first, then the oldest copy.
+      const users = [...accounts].sort(
+        (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1),
+      );
+      let stoppedEarly = false;
 
       let backed = 0;
       let usersFailed = 0;
@@ -142,6 +165,10 @@ export async function handleDataBackup(
       // six weeks in.
       let lastError: unknown;
       for (const user of users) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
         reportJobProgress({
           backup_user: user.id,
           backup_users_done: backed + usersFailed + skippedRestoring,
@@ -207,6 +234,7 @@ export async function handleDataBackup(
         users_failed: usersFailed,
         records_oversized: oversized,
         users_skipped_restoring: skippedRestoring,
+        stopped_early: stoppedEarly,
       };
 
       // A pass that wrote nothing for anybody protected nobody, and saying

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

@@ -21,6 +21,7 @@ import {
   runInsightStatusGenerate,
 } from "@/lib/jobs/insight-status-generate";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status";
 import { generateWeightStatusForUser } from "@/lib/insights/weight-status";
@@ -77,6 +78,12 @@ export interface MedicationComplianceStatusPayload {
  * 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.
+ *
+ * The pass walks every candidate with a provider call each, so on a large
+ * instance it can outlast its job. `shouldStop` is the job's time budget: the
+ * pass stops between users once it is spent and reports `stopped_early`. The
+ * users it did not reach keep yesterday's note until the next night or their
+ * next visit; nothing is retried beside a pass that is still running.
  */
 export async function runStatusCronGenerate(
   taskName: string,
@@ -84,6 +91,7 @@ export async function runStatusCronGenerate(
     userId: string,
     options: { locale: string | null; force: boolean },
   ) => Promise<unknown>,
+  shouldStop: () => boolean = () => false,
 ): Promise<JobOutcome> {
   return withBackgroundEvent(taskName, async (evt) => {
     const prisma = getWorkerPrisma();
@@ -96,8 +104,13 @@ export async function runStatusCronGenerate(
 
       let generated = 0;
       let failed = 0;
+      let stoppedEarly = false;
 
       for (const user of users) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
         // The capability before the generator builds a snapshot. The batch
         // entry below checks for itself.
         const capability = await aiCapabilityForJob(user.id, "statusText");
@@ -124,9 +137,14 @@ export async function runStatusCronGenerate(
 
       evt.setBackground({
         task_name: taskName,
-        result: { generated, failed, total: users.length },
+        result: { generated, failed, total: users.length, stoppedEarly },
+      });
+      return jobDone({
+        total: users.length,
+        generated,
+        failed,
+        stopped_early: stoppedEarly,
       });
-      return jobDone({ total: users.length, generated, failed });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -153,7 +171,10 @@ 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<JobOutcome> {
+async function runStatusBatchCron(
+  taskName: string,
+  shouldStop: () => boolean,
+): Promise<JobOutcome> {
   return withBackgroundEvent(taskName, async (evt) => {
     const prisma = getWorkerPrisma();
     try {
@@ -166,7 +187,14 @@ async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
       let generated = 0;
       let served = 0;
       let failed = 0;
+      let stoppedEarly = false;
       for (const user of users) {
+        // Same budget as the per-metric passes: stop between users, and
+        // leave the rest to the later crons and the next night.
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
         try {
           const result = await generateStatusBatchForUser(user.id, {
             // The stored locale, then the operator default — the prompt names
@@ -186,16 +214,34 @@ async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
 
       annotate({
         action: { name: "insights.status.batch.cron" },
-        meta: { generated, served, failed, total: users.length },
+        meta: {
+          generated,
+          served,
+          failed,
+          total: users.length,
+          stopped_early: stoppedEarly,
+        },
       });
       evt.setBackground({
         task_name: taskName,
-        result: { generated, served, failed, total: users.length },
+        result: {
+          generated,
+          served,
+          failed,
+          total: users.length,
+          stoppedEarly,
+        },
       });
       // 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 });
+      return jobDone({
+        total: users.length,
+        generated,
+        served,
+        failed,
+        stopped_early: stoppedEarly,
+      });
     } catch (err) {
       evt.setError(err);
       recordError();
@@ -208,61 +254,66 @@ async function runStatusBatchCron(taskName: string): Promise<JobOutcome> {
 export function handleGeneralStatusGenerate(
   jobs: Job<GeneralStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
-  return runStatusBatchCron("job.insights.batch");
+  return runStatusBatchCron("job.insights.batch", jobBudget(jobs));
 }
 
 export function handleBloodPressureStatusGenerate(
   jobs: Job<BloodPressureStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
   return runStatusCronGenerate(
     "job.insights.blood_pressure",
     generateBloodPressureStatusForUser,
+    jobBudget(jobs),
   );
 }
 
 export function handleWeightStatusGenerate(
   jobs: Job<WeightStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
   return runStatusCronGenerate(
     "job.insights.weight",
     generateWeightStatusForUser,
+    jobBudget(jobs),
   );
 }
 
 export function handlePulseStatusGenerate(
   jobs: Job<PulseStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
   return runStatusCronGenerate(
     "job.insights.pulse",
     generatePulseStatusForUser,
+    jobBudget(jobs),
   );
 }
 
 export function handleBmiStatusGenerate(
   jobs: Job<BmiStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
-  return runStatusCronGenerate("job.insights.bmi", generateBmiStatusForUser);
+  return runStatusCronGenerate(
+    "job.insights.bmi",
+    generateBmiStatusForUser,
+    jobBudget(jobs),
+  );
 }
 
 export function handleMoodStatusGenerate(
   jobs: Job<MoodStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
-  return runStatusCronGenerate("job.insights.mood", generateMoodStatusForUser);
+  return runStatusCronGenerate(
+    "job.insights.mood",
+    generateMoodStatusForUser,
+    jobBudget(jobs),
+  );
 }
 
 export function handleMedicationComplianceStatusGenerate(
   jobs: Job<MedicationComplianceStatusPayload>[],
 ): Promise<JobOutcome> {
-  void jobs;
   return runStatusCronGenerate(
     "job.insights.medication_compliance",
     generateMedicationComplianceStatusForUser,
+    jobBudget(jobs),
   );
 }
 
@@ -310,7 +361,9 @@ export async function handleInsightPregenerateJob(
       return jobDone({ forced: forced.length, scheduled: 0 });
     }
     try {
-      const summary = await runInsightPregenerate(getWorkerPrisma());
+      const summary = await runInsightPregenerate(getWorkerPrisma(), {
+        shouldStop: jobBudget(jobs),
+      });
       evt.setBackground({
         task_name: "job.insight_pregenerate",
         result: { ...summary },
@@ -325,6 +378,7 @@ export async function handleInsightPregenerateJob(
         skipped: summary.skipped,
         failed: summary.failed,
         budget_blocked: summary.budgetBlocked,
+        stopped_early: summary.stoppedEarly,
         assessments_warmed: summary.assessmentsWarmed,
         metric_assessments_warmed: summary.metricAssessmentsWarmed,
       });

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

@@ -4,6 +4,7 @@
  * Extracted from reminder-worker.ts, which owns the queue names, cron
  * schedules, and boss.work registrations.
  */
+import { jobBudget } from "@/lib/jobs/job-budget";
 import { type Job } from "pg-boss";
 import { runGeoBackfill } from "@/lib/jobs/geo-backfill";
 import { fetchGeoLite2Databases } from "@/lib/geo/geolite2-fetch";
@@ -223,6 +224,11 @@ export async function handlePrDetection(
   let insertedAcrossJobs = 0;
   let tiesAcrossJobs = 0;
   let usersFailed = 0;
+  let stoppedEarly = false;
+  // The cron pass walks every account's all-time history and ticks every
+  // thirty minutes. It stops between accounts once the job's budget is spent;
+  // the ingest jobs and the next tick carry the rest.
+  const shouldStop = jobBudget(jobs);
   for (const job of jobs) {
     await withBackgroundEvent("job.pr_detection", async (evt) => {
       const p = getWorkerPrisma();
@@ -241,6 +247,10 @@ export async function handlePrDetection(
       let insertedTotal = 0;
       let tiesTotal = 0;
       for (const userId of userIds) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          break;
+        }
         try {
           const result = await detectPersonalRecordsForUser(userId, {
             silent,
@@ -274,5 +284,6 @@ export async function handlePrDetection(
     pr_detection_inserted: insertedAcrossJobs,
     pr_detection_ties: tiesAcrossJobs,
     pr_detection_users_failed: usersFailed,
+    stopped_early: stoppedEarly,
   });
 }

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

@@ -31,8 +31,12 @@ import {
   PR_DETECTION_QUEUE,
   PR_DETECTION_CONCURRENCY,
   PR_DETECTION_FALLBACK_CRON,
+  PR_DETECTION_EXPIRE_SECONDS,
   type PrDetectionPayload,
 } from "@/lib/jobs/pr-detection";
+import { jobBudget } from "@/lib/jobs/job-budget";
+import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+import { OFFHOST_BACKUP_EXPIRE_SECONDS } from "@/lib/jobs/offhost-backup";
 import {
   MEDICATION_INVENTORY_EXPIRE_QUEUE,
   MEDICATION_INVENTORY_EXPIRE_CRON,
@@ -531,7 +535,11 @@ const schedules: ScheduleEntry[] = [
     STEP_UP_ELEVATION_CLEANUP_CRON,
     cronIsTheRetry,
   ],
-  [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
+  [
+    OFFHOST_BACKUP_QUEUE,
+    OFFHOST_BACKUP_CRON,
+    { expireInSeconds: OFFHOST_BACKUP_EXPIRE_SECONDS },
+  ],
   [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON],
   // Nightly backstop for the purge the deletion kicks straight away: a
   // request the bucket refused, or one written while the queue was down.
@@ -557,7 +565,11 @@ const schedules: ScheduleEntry[] = [
   // cron payload deliberately omits a `userId` so the handler iterates
   // every user; per-user push-suppression cannot apply on the cron
   // path (the silent flag is set by the ingest hooks).
-  [PR_DETECTION_QUEUE, PR_DETECTION_FALLBACK_CRON],
+  [
+    PR_DETECTION_QUEUE,
+    PR_DETECTION_FALLBACK_CRON,
+    { expireInSeconds: PR_DETECTION_EXPIRE_SECONDS },
+  ],
   // v1.4.25 W19b — daily expire-stale pass for the per-pen inventory
   // entities. Flips IN_USE rows whose 30-day clock has blown to
   // EXPIRED at 03:30 Europe/Berlin (in the existing 02:xx–03:xx
@@ -747,7 +759,7 @@ export async function registerMaintenanceQueues(
     boss,
     DATA_BACKUP_QUEUE,
     { localConcurrency: 1 },
-    handleDataBackup,
+    lockedPass(DATA_BACKUP_QUEUE, () => WHOLE_PASS, handleDataBackup),
   );
   await createAndWork<RateLimitCleanupPayload>(
     boss,
@@ -778,7 +790,7 @@ export async function registerMaintenanceQueues(
     boss,
     OFFHOST_BACKUP_QUEUE,
     { localConcurrency: 1, includeMetadata: true },
-    handleOffhostBackup,
+    lockedPass(OFFHOST_BACKUP_QUEUE, () => WHOLE_PASS, handleOffhostBackup),
   );
   await createAndWork(
     boss,
@@ -935,7 +947,16 @@ export async function registerMaintenanceQueues(
     boss,
     PR_DETECTION_QUEUE,
     { localConcurrency: PR_DETECTION_CONCURRENCY },
-    handlePrDetection,
+    // An ingest job is one account; the cron pass is every account, one run
+    // at a time.
+    lockedPass(
+      PR_DETECTION_QUEUE,
+      (job) => {
+        const userId = (job.data as PrDetectionPayload | undefined)?.userId;
+        return userId ? `user:${userId}` : WHOLE_PASS;
+      },
+      handlePrDetection,
+    ),
   );
   await createAndWork<MedicationInventoryExpirePayload>(
     boss,
@@ -1159,7 +1180,11 @@ export async function registerMaintenanceQueues(
     boss,
     ENCRYPTION_KEY_ROTATE_QUEUE,
     { localConcurrency: ENCRYPTION_KEY_ROTATE_CONCURRENCY },
-    handleEncryptionKeyRotate,
+    lockedPass(
+      ENCRYPTION_KEY_ROTATE_QUEUE,
+      () => WHOLE_PASS,
+      handleEncryptionKeyRotate,
+    ),
   );
 
   // v1.25 — medication-note encryption backfill worker. The boot enqueue
@@ -1318,56 +1343,65 @@ export async function registerMaintenanceQueues(
     boss,
     CONTENT_INDEX_BACKFILL_QUEUE,
     { localConcurrency: CONTENT_INDEX_BACKFILL_CONCURRENCY },
-    async (jobs) => {
-      let users = 0;
-      let skipped = 0;
-      let documentsIndexed = 0;
-      let documentsRetokenised = 0;
-      let documentsSkipped = 0;
-      let documentsFailed = 0;
-      for (const job of jobs) {
-        const { userId } = job.data;
-        if (!userId) {
-          skipped++;
-          continue;
+    lockedPass(
+      CONTENT_INDEX_BACKFILL_QUEUE,
+      (job) => `user:${job.data.userId ?? ""}`,
+      async (jobs) => {
+        let users = 0;
+        let stoppedEarly = false;
+        const shouldStop = jobBudget(jobs);
+        let skipped = 0;
+        let documentsIndexed = 0;
+        let documentsRetokenised = 0;
+        let documentsSkipped = 0;
+        let documentsFailed = 0;
+        for (const job of jobs) {
+          const { userId } = job.data;
+          if (!userId) {
+            skipped++;
+            continue;
+          }
+          try {
+            const {
+              indexed,
+              retokenised,
+              skipped: docsSkipped,
+              failed: docsFailed,
+              reason,
+              stoppedEarly: stopped,
+            } = await runContentIndexBackfillForUser(userId, shouldStop);
+            users++;
+            stoppedEarly ||= stopped === true;
+            documentsIndexed += indexed;
+            documentsRetokenised += retokenised;
+            documentsSkipped += docsSkipped;
+            documentsFailed += docsFailed;
+            workerLog(
+              "info",
+              `[document-content-index-backfill] user=${userId} indexed=${indexed} retokenised=${retokenised} skipped=${docsSkipped} failed=${docsFailed} reason=${reason}`,
+            );
+          } catch (err) {
+            recordError();
+            workerLog(
+              "error",
+              `[document-content-index-backfill] user=${userId} failed`,
+              err,
+            );
+            throw err;
+          }
         }
-        try {
-          const {
-            indexed,
-            retokenised,
-            skipped: docsSkipped,
-            failed: docsFailed,
-            reason,
-          } = await runContentIndexBackfillForUser(userId);
-          users++;
-          documentsIndexed += indexed;
-          documentsRetokenised += retokenised;
-          documentsSkipped += docsSkipped;
-          documentsFailed += docsFailed;
-          workerLog(
-            "info",
-            `[document-content-index-backfill] user=${userId} indexed=${indexed} retokenised=${retokenised} skipped=${docsSkipped} failed=${docsFailed} reason=${reason}`,
-          );
-        } catch (err) {
-          recordError();
-          workerLog(
-            "error",
-            `[document-content-index-backfill] user=${userId} failed`,
-            err,
-          );
-          throw err;
-        }
-      }
-      return jobDone({
-        jobs: jobs.length,
-        users,
-        skipped,
-        documents_indexed: documentsIndexed,
-        documents_retokenised: documentsRetokenised,
-        documents_skipped: documentsSkipped,
-        documents_failed: documentsFailed,
-      });
-    },
+        return jobDone({
+          jobs: jobs.length,
+          users,
+          skipped,
+          documents_indexed: documentsIndexed,
+          documents_retokenised: documentsRetokenised,
+          documents_skipped: documentsSkipped,
+          documents_failed: documentsFailed,
+          stopped_early: stoppedEarly,
+        });
+      },
+    ),
   );
 
   // Document AI — automatic per-document content indexing. Enqueued on upload

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

@@ -102,11 +102,16 @@ import {
 } from "@/lib/jobs/insight-status-generate";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
+import { jobBudget } from "@/lib/jobs/job-budget";
+import { lockedPass, WHOLE_PASS } from "@/lib/jobs/long-pass";
+import { INSIGHT_PREGENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-pregenerate-shared";
+import { PERIOD_NARRATIVE_EXPIRE_SECONDS } from "@/lib/jobs/period-narrative-shared";
 import { getWorkerPrisma, workerLog } from "./shared";
 import {
   createAndSchedule,
   createAndWork,
   insightRetryOptions,
+  nightlyInsightPassOptions,
   type QueuePolicyTable,
   type ScheduleEntry,
 } from "./registrar-shared";
@@ -232,25 +237,32 @@ const allQueues = [
 ];
 
 const schedules: ScheduleEntry[] = [
-  [GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, insightRetryOptions],
+  [GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, nightlyInsightPassOptions],
   [
     BLOOD_PRESSURE_STATUS_QUEUE,
     BLOOD_PRESSURE_STATUS_CRON,
-    insightRetryOptions,
+    nightlyInsightPassOptions,
   ],
-  [WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, insightRetryOptions],
-  [PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, insightRetryOptions],
-  [BMI_STATUS_QUEUE, BMI_STATUS_CRON, insightRetryOptions],
+  [WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, nightlyInsightPassOptions],
+  [PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, nightlyInsightPassOptions],
+  [BMI_STATUS_QUEUE, BMI_STATUS_CRON, nightlyInsightPassOptions],
   // v1.15.20 — mood status nightly, continuing the 02:xx ladder.
-  [MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, insightRetryOptions],
+  [MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, nightlyInsightPassOptions],
   [
     MEDICATION_COMPLIANCE_STATUS_QUEUE,
     MEDICATION_COMPLIANCE_STATUS_CRON,
-    insightRetryOptions,
+    nightlyInsightPassOptions,
   ],
   // v1.7.0 — nightly 04:30 Europe/Berlin comprehensive-insight
   // pre-generation. Budget-gated per user inside the handler.
-  [INSIGHT_PREGENERATE_QUEUE, INSIGHT_PREGENERATE_CRON, insightRetryOptions],
+  [
+    INSIGHT_PREGENERATE_QUEUE,
+    INSIGHT_PREGENERATE_CRON,
+    {
+      ...insightRetryOptions,
+      expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
+    },
+  ],
   // v1.10.0 — computed scores (WX-C). Nightly 04:45 Europe/Berlin
   // Recovery-score compute + store, after the rollup-feeding consolidation
   // + drain so the signals it reads are already folded.
@@ -266,7 +278,14 @@ const schedules: ScheduleEntry[] = [
   // v1.11.0 — nightly 05:05 Europe/Berlin period-narrative warm. The
   // handler only fans out on a week (Mon) / month (1st) boundary; every
   // other night is a cheap no-op. Budget-gated per user inside the runner.
-  [PERIOD_NARRATIVE_QUEUE, PERIOD_NARRATIVE_CRON, insightRetryOptions],
+  [
+    PERIOD_NARRATIVE_QUEUE,
+    PERIOD_NARRATIVE_CRON,
+    {
+      ...insightRetryOptions,
+      expireInSeconds: PERIOD_NARRATIVE_EXPIRE_SECONDS,
+    },
+  ],
   // v1.15.20 — daily 05:15 Europe/Berlin proactive Coach nudge, after
   // the 04:45–04:55 score crons so the recovery-score trigger reads
   // settled rows. Deterministic triggers only — no AI call on this path.
@@ -369,43 +388,59 @@ export async function registerStatusQueues(
     boss,
     GENERAL_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handleGeneralStatusGenerate,
+    lockedPass(
+      GENERAL_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleGeneralStatusGenerate,
+    ),
   );
   await createAndWork<BloodPressureStatusPayload>(
     boss,
     BLOOD_PRESSURE_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handleBloodPressureStatusGenerate,
+    lockedPass(
+      BLOOD_PRESSURE_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleBloodPressureStatusGenerate,
+    ),
   );
   await createAndWork<WeightStatusPayload>(
     boss,
     WEIGHT_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handleWeightStatusGenerate,
+    lockedPass(
+      WEIGHT_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleWeightStatusGenerate,
+    ),
   );
   await createAndWork<PulseStatusPayload>(
     boss,
     PULSE_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handlePulseStatusGenerate,
+    lockedPass(PULSE_STATUS_QUEUE, () => WHOLE_PASS, handlePulseStatusGenerate),
   );
   await createAndWork<BmiStatusPayload>(
     boss,
     BMI_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handleBmiStatusGenerate,
+    lockedPass(BMI_STATUS_QUEUE, () => WHOLE_PASS, handleBmiStatusGenerate),
   );
   await createAndWork<MoodStatusPayload>(
     boss,
     MOOD_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handleMoodStatusGenerate,
+    lockedPass(MOOD_STATUS_QUEUE, () => WHOLE_PASS, handleMoodStatusGenerate),
   );
   await createAndWork<MedicationComplianceStatusPayload>(
     boss,
     MEDICATION_COMPLIANCE_STATUS_QUEUE,
     { localConcurrency: 1 },
-    handleMedicationComplianceStatusGenerate,
+    lockedPass(
+      MEDICATION_COMPLIANCE_STATUS_QUEUE,
+      () => WHOLE_PASS,
+      handleMedicationComplianceStatusGenerate,
+    ),
   );
   // v1.7.0 — nightly comprehensive-insight pre-generation.
   //
@@ -422,7 +457,16 @@ export async function registerStatusQueues(
     boss,
     INSIGHT_PREGENERATE_QUEUE,
     { localConcurrency: 2 },
-    handleInsightPregenerateJob,
+    // The cohort walk is one pass at a time; a forced warm is one per
+    // account, so it never waits behind the walk.
+    lockedPass(
+      INSIGHT_PREGENERATE_QUEUE,
+      (job) =>
+        job.data?.force && job.data.userId
+          ? `user:${job.data.userId}`
+          : WHOLE_PASS,
+      handleInsightPregenerateJob,
+    ),
   );
   // S4 — event-driven morning digest refresh. One forced comprehensive
   // regeneration per user per local morning, enqueued by the sleep-arrival
@@ -749,44 +793,53 @@ export async function registerStatusQueues(
     boss,
     PERIOD_NARRATIVE_QUEUE,
     { localConcurrency: 1 },
-    async (jobs): Promise<JobOutcome> => {
-      // Two modes on one queue, so the outcome counts them separately: a
-      // non-boundary night is a scheduled pass that generated nothing, which
-      // is a success with zero counts, not an idle queue.
-      let singleUser = 0;
-      let scheduled = 0;
-      let generated = 0;
-      let failed = 0;
-      for (const job of jobs) {
-        try {
-          if (job.data?.userId) {
-            await warmOneNarrative(job.data);
-            singleUser++;
-          } else {
-            const summary = await runPeriodNarrativeWarm(getWorkerPrisma());
-            scheduled++;
-            generated += summary.generated;
-            failed += summary.failed;
-            workerLog(
-              "info",
-              `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
-            );
+    lockedPass(
+      PERIOD_NARRATIVE_QUEUE,
+      (job) => (job.data?.userId ? `user:${job.data.userId}` : WHOLE_PASS),
+      async (jobs): Promise<JobOutcome> => {
+        // Two modes on one queue, so the outcome counts them separately: a
+        // non-boundary night is a scheduled pass that generated nothing, which
+        // is a success with zero counts, not an idle queue.
+        let singleUser = 0;
+        let scheduled = 0;
+        let generated = 0;
+        let failed = 0;
+        let stoppedEarly = false;
+        for (const job of jobs) {
+          try {
+            if (job.data?.userId) {
+              await warmOneNarrative(job.data);
+              singleUser++;
+            } else {
+              const summary = await runPeriodNarrativeWarm(getWorkerPrisma(), {
+                shouldStop: jobBudget(jobs),
+              });
+              scheduled++;
+              generated += summary.generated;
+              failed += summary.failed;
+              stoppedEarly ||= summary.stoppedEarly;
+              workerLog(
+                "info",
+                `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
+              );
+            }
+          } catch (err) {
+            recordError();
+            await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
+              mode: job.data?.userId ? "single-user" : "scheduled",
+            });
+            throw err;
           }
-        } catch (err) {
-          recordError();
-          await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
-            mode: job.data?.userId ? "single-user" : "scheduled",
-          });
-          throw err;
         }
-      }
-      return jobDone({
-        single_user: singleUser,
-        scheduled,
-        generated,
-        failed,
-      });
-    },
+        return jobDone({
+          single_user: singleUser,
+          scheduled,
+          generated,
+          failed,
+          stopped_early: stoppedEarly,
+        });
+      },
+    ),
   );
   // v1.11.1 — combined Coach memory refresh: rolling conversation summary +
   // durable fact extraction for one long conversation. localConcurrency 1 so a

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

@@ -168,6 +168,22 @@ export const insightRetryOptions = {
   retryBackoff: true,
 } as const;
 
+/**
+ * Send options for a nightly AI pass over every account (the status crons).
+ *
+ * The retries of `insightRetryOptions`, plus a two-hour expiry. Each pass
+ * makes a provider call per account, and on a large instance that outlasted
+ * pg-boss's fifteen-minute default: the job was declared dead and retried
+ * beside the pass still running. The pass now stops itself at three quarters
+ * of this (`jobBudget`) and holds a lock (`lockedPass`). The options ride on
+ * the schedule because the cron is the only sender.
+ */
+export const NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS = 2 * 60 * 60;
+export const nightlyInsightPassOptions = {
+  ...insightRetryOptions,
+  expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS,
+} as const;
+
 /**
  * Retry policy for a cron-driven pass whose failure mode is deterministic.
  *

src/lib/openapi/routes/measurements.ts

@@ -389,6 +389,12 @@ const listMeasurementsMeta = z
       .describe(
         "Rows discarded by the cross-source canonical picker on the day-collapse mode.",
       ),
+    windowFrom: z.iso
+      .datetime({ offset: true })
+      .optional()
+      .describe(
+        "Set on the per-night SLEEP_DURATION mode when the requested window was longer than 365 days: the instant the read started from. The window is a year ending at `to` (or now); page further back by moving `to`.",
+      ),
   })
   .meta({ id: "ListMeasurementsMeta" });
 
@@ -696,7 +702,7 @@ const seriesPointSchema = z.object({
   id: z
     .string()
     .describe(
-      "Measurement row id; `sleep:<wake-day>` for a sleep night; `day:<local-day>` for a day bucket (`pulse` / `glucose` beyond 90 days); `hour:<ISO start>` for an hour bucket (`pulse` / `glucose` within 90 days when the window holds more than 10 000 readings).",
+      "Measurement row id; `sleep:<wake-day>` for a sleep night; `day:<local-day>` for a day bucket (`pulse`, `glucose`, `heartRateVariability`, `oxygenSaturation` beyond 90 days); `hour:<ISO start>` for an hour bucket (the same kinds within 90 days when the window holds more than 10 000 readings).",
     ),
   at: z.iso.datetime({ offset: true }).describe("Point timestamp (ISO-8601)."),
   value: z.number().describe("Primary value in the top-level `unit`."),
@@ -751,7 +757,7 @@ const seriesResponse = z
   .meta({
     id: "MeasurementsSeriesResponse",
     description:
-      "iOS-friendly per-kind time series: one point per reading (or per reconstructed night for `sleep`), an explicit `unit` token, and a summary `stats` block. Sample-dense kinds are bucketed server-side: `pulse` and `glucose` beyond 90 days per local day, and `pulse` / `glucose` within 90 days per local hour when the window holds more than 10 000 readings (a continuous glucose sensor, a watch recording heart rate every minute; a cuff or fingerstick meter stays raw). A bucket carries the mean as `value`; a `pulse` bucket also carries its range as `valueMin` / `valueMax`, and a `glucose` bucket is converted to the user's unit like a raw reading. `stats` is always computed over the underlying readings, never over the buckets.",
+      "iOS-friendly per-kind time series: one point per reading (or per reconstructed night for `sleep`), an explicit `unit` token, and a summary `stats` block. Sample-dense kinds (`pulse`, `glucose`, `heartRateVariability`, `oxygenSaturation`) are bucketed server-side: beyond 90 days per local day, and within 90 days per local hour when the window holds more than 10 000 readings (a continuous glucose sensor, a watch or ring recording every minute; a cuff, a fingerstick meter or a few readings a night stay raw). A bucket carries the mean as `value` and otherwise has the shape of a reading of its kind; a `pulse` bucket also carries its range as `valueMin` / `valueMax`, and a `glucose` bucket is converted to the user's unit like a raw reading. `stats` is always computed over the underlying readings, never over the buckets.",
   });
 
 // ── Batched daily series (v1.18.6 dashboard fetch coalescing) ────────

src/lib/targets/__tests__/builders.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from "vitest";
 import { buildGlucoseTargets } from "../glucose-builder";
+import { foldGlucoseTargetRows } from "../glucose-read";
 import { buildMedicationTarget } from "../medication-builder";
 import { buildMoodTargets } from "../mood-builder";
 import { buildSleepTarget } from "../sleep-builder";
@@ -8,11 +9,13 @@ import { buildVitalTargets } from "../vitals-builder";
 
 const NOW = new Date("2026-07-21T12:00:00.000Z");
 const TZ = "UTC";
+const RECENT = new Date(NOW.getTime() - 30 * 86_400_000);
 
 describe("target section builders", () => {
   it("keeps the vital target order and profile-gated cards on an empty profile", () => {
     const result = buildVitalTargets({
       recentMeasurements: [],
+      restingPulseProxy: [],
       latestByType: {},
       average30ByType: {},
       heightCm: null,
@@ -108,7 +111,7 @@ describe("target section builders", () => {
       },
     ];
     const targets = buildGlucoseTargets({
-      rows,
+      summaries: foldGlucoseTargetRows(rows, RECENT, TZ),
       profile: {
         heightCm: 180,
         dateOfBirth: new Date("1985-01-01T00:00:00.000Z"),
@@ -140,10 +143,14 @@ describe("target section builders", () => {
     // metadata, so every row arrives with a null context. Iterating the four
     // named contexts alone returned no glucose target at all.
     const targets = buildGlucoseTargets({
-      rows: [
-        { value: 112, measuredAt: NOW, glucoseContext: null },
-        { value: 104, measuredAt: NOW, glucoseContext: null },
-      ],
+      summaries: foldGlucoseTargetRows(
+        [
+          { value: 112, measuredAt: NOW, glucoseContext: null },
+          { value: 104, measuredAt: NOW, glucoseContext: null },
+        ],
+        RECENT,
+        TZ,
+      ),
       profile: {
         heightCm: 180,
         dateOfBirth: new Date("1985-01-01T00:00:00.000Z"),
@@ -173,11 +180,15 @@ describe("target section builders", () => {
 
   it("keeps the tagged breakdown and puts the untagged bucket last (#943)", () => {
     const targets = buildGlucoseTargets({
-      rows: [
-        { value: 130, measuredAt: NOW, glucoseContext: "RANDOM" as const },
-        { value: 90, measuredAt: NOW, glucoseContext: "FASTING" as const },
-        { value: 111, measuredAt: NOW, glucoseContext: null },
-      ],
+      summaries: foldGlucoseTargetRows(
+        [
+          { value: 130, measuredAt: NOW, glucoseContext: "RANDOM" as const },
+          { value: 90, measuredAt: NOW, glucoseContext: "FASTING" as const },
+          { value: 111, measuredAt: NOW, glucoseContext: null },
+        ],
+        RECENT,
+        TZ,
+      ),
       profile: {
         heightCm: 180,
         dateOfBirth: new Date("1985-01-01T00:00:00.000Z"),
@@ -235,6 +246,7 @@ describe("target section builders", () => {
 describe("buildVitalTargets — user weight target", () => {
   const base = {
     recentMeasurements: [],
+    restingPulseProxy: [],
     latestByType: {},
     average30ByType: {},
     heightCm: 178,

src/lib/targets/build-response.ts

@@ -27,7 +27,9 @@ import {
   ensureUserMoodRollupsFresh,
   readMoodDayRollups,
 } from "@/lib/rollups/mood-rollups";
+import { readRestingPulseProxy } from "@/lib/analytics/resting-pulse-read";
 import { buildGlucoseTargets } from "./glucose-builder";
+import { readGlucoseTargetSummaries } from "./glucose-read";
 import { buildMedicationTarget } from "./medication-builder";
 import { buildMoodTargets } from "./mood-builder";
 import { buildSleepTarget } from "./sleep-builder";
@@ -76,6 +78,10 @@ export async function buildTargetsResponse(user: AuthedUser) {
     "BODY_FAT",
     "ACTIVITY_STEPS",
   ];
+  // Heart rate is the one vital a device streams. Its thirty days reach the
+  // resting-pulse card as one proxy point per day, folded in SQL below; every
+  // other vital is written a few times a day and is read as rows.
+  const recentTypes = types.filter((type) => type !== "PULSE");
   const thirtyDaysAgo = new Date(queryNow.getTime() - 30 * 24 * 60 * 60 * 1000);
   const oneYearAgo = new Date(queryNow.getTime() - 365 * 24 * 60 * 60 * 1000);
   const moodSince = new Date(
@@ -96,7 +102,8 @@ export async function buildTargetsResponse(user: AuthedUser) {
     medicationBundle,
     moodRollups,
     latestMoodEntry,
-    glucoseRows,
+    glucoseSummaries,
+    restingPulseProxy,
   ] = await Promise.all([
     limit(() =>
       prisma.user.findUnique({
@@ -116,7 +123,7 @@ export async function buildTargetsResponse(user: AuthedUser) {
       prisma.measurement.findMany({
         where: {
           userId,
-          type: { in: types },
+          type: { in: recentTypes },
           measuredAt: { gte: thirtyDaysAgo },
           deletedAt: null,
         },
@@ -212,16 +219,22 @@ export async function buildTargetsResponse(user: AuthedUser) {
         select: { score: true, moodLoggedAt: true },
       }),
     ),
+    // A sensor reporting every five minutes writes 105 000 glucose readings
+    // a year. The cards need the newest per meal context and the last thirty
+    // days per local day, so both fold in SQL.
     limit(() =>
-      prisma.measurement.findMany({
-        where: {
-          userId,
-          type: "BLOOD_GLUCOSE",
-          measuredAt: { gte: oneYearAgo },
-          deletedAt: null,
-        },
-        orderBy: { measuredAt: "desc" },
-        select: { value: true, measuredAt: true, glucoseContext: true },
+      readGlucoseTargetSummaries({
+        userId,
+        since: oneYearAgo,
+        recentSince: thirtyDaysAgo,
+        timeZone: timezone,
+      }),
+    ),
+    limit(() =>
+      readRestingPulseProxy({
+        userId,
+        since: thirtyDaysAgo,
+        timeZone: timezone,
       }),
     ),
   ]);
@@ -252,6 +265,8 @@ export async function buildTargetsResponse(user: AuthedUser) {
   for (const type of types) {
     latestByType[type] =
       latestEverByType.find((row) => row.type === type)?.value ?? null;
+  }
+  for (const type of recentTypes) {
     const recent = recentMeasurements.filter(
       (measurement) => measurement.type === type,
     );
@@ -267,6 +282,7 @@ export async function buildTargetsResponse(user: AuthedUser) {
 
   const vitalSection = buildVitalTargets({
     recentMeasurements,
+    restingPulseProxy,
     latestByType,
     average30ByType,
     heightCm,
@@ -326,7 +342,7 @@ export async function buildTargetsResponse(user: AuthedUser) {
   const glucoseUnit = resolveGlucoseUnit(dbUser?.glucoseUnit ?? null);
   targets.push(
     ...buildGlucoseTargets({
-      rows: glucoseRows,
+      summaries: glucoseSummaries,
       profile: {
         heightCm,
         dateOfBirth: dbUser?.dateOfBirth ?? null,

src/lib/targets/consistency.ts

@@ -43,12 +43,13 @@ function last7DayKeys(now: Date, timezone: string): string[] {
   return keys;
 }
 
-function computeStrip(
+/** A day's readings folded to their sum and count, keyed by local day. */
+export type DaySums = ReadonlyMap<string, { sum: number; count: number }>;
+
+function sumByDay(
   events: ReadonlyArray<{ measuredAt: Date; value: number }>,
-  classify: (mean: number) => DayBand | null,
   timezone: string,
-  now: Date,
-) {
+): Map<string, { sum: number; count: number }> {
   const byDay = new Map<string, { sum: number; count: number }>();
   for (const event of events) {
     const key = userDayKey(event.measuredAt, timezone);
@@ -57,7 +58,15 @@ function computeStrip(
     bucket.count += 1;
     byDay.set(key, bucket);
   }
+  return byDay;
+}
 
+function computeStrip(
+  byDay: DaySums,
+  classify: (mean: number) => DayBand | null,
+  timezone: string,
+  now: Date,
+) {
   const consistency7d: Array<DayBand | null> = [];
   let daysInRange7d = 0;
   let daysLogged7d = 0;
@@ -77,13 +86,40 @@ function computeStrip(
 
 export function rollupConsistency({
   events,
+  ...rest
+}: RollupConsistencyInput): TargetConsistency {
+  return rollupConsistencyFromDays({
+    ...rest,
+    byDay: sumByDay(events, rest.timezone),
+    readingCount: events.length,
+  });
+}
+
+interface RollupDaySumsInput extends ConsistencyClock {
+  /** Each local day's readings as sum and count. */
+  byDay: DaySums;
+  /** Readings behind `byDay`, for the thin-data threshold. */
+  readingCount: number;
+  classify: (mean: number) => DayBand | null;
+  totalReadingsThreshold?: number;
+}
+
+/**
+ * The same consistency as {@link rollupConsistency}, from readings already
+ * folded per local day. Every figure here is a function of each day's mean
+ * and of the reading count, so a reader that folds in SQL hands over one row
+ * per day instead of every reading and gets the same answer.
+ */
+export function rollupConsistencyFromDays({
+  byDay,
+  readingCount,
   classify,
   timezone,
   now,
   totalReadingsThreshold = 3,
-}: RollupConsistencyInput): TargetConsistency {
-  if (events.length < totalReadingsThreshold) {
-    const partial = computeStrip(events, classify, timezone, now);
+}: RollupDaySumsInput): TargetConsistency {
+  if (readingCount < totalReadingsThreshold) {
+    const partial = computeStrip(byDay, classify, timezone, now);
     return {
       ...EMPTY_CONSISTENCY,
       ...partial,
@@ -91,15 +127,7 @@ export function rollupConsistency({
     };
   }
 
-  const byDay = new Map<string, { sum: number; count: number }>();
-  for (const event of events) {
-    const key = userDayKey(event.measuredAt, timezone);
-    const bucket = byDay.get(key) ?? { sum: 0, count: 0 };
-    bucket.sum += event.value;
-    bucket.count += 1;
-    byDay.set(key, bucket);
-  }
-  const strip = computeStrip(events, classify, timezone, now);
+  const strip = computeStrip(byDay, classify, timezone, now);
 
   let daysInRange30d = 0;
   let daysLogged30d = 0;
@@ -136,7 +164,7 @@ export function rollupConsistency({
     lastMetGoalAt,
     streakDays,
     insufficientData:
-      events.length < totalReadingsThreshold || strip.daysLogged7d < 1,
+      readingCount < totalReadingsThreshold || strip.daysLogged7d < 1,
     consistency7d: strip.consistency7d,
   };
 }

src/lib/targets/glucose-builder.ts

@@ -4,8 +4,9 @@ import {
   type GlucoseContextBucket,
 } from "@/lib/glucose";
 import { resolveGlucoseTarget } from "./glucose-targets";
-import { makeRangeClassifier, rollupConsistency } from "./consistency";
-import type { TargetGlucoseRow, TargetItem, TargetProfile } from "./types";
+import { makeRangeClassifier, rollupConsistencyFromDays } from "./consistency";
+import type { TargetGlucoseContextSummary } from "./glucose-read";
+import type { TargetItem, TargetProfile } from "./types";
 
 const LABEL_BY_CONTEXT: Record<GlucoseContextBucket, string> = {
   FASTING: "targets.glucoseFasting",
@@ -16,38 +17,53 @@ const LABEL_BY_CONTEXT: Record<GlucoseContextBucket, string> = {
 };
 
 interface GlucoseTargetsInput {
-  rows: TargetGlucoseRow[];
+  /**
+   * One summary per stored meal context: the newest reading of the year and
+   * the last thirty days per local day (`readGlucoseTargetSummaries`).
+   */
+  summaries: TargetGlucoseContextSummary[];
   profile: TargetProfile;
   timezone: string;
   now: Date;
 }
 
 export function buildGlucoseTargets({
-  rows,
+  summaries,
   profile,
   timezone,
   now,
 }: GlucoseTargetsInput): TargetItem[] {
   const targets: TargetItem[] = [];
   const unit = resolveGlucoseUnit(profile.glucoseUnit);
-  const thirtyDaysAgo = new Date(now.getTime() - 30 * 24 * 60 * 60 * 1000);
 
   // #943 — the untagged bucket rides the same loop. Iterating the four named
   // contexts alone left an account whose source writes no meal-time tag with
   // no glucose target at all, the same gap the dashboard tile had.
-  for (const [context, contextRows] of groupByGlucoseContext(
-    rows,
-    (row) => row.glucoseContext,
+  for (const [context, contextSummaries] of groupByGlucoseContext(
+    summaries,
+    (summary) => summary.glucoseContext,
   )) {
-    const latest = contextRows[0].value;
-    const recent = contextRows.filter((row) => row.measuredAt >= thirtyDaysAgo);
+    // A bucket is one stored context today; folding several keeps the newest
+    // reading and adds the days up, should two ever share a bucket.
+    const newest = contextSummaries.reduce((a, b) =>
+      b.latestAt > a.latestAt ? b : a,
+    );
+    const latest = newest.latest;
+    const recentByDay = new Map<string, { sum: number; count: number }>();
+    let recentCount = 0;
+    let recentSum = 0;
+    for (const summary of contextSummaries) {
+      recentCount += summary.recentCount;
+      for (const [day, bucket] of summary.recentByDay) {
+        const current = recentByDay.get(day) ?? { sum: 0, count: 0 };
+        current.sum += bucket.sum;
+        current.count += bucket.count;
+        recentByDay.set(day, current);
+        recentSum += bucket.sum;
+      }
+    }
     const average30 =
-      recent.length > 0
-        ? Math.round(
-            (recent.reduce((sum, row) => sum + row.value, 0) / recent.length) *
-              10,
-          ) / 10
-        : null;
+      recentCount > 0 ? Math.round((recentSum / recentCount) * 10) / 10 : null;
     const resolved = resolveGlucoseTarget({
       context,
       hasDiabetes: profile.hasDiabetes,
@@ -97,11 +113,9 @@ export function buildGlucoseTargets({
           : resolved.source === "ADA goal (diabetes)"
             ? "ADA goal (diabetes)"
             : "ADA 2024 / DDG",
-      ...rollupConsistency({
-        events: recent.map((row) => ({
-          measuredAt: row.measuredAt,
-          value: row.value,
-        })),
+      ...rollupConsistencyFromDays({
+        byDay: recentByDay,
+        readingCount: recentCount,
         classify: makeRangeClassifier(
           range,
           effectiveRange

src/lib/targets/glucose-read.ts

@@ -0,0 +1,148 @@
+/**
+ * The glucose rows the targets page needs, folded in Postgres.
+ *
+ * The glucose targets used to read every glucose reading of the last year and
+ * keep three things from them per meal context: the newest value, and for the
+ * last thirty days each local day's mean and the number of readings. A sensor
+ * that reports every five minutes puts 105 000 readings in that year, and the
+ * targets page is read on every visit, on the web and in the app. So the year
+ * is reduced to one row per context (`DISTINCT ON`) and the thirty days to one
+ * row per context and local day, and only those rows cross the wire.
+ *
+ * Output matches {@link foldGlucoseTargetRows} over the raw readings: the same
+ * newest value per context (ties broken by id, newest first), and the same
+ * per-day sums and counts, with days cut in the zone the caller passes by the
+ * expression every other day fold uses.
+ */
+import { Prisma, type PrismaClient } from "@/generated/prisma/client";
+import type { GlucoseContext } from "@/generated/prisma/client";
+import { prisma } from "@/lib/db";
+import { isValidTimezone, userDayKey } from "@/lib/tz/format";
+import type { TargetGlucoseRow } from "./types";
+
+/** One meal context's newest reading and its last thirty days, per day. */
+export interface TargetGlucoseContextSummary {
+  glucoseContext: GlucoseContext | null;
+  latest: number;
+  latestAt: Date;
+  /** Local day → the day's readings as sum and count, since `recentSince`. */
+  recentByDay: Map<string, { sum: number; count: number }>;
+  /** Readings behind `recentByDay`. */
+  recentCount: number;
+}
+
+/**
+ * The in-memory fold over raw readings, newest first. Pure; the reference the
+ * SQL reader is tested against.
+ */
+export function foldGlucoseTargetRows(
+  rows: readonly TargetGlucoseRow[],
+  recentSince: Date,
+  timeZone: string,
+): TargetGlucoseContextSummary[] {
+  const byContext = new Map<string, TargetGlucoseContextSummary>();
+  for (const row of rows) {
+    const key = row.glucoseContext ?? "";
+    let summary = byContext.get(key);
+    if (!summary) {
+      summary = {
+        glucoseContext: row.glucoseContext,
+        latest: row.value,
+        latestAt: row.measuredAt,
+        recentByDay: new Map(),
+        recentCount: 0,
+      };
+      byContext.set(key, summary);
+    }
+    if (row.measuredAt < recentSince) continue;
+    const day = userDayKey(row.measuredAt, timeZone);
+    const bucket = summary.recentByDay.get(day) ?? { sum: 0, count: 0 };
+    bucket.sum += row.value;
+    bucket.count += 1;
+    summary.recentByDay.set(day, bucket);
+    summary.recentCount += 1;
+  }
+  return [...byContext.values()];
+}
+
+export async function readGlucoseTargetSummaries(opts: {
+  userId: string;
+  /** Inclusive lower bound for the newest reading per context. */
+  since: Date;
+  /** Inclusive lower bound for the per-day fold. */
+  recentSince: Date;
+  /** IANA zone the day boundary is cut in. */
+  timeZone: string;
+  db?: Pick<PrismaClient, "$queryRaw">;
+}): Promise<TargetGlucoseContextSummary[]> {
+  const db = opts.db ?? prisma;
+  const timeZone = isValidTimezone(opts.timeZone) ? opts.timeZone : "UTC";
+  const [latestRows, dayRows] = await Promise.all([
+    db.$queryRaw<
+      Array<{
+        glucose_context: GlucoseContext | null;
+        value: number;
+        measured_at: Date;
+      }>
+    >(Prisma.sql`
+      SELECT DISTINCT ON (m."glucose_context")
+        m."glucose_context" AS glucose_context,
+        m."value"           AS value,
+        m."measured_at"     AS measured_at
+      FROM measurements m
+      WHERE m."user_id" = ${opts.userId}
+        AND m."type" = 'BLOOD_GLUCOSE'::measurement_type
+        AND m."deleted_at" IS NULL
+        AND m."measured_at" >= ${opts.since}
+      ORDER BY m."glucose_context", m."measured_at" DESC, m."id" DESC
+    `),
+    db.$queryRaw<
+      Array<{
+        glucose_context: GlucoseContext | null;
+        day: string;
+        n: number;
+        sum: number;
+      }>
+    >(Prisma.sql`
+      WITH src AS (
+        SELECT
+          m."glucose_context" AS glucose_context,
+          to_char((m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}, 'YYYY-MM-DD') AS day,
+          m."value" AS value
+        FROM measurements m
+        WHERE m."user_id" = ${opts.userId}
+          AND m."type" = 'BLOOD_GLUCOSE'::measurement_type
+          AND m."deleted_at" IS NULL
+          AND m."measured_at" >= ${opts.recentSince}
+      )
+      SELECT
+        glucose_context,
+        day,
+        COUNT(*)::int                AS n,
+        SUM(value)::double precision AS sum
+      FROM src
+      GROUP BY glucose_context, day
+    `),
+  ]);
+
+  const byContext = new Map<string, TargetGlucoseContextSummary>();
+  for (const row of latestRows) {
+    byContext.set(row.glucose_context ?? "", {
+      glucoseContext: row.glucose_context,
+      latest: Number(row.value),
+      latestAt: row.measured_at,
+      recentByDay: new Map(),
+      recentCount: 0,
+    });
+  }
+  for (const row of dayRows) {
+    // Every recent reading is also inside the year, so its context has a
+    // newest row unless `recentSince` predates `since`, which no caller does.
+    const summary = byContext.get(row.glucose_context ?? "");
+    if (!summary) continue;
+    const n = Number(row.n);
+    summary.recentByDay.set(row.day, { sum: Number(row.sum), count: n });
+    summary.recentCount += n;
+  }
+  return [...byContext.values()];
+}

src/lib/targets/vitals-builder.ts

@@ -13,7 +13,10 @@ import {
   classifyPulseByTarget,
   getPersonalizedPulseTarget,
 } from "@/lib/analytics/pulse-targets";
-import { resolveRestingPulseSeries } from "@/lib/analytics/resting-pulse";
+import {
+  mergeRestingWithProxy,
+  type PulseSample,
+} from "@/lib/analytics/resting-pulse";
 import { getBodyFatTargetRange } from "@/lib/analytics/value-bands";
 import { binaryReferenceSex, type ProfileSex } from "@/lib/profile/sex";
 import { userDayKey } from "@/lib/tz/resolver";
@@ -38,7 +41,15 @@ export interface BloodPressureTargetRange {
 }
 
 interface VitalTargetsInput {
+  /** The last thirty days of every vital type except `PULSE`. */
   recentMeasurements: TargetMeasurement[];
+  /**
+   * The resting-pulse proxy over the same thirty days, one point per local
+   * day, folded from raw `PULSE` in Postgres (`readRestingPulseProxy`). Raw
+   * heart rate is never read here: a watch that records it once a minute puts
+   * 43 000 readings in thirty days, and the card needs one figure per day.
+   */
+  restingPulseProxy: PulseSample[];
   latestByType: TargetValueByType;
   average30ByType: TargetValueByType;
   heightCm: number | null;
@@ -149,6 +160,7 @@ function buildBpPairsByDay(
 
 export function buildVitalTargets({
   recentMeasurements,
+  restingPulseProxy,
   latestByType,
   average30ByType,
   heightCm,
@@ -285,19 +297,14 @@ export function buildVitalTargets({
   }
 
   const pulseTarget = getPersonalizedPulseTarget(age, referenceSex);
-  const restingResolved = resolveRestingPulseSeries({
+  const restingResolved = mergeRestingWithProxy({
     restingSamples: recentMeasurements
       .filter((measurement) => measurement.type === "RESTING_HEART_RATE")
       .map((measurement) => ({
         measuredAt: measurement.measuredAt,
         value: measurement.value,
       })),
-    pulseSamples: recentMeasurements
-      .filter((measurement) => measurement.type === "PULSE")
-      .map((measurement) => ({
-        measuredAt: measurement.measuredAt,
-        value: measurement.value,
-      })),
+    proxySamples: restingPulseProxy,
     dayKeyOf: (date) => userDayKey(date, timezone),
   });
   const restingEvents = restingResolved.series;