Merge branch 'worktree-agent-a87a74abcff7c248e' into release/v1.30.22

src/lib/jobs/__tests__/queue-policy-tables.test.ts

@@ -0,0 +1,158 @@
+/**
+ * Dead-policy guard, in the same spirit as the v1.4.37 dead-queue guards.
+ *
+ * A queue's `singletonKey` only de-duplicates if its registrar names a policy
+ * for it: pg-boss constrains `singleton_key` through partial unique indexes
+ * scoped by queue policy, and the default `standard` policy has none. So the
+ * enqueue site and the registrar entry are two halves of one guarantee that
+ * live in different files, and nothing in the type system couples them —
+ * deleting the registrar entry leaves the `singletonKey` at the call site
+ * looking entirely correct while it silently stops doing anything.
+ *
+ * These assertions are that coupling. Each queue below is listed with the
+ * policy its call site depends on; dropping or weakening an entry turns this
+ * suite red instead of quietly restoring the storm.
+ *
+ * The behavioural proof that the policies mean what the names say lives in
+ * `tests/integration/queue-singleton-policy.integration.test.ts`, which runs a
+ * real pg-boss against a real Postgres.
+ */
+import fs from "node:fs";
+import path from "node:path";
+
+import { describe, expect, it } from "vitest";
+
+const read = (file: string) =>
+  fs.readFileSync(path.resolve(__dirname, "../reminder", file), "utf8");
+
+/**
+ * Assert that `queue` appears in the registrar's `queuePolicies` table with
+ * `policy`. Anchored to the table so an unrelated mention of the constant
+ * elsewhere in the file cannot satisfy it.
+ */
+function expectPolicy(source: string, queue: string, policy: string) {
+  const table = source.match(
+    /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/,
+  );
+  expect(table, "registrar has no queuePolicies table").not.toBeNull();
+  const entry = new RegExp(`\\[${queue}\\]:\\s*\\{\\s*policy:\\s*"${policy}"`);
+  expect(
+    table![1],
+    `${queue} must carry the "${policy}" policy — without it the singletonKey at its enqueue site is inert`,
+  ).toMatch(entry);
+}
+
+describe("rollup registrar queue policies", () => {
+  const source = read("register-rollup.ts");
+
+  // `short`, not `exclusive`: the handler re-reads the bucket when it starts,
+  // so only QUEUED duplicates may be collapsed. Promoting either of these to
+  // `exclusive` would strand a measurement written while a recompute ran.
+  it.each([
+    ["ROLLUP_RECOMPUTE_QUEUE", "short"],
+    ["MOOD_ROLLUP_RECOMPUTE_QUEUE", "short"],
+  ])("%s is %s", (queue, policy) => {
+    expectPolicy(source, queue, policy);
+  });
+
+  // Per-user, self-converging backfills: a second concurrent run is pure
+  // duplicated work and discovery re-offers the job while it is outstanding.
+  it.each([
+    ["ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"],
+    ["MOOD_ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"],
+    ["MEDICATION_COMPLIANCE_BACKFILL_QUEUE", "exclusive"],
+    ["STEP_CONSOLIDATION_QUEUE", "exclusive"],
+    ["STEP_CONSOLIDATION_REPAIR_QUEUE", "exclusive"],
+    ["CUMULATIVE_PR_REDERIVE_QUEUE", "exclusive"],
+    ["MEAN_CONSOLIDATION_QUEUE", "exclusive"],
+    ["DENSE_INTRADAY_RETENTION_QUEUE", "exclusive"],
+    ["DENSE_INTRADAY_HOURLY_REBUILD_QUEUE", "exclusive"],
+  ])("%s is %s", (queue, policy) => {
+    expectPolicy(source, queue, policy);
+  });
+});
+
+describe("status registrar queue policies", () => {
+  const source = read("register-status.ts");
+
+  it("MORNING_DIGEST_REFRESH_QUEUE is exclusive", () => {
+    // Seven sleep-write seams enqueue the same user+localDate key. Without
+    // this the at-most-once-per-morning contract does not hold and the failure
+    // path re-triggers forced comprehensive generations against the provider.
+    expectPolicy(source, "MORNING_DIGEST_REFRESH_QUEUE", "exclusive");
+  });
+});
+
+describe("integration-sync registrar queue policies", () => {
+  const source = read("register-integration-sync.ts");
+
+  it.each([
+    ["WHOOP_BACKFILL_QUEUE", "exclusive"],
+    ["FITBIT_BACKFILL_QUEUE", "exclusive"],
+    ["GOOGLE_HEALTH_BACKFILL_QUEUE", "exclusive"],
+    ["STRAVA_BACKFILL_QUEUE", "exclusive"],
+    ["GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE", "exclusive"],
+    ["SLEEP_TIMELINE_BACKFILL_QUEUE", "exclusive"],
+    ["LAB_BIOMARKER_BACKFILL_QUEUE", "exclusive"],
+  ])("%s is %s", (queue, policy) => {
+    expectPolicy(source, queue, policy);
+  });
+});
+
+describe("maintenance registrar queue policies", () => {
+  const source = read("register-maintenance.ts");
+
+  it.each([
+    ["INTAKE_SLOT_DEDUP_QUEUE", "exclusive"],
+    ["NOTE_ENCRYPTION_BACKFILL_QUEUE", "exclusive"],
+    ["MED_NOTES_ENCRYPTION_BACKFILL_QUEUE", "exclusive"],
+    ["DOCUMENT_THUMBNAIL_BACKFILL_QUEUE", "exclusive"],
+    ["CONTENT_INDEX_BACKFILL_QUEUE", "exclusive"],
+    ["ENCRYPTION_KEY_ROTATE_QUEUE", "exclusive"],
+  ])("%s is %s", (queue, policy) => {
+    expectPolicy(source, queue, policy);
+  });
+
+  // Per-document queues have no discovery pass to re-converge, so a re-process
+  // requested after the running job read the old bytes must be admitted.
+  it.each([
+    ["DOCUMENT_INDEX_QUEUE", "short"],
+    ["DOCUMENT_THUMBNAIL_QUEUE", "short"],
+    ["DOCUMENT_SUMMARY_QUEUE", "short"],
+  ])("%s is %s", (queue, policy) => {
+    expectPolicy(source, queue, policy);
+  });
+
+  it("ENVIRONMENT_FETCH_QUEUE is deliberately left on standard", () => {
+    // Its explicit-range backfill sends with no key at all, on purpose, so any
+    // policy would collapse two different requested date ranges onto the shared
+    // empty key. This asserts the omission is intentional and documented, so a
+    // future reader does not "complete" the table and silently merge them.
+    const table = source.match(
+      /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/,
+    );
+    expect(table![1]).not.toMatch(/\[ENVIRONMENT_FETCH_QUEUE\]/);
+    expect(source).toMatch(/ENVIRONMENT_FETCH_QUEUE is LEFT ALONE/);
+  });
+});
+
+describe("every policy decision carries a reason", () => {
+  it.each([
+    "register-rollup.ts",
+    "register-status.ts",
+    "register-integration-sync.ts",
+    "register-maintenance.ts",
+  ])("%s", (file) => {
+    const table = read(file).match(
+      /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/,
+    );
+    expect(table).not.toBeNull();
+
+    const policies = table![1].match(/policy:\s*"(?:short|exclusive)"/g) ?? [];
+    const reasons = table![1].match(/reason:\s*\n?\s*"/g) ?? [];
+    expect(policies.length).toBeGreaterThan(0);
+    // A policy without a stated reason is the thing this whole change exists
+    // to prevent: the next person re-deriving intent from the call sites.
+    expect(reasons.length).toBe(policies.length);
+  });
+});

src/lib/jobs/morning-digest-refresh-shared.ts

@@ -37,10 +37,18 @@ export interface MorningDigestRefreshPayload {
  * pre-checked in the trigger) stops any later sample from re-running it — the
  * pair is at-most-once per user per local morning.
  *
+ * The key alone does NOT buy that: pg-boss only constrains `singleton_key`
+ * when the queue carries a policy that indexes it, and under the default
+ * `standard` policy no such index exists, so this key coalesced nothing at all
+ * until the queue was given `exclusive` in `reminder/register-status.ts`. If
+ * that entry is ever removed, this debounce silently disappears again — the two
+ * belong together.
+ *
  * No `singletonSeconds`: a time-slot throttle would let a second morning's
- * refresh be swallowed if it fell inside the window; the local-date-scoped key
- * is already unique per morning, so the plain active-job constraint is exactly
- * the debounce we want. No-ops cleanly when no global boss instance is
+ * refresh be swallowed if it fell inside the window, and its wall-clock
+ * `floor()` bucketing cannot express a user's local date anyway; the
+ * local-date-scoped key plus the `exclusive` policy is exactly the debounce we
+ * want. No-ops cleanly when no global boss instance is
  * available (a web process without an embedded worker) — the next sleep
  * landing, and the nightly cron, remain the catch-net.
  */

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

@@ -64,7 +64,11 @@ import {
   type LabBiomarkerBackfillPayload,
 } from "@/lib/jobs/lab-biomarker-backfill";
 import { workerLog } from "./shared";
-import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  createAndSchedule,
+  type QueuePolicyTable,
+  type ScheduleEntry,
+} from "./registrar-shared";
 import {
   WithingsSyncPayload,
   WithingsActivitySyncPayload,
@@ -326,6 +330,62 @@ const schedules: ScheduleEntry[] = [
   [OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, OIDC_NATIVE_HANDOFF_CLEANUP_CRON],
 ];
 
+/**
+ * De-duplication policy per integration-sync queue.
+ *
+ * The recurring poll queues (Withings / WHOOP / Fitbit / Google Health /
+ * Nightscout / Polar / Oura / Strava sync, and every OAuth-state cleanup) are
+ * keyless cron ticks — no `singletonKey` on any send — so a policy would only
+ * constrain the empty key. They are deliberately absent.
+ *
+ * The one-shot backfills and repairs below all share the same shape: one job
+ * per connection, keyed per user (per provider where a provider can differ),
+ * enqueued by a boot-discovery predicate that only stops offering the job once
+ * it has COMPLETED. That last property is what made the bug bite — a worker
+ * restarting during a heavy account's multi-hour full-history pass appended
+ * another identical job per restart, sustaining the boot storm the stagger
+ * deferrals exist to damp. `exclusive` covers queued, active, and retry-backoff,
+ * so the restart case is closed; and because discovery re-offers the job while
+ * the work is still outstanding, a suppressed send can never lose work.
+ */
+const queuePolicies: QueuePolicyTable = {
+  [WHOOP_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-connection full-history backfill; long-running, and boot discovery re-enqueues until the connection is marked backfilled.",
+  },
+  [FITBIT_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-connection full-history backfill; boot-discovery driven and self-converging.",
+  },
+  [GOOGLE_HEALTH_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-connection full-history backfill; boot-discovery driven and self-converging.",
+  },
+  [STRAVA_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-connection full-history backfill; boot-discovery driven and self-converging.",
+  },
+  [GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-connection one-shot sleep re-read; discovery drops the connection once it is marked repaired.",
+  },
+  [SLEEP_TIMELINE_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-connection one-shot timeline backfill, keyed per provider+user; discovery drops the connection once its rows carry the new shape.",
+  },
+  [LAB_BIOMARKER_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user one-shot catalog link-up; discovery drops the user once no unlinked lab readings remain.",
+  },
+};
+
 /**
  * Register every integration-sync queue: create, schedule, and bind handlers.
  * Returns the queue names created (for the boot-level aggregate assertion).
@@ -333,7 +393,7 @@ const schedules: ScheduleEntry[] = [
 export async function registerIntegrationSyncQueues(
   boss: PgBoss,
 ): Promise<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   await boss.work<WithingsSyncPayload>(
     WITHINGS_SYNC_QUEUE,

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

@@ -115,7 +115,11 @@ import {
 } from "@/lib/jobs/environment-fetch";
 import { recordError } from "@/lib/jobs/worker-status";
 import { workerLog, BOOT_BACKFILL_STAGGER_SECONDS } from "./shared";
-import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  createAndSchedule,
+  type QueuePolicyTable,
+  type ScheduleEntry,
+} from "./registrar-shared";
 import {
   DataBackupPayload,
   OffhostBackupPayload,
@@ -425,6 +429,86 @@ const schedules: ScheduleEntry[] = [
   [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON],
 ];
 
+/**
+ * De-duplication policy per maintenance queue.
+ *
+ * Every retention / cleanup cron here (rate-limit, idempotency, audit-log,
+ * push-attempt, tombstone, coach-message, MCP-token, document-purge, …) is a
+ * keyless tick with no `singletonKey` on any send, so a policy would only
+ * constrain the empty key. Those are deliberately absent from this table.
+ *
+ * Two deliberate omissions worth naming, because both look like candidates:
+ *
+ *   - ENVIRONMENT_FETCH_QUEUE is LEFT ALONE. It is genuinely ambiguous: the
+ *     lookback refresh sends a per-user key, but an explicit-range backfill
+ *     sends with NO options at all, deliberately, so that "it always runs".
+ *     Any policy would collapse every keyless explicit-range backfill onto the
+ *     shared empty key, so two different requested date ranges would silently
+ *     become one. That is exactly the class of silent work-dropping a policy is
+ *     supposed to prevent, so the queue keeps `standard` until the enqueue side
+ *     gives the explicit-range variant a key of its own.
+ *   - APPLE_HEALTH_IMPORT_QUEUE, DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send
+ *     keylessly by design (each import / backup / detection run is a distinct
+ *     unit of work). A policy would coalesce independent runs.
+ */
+const queuePolicies: QueuePolicyTable = {
+  // Per-user, boot- and cron-discovery driven, self-converging one-shots. The
+  // discovery predicate re-offers the job while the work is outstanding, so
+  // suppressing a duplicate of a queued OR active pass cannot lose work.
+  [INTAKE_SLOT_DEDUP_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user duplicate-slot cleanup; discovery drops the user once no colliding intake pair remains.",
+  },
+  [NOTE_ENCRYPTION_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user note-encryption backfill; discovery drops the user once no plaintext note remains.",
+  },
+  [MED_NOTES_ENCRYPTION_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user medication-note encryption backfill; discovery drops the user once no plaintext note remains.",
+  },
+  [DOCUMENT_THUMBNAIL_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user thumbnail backfill; discovery drops the user once every thumbnailable document has a preview.",
+  },
+  [CONTENT_INDEX_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user index-all run, triggered from the documents surface. Idempotent (it indexes only not-yet-indexed documents), so a second press while one runs is correctly a no-op.",
+  },
+  [ENCRYPTION_KEY_ROTATE_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Fixed singleton key, admin-triggered. Two concurrent corpus rotations must never overlap; the route already reports a suppressed send back as alreadyQueued.",
+  },
+
+  // Per-document, enqueued on upload. `short`, NOT `exclusive`: each handler
+  // re-reads the document when it starts, so collapsing sends that arrive while
+  // an identical job is still queued is safe. `exclusive` would additionally
+  // suppress a re-process requested after the current job had already read the
+  // old bytes, leaving a stale summary/thumbnail/index behind with no discovery
+  // pass to re-offer it — these queues have no cron to re-converge.
+  [DOCUMENT_INDEX_QUEUE]: {
+    policy: "short",
+    reason:
+      "Per-document, no re-converging discovery pass. Collapse queued duplicates only, so a re-index after a content change is never dropped.",
+  },
+  [DOCUMENT_THUMBNAIL_QUEUE]: {
+    policy: "short",
+    reason:
+      "Per-document, no re-converging discovery pass. Collapse queued duplicates only.",
+  },
+  [DOCUMENT_SUMMARY_QUEUE]: {
+    policy: "short",
+    reason:
+      "Per-document, no re-converging discovery pass. Collapse queued duplicates only.",
+  },
+};
+
 /**
  * Register every maintenance / ops / cleanup queue. Returns the queue names
  * created (for the boot-level aggregate assertion).
@@ -432,7 +516,7 @@ const schedules: ScheduleEntry[] = [
 export async function registerMaintenanceQueues(
   boss: PgBoss,
 ): Promise<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   await boss.work<DataBackupPayload>(
     DATA_BACKUP_QUEUE,

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

@@ -102,7 +102,11 @@ import {
   workerLog,
   BOOT_BACKFILL_STAGGER_SECONDS,
 } from "./shared";
-import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  createAndSchedule,
+  type QueuePolicyTable,
+  type ScheduleEntry,
+} from "./registrar-shared";
 // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows.
 // Collapses each user × cumulative-type × calendar-day bucket into one
 // `stats:…` row so the list view stops painting hundreds of step chunks per
@@ -169,6 +173,93 @@ const schedules: ScheduleEntry[] = [
   [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON],
 ];
 
+/**
+ * De-duplication policy per rollup queue. Until this table existed every queue
+ * here ran under pg-boss's `standard` policy, under which no unique index
+ * constrains `singleton_key` — so the `singletonKey` each enqueue site passes
+ * was inert and de-duplicated nothing.
+ *
+ * The split below is deliberate and the two halves are NOT interchangeable.
+ */
+const queuePolicies: QueuePolicyTable = {
+  // `short`, not `exclusive`. A recompute reads the bucket's live rows when it
+  // STARTS, so collapsing sends that pile up while an identical job is still
+  // queued is safe — the queued job will observe every write that landed in the
+  // meantime. This is what kills the storm: a sync posting a few hundred batch
+  // requests fanned out three jobs per type-day per request, thousands of jobs
+  // re-deriving byte-identical rows. Under `short` they collapse to one queued
+  // job per bucket. `exclusive` would be wrong here: it would also suppress a
+  // send issued after the recompute had already started reading, stranding the
+  // newer measurement in a stale bucket until the nightly pass.
+  [ROLLUP_RECOMPUTE_QUEUE]: {
+    policy: "short",
+    reason:
+      "Burst coalescing for the write-path fan-out. The handler re-reads the bucket at run time, so collapsing queued duplicates cannot strand a write; exclusive could.",
+  },
+  [MOOD_ROLLUP_RECOMPUTE_QUEUE]: {
+    policy: "short",
+    reason:
+      "Same shape as the measurement recompute: re-reads the bucket at run time, so only queued duplicates may be collapsed.",
+  },
+
+  // The rest are per-user, self-converging boot/discovery backfills. A second
+  // concurrent run is pure duplicated work, and the discovery predicate
+  // re-enqueues on the next boot or cron tick while work remains outstanding —
+  // so `exclusive` (queued OR active OR awaiting retry) can never lose work it
+  // does not re-offer. `short` would be too weak: these passes run for minutes
+  // to hours, and the observed failure was precisely a worker restarting
+  // mid-pass and appending another identical full-history job per restart.
+  [ROLLUP_FULL_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user full-history backfill; long-running. Boot discovery re-enqueues while coverage is still missing, so suppressing a duplicate of an ACTIVE pass is safe.",
+  },
+  [MOOD_ROLLUP_FULL_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user mood backfill; boot-discovery driven and self-converging on rollup coverage.",
+  },
+  [MEDICATION_COMPLIANCE_BACKFILL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user compliance backfill; boot-discovery driven and self-converging on rollup coverage.",
+  },
+  [STEP_CONSOLIDATION_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user legacy step consolidation; discovery drops the user once no live granular rows remain.",
+  },
+  [STEP_CONSOLIDATION_REPAIR_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user one-shot repair; discovery drops the user once the affected provider rows are repaired.",
+  },
+  [CUMULATIVE_PR_REDERIVE_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user one-shot re-derivation; discovery drops the user once the inflated records are rewritten.",
+  },
+  [MEAN_CONSOLIDATION_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user daily-mean consolidation; discovery drops the user once the spot rows are folded.",
+  },
+  [DENSE_INTRADAY_RETENTION_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user dense-tier drain; the multi-hour pass on a heavy account is exactly the case where a restart used to append a duplicate full-history job.",
+  },
+  [DENSE_INTRADAY_HOURLY_REBUILD_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Per-user one-shot hourly rebuild; discovery drops the day once its retired daily row is gone.",
+  },
+
+  // DRAIN_CUMULATIVE_QUEUE is deliberately absent: it is a keyless nightly cron
+  // tick with no singletonKey on any send, so a policy would only constrain the
+  // empty key and buy nothing.
+};
+
 /**
  * Register every rollup / consolidation queue. Returns the queue names created
  * (for the boot-level aggregate assertion).
@@ -176,7 +267,7 @@ const schedules: ScheduleEntry[] = [
 export async function registerRollupQueues(
   boss: PgBoss,
 ): Promise<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   // v1.5.0 — persistent measurement rollup worker. Folds the
   // WEEK / MONTH / YEAR buckets that the write-path hooks enqueue;

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

@@ -92,6 +92,7 @@ import { getWorkerPrisma, workerLog } from "./shared";
 import {
   createAndSchedule,
   insightRetryOptions,
+  type QueuePolicyTable,
   type ScheduleEntry,
 } from "./registrar-shared";
 import {
@@ -261,6 +262,40 @@ const schedules: ScheduleEntry[] = [
   [DAILY_BRIEFING_QUEUE, DAILY_BRIEFING_CRON],
 ];
 
+/**
+ * De-duplication policy per status queue.
+ *
+ * Most queues here are keyless nightly cron ticks — no `singletonKey` on any
+ * send — so a policy would only constrain the empty key and is deliberately
+ * omitted. The LLM-bound warm queues (insight pre-generate, per-metric status
+ * generate, period narrative, Coach memory refresh) already pass an explicit
+ * `singletonSeconds` on every send, which populates `singleton_on` and is
+ * therefore constrained by pg-boss's `job_i4` index REGARDLESS of queue policy.
+ * Those already de-duplicate today; changing their policy would layer a second,
+ * differently-shaped constraint over a working one for no gain, so they are
+ * left alone.
+ *
+ * That leaves exactly one queue here whose key is inert today.
+ */
+const queuePolicies: QueuePolicyTable = {
+  // The sleep-arrival trigger has seven write seams that can each enqueue for
+  // the same user and the same local date. The key is
+  // `morning-refresh:<user>:<localDate>`, and `exclusive` turns that into a
+  // real at-most-once-per-user-per-local-morning contract across queued,
+  // active, and retry-backoff states.
+  //
+  // `singletonSeconds` is NOT the tool here and the existing comment at the
+  // enqueue site says so: its window is a floor() over wall-clock, which does
+  // not line up with a user's local date, so it would both split one morning
+  // across two buckets and merge two mornings into one. The local date already
+  // lives in the key; the policy just has to honour it.
+  [MORNING_DIGEST_REFRESH_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Seven sleep-write seams enqueue the same user+localDate key; exclusive makes the at-most-once-per-morning contract real. A wall-clock singletonSeconds window cannot express a local date.",
+  },
+};
+
 /**
  * Register every status / insight / computed-score queue. Returns the queue
  * names created (for the boot-level aggregate assertion).
@@ -268,7 +303,7 @@ const schedules: ScheduleEntry[] = [
 export async function registerStatusQueues(
   boss: PgBoss,
 ): Promise<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   await boss.work<GeneralStatusPayload>(
     GENERAL_STATUS_QUEUE,

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

@@ -18,6 +18,10 @@
  */
 import { PgBoss } from "pg-boss";
 
+import { prisma } from "@/lib/db";
+
+import { workerLog } from "./shared";
+
 /**
  * Cron schedule tuple: `[queueName, cronExpression, sendOptions?]`. The
  * optional third element carries per-queue send options (e.g. the LLM-bound
@@ -25,6 +29,102 @@ import { PgBoss } from "pg-boss";
  */
 export type ScheduleEntry = [string, string, Record<string, unknown>?];
 
+/**
+ * The pg-boss queue policies this codebase names explicitly. `standard` —
+ * pg-boss's default — is deliberately absent: under `standard` NO partial
+ * unique index covers `singleton_key` at all, so a bare `singletonKey` on a
+ * `send()` is inert and de-duplicates nothing. A queue that wants
+ * de-duplication has to name one of the two policies below.
+ *
+ * From the partial unique indexes pg-boss 12.26 creates on the shared
+ * `job_common` table:
+ *
+ *   - `short`     → UNIQUE (name, COALESCE(singleton_key,'')) WHERE state = 'created'
+ *                   Collapses duplicate sends only while a job is still QUEUED.
+ *                   Once it goes active, a fresh send is admitted again.
+ *   - `exclusive` → UNIQUE (name, COALESCE(singleton_key,'')) WHERE state <= 'active'
+ *                   Collapses duplicate sends while a job is queued OR active
+ *                   OR waiting out a retry backoff.
+ *
+ * Picking between them is a correctness question, not a taste question:
+ *
+ *   - `short` when the handler re-reads current state at run time. Suppressing
+ *     a send while an identical job is still QUEUED is then provably safe —
+ *     that job has read nothing yet and will observe the newer write when it
+ *     starts. Suppressing a send after the reader already STARTED would strand
+ *     the newer write, which is why these queues must not be `exclusive`.
+ *   - `exclusive` when a second concurrent run is pure duplicated work AND the
+ *     enqueue side is self-converging — a discovery pass that re-enqueues on
+ *     the next boot or cron tick while the work is still outstanding.
+ *
+ * On a conflict pg-boss inserts with `ON CONFLICT DO NOTHING ... RETURNING`, so
+ * `boss.send()` resolves to `null` instead of throwing. Every enqueue helper in
+ * this tree already counts a null id as `skipped`.
+ */
+export type QueuePolicy = "short" | "exclusive";
+
+/**
+ * A per-queue policy decision plus the reason behind it. The reason is not
+ * decoration: the policy encodes a claim about whether the handler re-reads
+ * state at run time and whether the enqueue side re-converges. Anyone changing
+ * a handler needs that claim written down rather than re-derived from the call
+ * sites.
+ */
+export type QueuePolicyDecision = {
+  policy: QueuePolicy;
+  reason: string;
+};
+
+/** Queue name → decision. Queues absent from a table keep pg-boss's `standard`. */
+export type QueuePolicyTable = Readonly<Record<string, QueuePolicyDecision>>;
+
+/**
+ * Bring the policy of ALREADY-EXISTING queues in line with the tables below.
+ *
+ * Why this is needed at all: pg-boss's `create_queue()` SQL function ends in
+ * `ON CONFLICT DO NOTHING`, so passing `{ policy }` to `boss.createQueue()`
+ * only takes effect the first time a queue is provisioned. Every queue on an
+ * already-running instance was created under the default `standard` policy, and
+ * `boss.updateQueue()` rejects a policy change outright ("queue policy cannot be
+ * changed after creation"). Without this reconcile the policy tables would be
+ * correct on a fresh database and completely inert on every existing
+ * deployment — a change that tests green and fixes nothing.
+ *
+ * Why writing the column directly is sound here: the partial unique indexes for
+ * every policy are created once on the shared `job_common` table at schema
+ * setup (this deployment does not use per-queue table partitioning), so the
+ * index a newly-claimed policy needs already exists. A job row's own `policy`
+ * value is resolved by joining `pgboss.queue` at insert time, so a changed
+ * column takes effect on the next `send()` without a worker restart and without
+ * depending on pg-boss's in-process queue cache. Jobs already in flight keep
+ * the policy they were inserted under and simply do not participate in the new
+ * index, which is the correct transitional behaviour.
+ *
+ * Scope is deliberately narrow: only queue names this codebase decided on, only
+ * where the stored policy actually differs, parameter-bound, and never a change
+ * to a queue absent from the tables.
+ */
+async function reconcileQueuePolicies(
+  policies: QueuePolicyTable,
+): Promise<void> {
+  for (const [name, { policy }] of Object.entries(policies)) {
+    try {
+      const changed = await prisma.$executeRaw`
+        UPDATE pgboss.queue
+        SET policy = ${policy}, updated_on = now()
+        WHERE name = ${name} AND policy IS DISTINCT FROM ${policy}
+      `;
+      if (changed > 0) {
+        workerLog("info", `[queue-policy] reconciled ${name} to ${policy}`);
+      }
+    } catch (err) {
+      // Never fail worker boot on a reconcile miss: the queue still works, it
+      // just keeps whatever de-duplication semantics it already had.
+      workerLog("error", `[queue-policy] failed to reconcile ${name}`, err);
+    }
+  }
+}
+
 /**
  * Shared retry policy for the LLM-bound insight queues. A transient failure
  * (provider hiccup, pool exhaustion) used to fail the nightly tick silently
@@ -42,15 +142,25 @@ export const insightRetryOptions = {
  * Centralised so each registrar provisions before it schedules in the exact
  * order the monolith did, and the `Europe/Berlin` tz default stays in one
  * place.
+ *
+ * `policies` carries the registrar's per-queue de-duplication decisions. It is
+ * applied on both legs — as the creation policy for a queue that does not exist
+ * yet, and through `reconcileQueuePolicies` for one that does. A queue absent
+ * from the table keeps pg-boss's `standard` policy, which means no
+ * de-duplication at all; that is a valid choice, but it must be a deliberate
+ * one, so the tables document the omissions too.
  */
 export async function createAndSchedule(
   boss: PgBoss,
   queues: readonly string[],
   schedules: readonly ScheduleEntry[],
+  policies: QueuePolicyTable = {},
 ): Promise<void> {
   for (const q of queues) {
-    await boss.createQueue(q);
+    const decision = policies[q];
+    await boss.createQueue(q, decision ? { policy: decision.policy } : {});
   }
+  await reconcileQueuePolicies(policies);
   for (const [name, cron, sendOptions] of schedules) {
     await boss.schedule(
       name,

src/lib/rollups/measurement-rollups.ts

@@ -302,6 +302,15 @@ export async function enqueueRollupRecompute(input: {
     // measurements landing on the same day shouldn't each spawn a
     // separate worker run. The singleton key uses bucket-start so
     // two writes inside the same week share one queued job.
+    //
+    // The key only does that because the queue carries the `short` policy
+    // (see `reminder/register-rollup.ts`); under pg-boss's default `standard`
+    // policy nothing indexes `singleton_key` and this key coalesced nothing,
+    // which is how a large sync came to fan out thousands of jobs re-deriving
+    // byte-identical rows. `short` collapses sends only while an identical job
+    // is still QUEUED — safe here precisely because the handler re-reads the
+    // bucket's live rows when it starts, so the queued job still sees writes
+    // that land after it was enqueued.
     singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`,
   });
 }

tests/integration/queue-singleton-policy.integration.test.ts

@@ -0,0 +1,281 @@
+/**
+ * Queue de-duplication semantics, asserted against a real pg-boss on a real
+ * Postgres.
+ *
+ * This suite exists because the bug it pins is invisible to a mocked test. A
+ * bare `singletonKey` on a `send()` LOOKS like de-duplication at every call
+ * site, and a unit test with a stubbed boss will happily agree. But pg-boss
+ * only constrains `singleton_key` through partial unique indexes that are
+ * scoped BY QUEUE POLICY, and under the default `standard` policy no such index
+ * exists — so every bare `singletonKey` in the tree de-duplicated nothing.
+ * Only a real insert against a real index can tell the two apart.
+ *
+ * The `standard` case below is deliberately kept as a permanent, executable
+ * statement of the original defect: if someone drops a policy entry from a
+ * registrar table, the matching assertion here goes red instead of the
+ * behaviour silently reverting to "the key does nothing".
+ */
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { PgBoss } from "pg-boss";
+
+let boss: PgBoss;
+
+/** Unique per run so a re-run never collides with a queue left behind. */
+const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`;
+const q = (name: string) => `test-${name}-${suffix}`;
+
+beforeAll(async () => {
+  boss = new PgBoss({
+    connectionString: process.env.DATABASE_URL,
+    // Keep the maintenance/monitor loops out of the way; this suite only
+    // exercises createQueue/send/fetch.
+    schedule: false,
+    supervise: false,
+  });
+  await boss.start();
+}, 120_000);
+
+afterAll(async () => {
+  await boss?.stop({ graceful: false });
+});
+
+describe("pg-boss singleton semantics by queue policy", () => {
+  it("standard: a bare singletonKey de-duplicates NOTHING (the original defect)", async () => {
+    const name = q("standard");
+    await boss.createQueue(name);
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
+
+    // Both inserted. This is the finding: under the default policy the key is
+    // inert, so every "coalesced by singletonKey" comment in the tree was
+    // describing a guarantee that did not exist.
+    expect(first).toBeTruthy();
+    expect(second).toBeTruthy();
+    expect(second).not.toBe(first);
+  });
+
+  it("exclusive: suppresses a duplicate while the first is QUEUED", async () => {
+    const name = q("excl-queued");
+    await boss.createQueue(name, { policy: "exclusive" });
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" });
+
+    expect(first).toBeTruthy();
+    // pg-boss inserts with ON CONFLICT DO NOTHING ... RETURNING, so a
+    // suppressed send resolves to null rather than throwing. Every boot
+    // discovery helper in the tree counts exactly this as `skipped`.
+    expect(second).toBeNull();
+  });
+
+  it("exclusive: still suppresses once the first job is ACTIVE", async () => {
+    const name = q("excl-active");
+    await boss.createQueue(name, { policy: "exclusive" });
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" });
+    expect(first).toBeTruthy();
+
+    // Move the job out of `created` and into `active`.
+    const fetched = await boss.fetch(name);
+    expect(fetched).toHaveLength(1);
+
+    // This is the case that matters for the backfills: a worker restarting
+    // during a heavy account's multi-hour pass used to append another
+    // identical full-history job, because `short` semantics would have let it
+    // through. `exclusive` covers queued OR active.
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" });
+    expect(second).toBeNull();
+  });
+
+  it("exclusive: a genuinely distinct key still enqueues", async () => {
+    const name = q("excl-distinct");
+    await boss.createQueue(name, { policy: "exclusive" });
+
+    const a = await boss.send(name, { n: 1 }, { singletonKey: "user-a" });
+    const b = await boss.send(name, { n: 2 }, { singletonKey: "user-b" });
+
+    expect(a).toBeTruthy();
+    expect(b).toBeTruthy();
+    expect(b).not.toBe(a);
+  });
+
+  it("short: suppresses a duplicate while the first is QUEUED", async () => {
+    const name = q("short-queued");
+    await boss.createQueue(name, { policy: "short" });
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" });
+
+    expect(first).toBeTruthy();
+    expect(second).toBeNull();
+  });
+
+  it("short: ADMITS a send once the first job is ACTIVE", async () => {
+    const name = q("short-active");
+    await boss.createQueue(name, { policy: "short" });
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" });
+    expect(first).toBeTruthy();
+
+    const fetched = await boss.fetch(name);
+    expect(fetched).toHaveLength(1);
+
+    // This is the property that makes `short` the correct choice for the
+    // rollup recompute and the per-document jobs: the running job already read
+    // its input, so a write that lands afterwards MUST be able to enqueue its
+    // own recompute or it would be stranded in a stale bucket. `exclusive`
+    // would return null here and silently drop that work.
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" });
+    expect(second).toBeTruthy();
+    expect(second).not.toBe(first);
+  });
+
+  it("short: a genuinely distinct key still enqueues", async () => {
+    const name = q("short-distinct");
+    await boss.createQueue(name, { policy: "short" });
+
+    const a = await boss.send(name, { n: 1 }, { singletonKey: "bucket-1" });
+    const b = await boss.send(name, { n: 2 }, { singletonKey: "bucket-2" });
+
+    expect(a).toBeTruthy();
+    expect(b).toBeTruthy();
+    expect(b).not.toBe(a);
+  });
+});
+
+describe("policy reconcile for queues that already exist", () => {
+  it("createQueue does NOT change the policy of an existing queue", async () => {
+    const name = q("recreate");
+    await boss.createQueue(name);
+
+    // pg-boss's create_queue() ends in ON CONFLICT DO NOTHING, so this is a
+    // no-op rather than an upgrade. Without the reconcile in
+    // `registrar-shared.ts`, shipping the policy tables would therefore change
+    // nothing at all on an instance whose queues already exist -- which is
+    // every running instance.
+    await boss.createQueue(name, { policy: "exclusive" });
+
+    const queue = await boss.getQueue(name);
+    expect(queue?.policy).toBe("standard");
+
+    // And the behaviour confirms it: the key is still inert.
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
+    expect(first).toBeTruthy();
+    expect(second).toBeTruthy();
+  });
+
+  it("updateQueue refuses a policy change outright", async () => {
+    const name = q("update");
+    await boss.createQueue(name);
+
+    await expect(
+      // @ts-expect-error -- policy is excluded from UpdateQueueOptions by
+      // design; this asserts the runtime guard that forces the direct-column
+      // reconcile the registrar performs.
+      boss.updateQueue(name, { policy: "exclusive" }),
+    ).rejects.toThrow(/policy cannot be changed/i);
+  });
+
+  it("writing the policy column takes effect on the next send", async () => {
+    const name = q("reconciled");
+    await boss.createQueue(name);
+
+    // Same statement `reconcileQueuePolicies` issues. The partial unique
+    // indexes for every policy already exist on the shared `job_common` table
+    // (this deployment does not partition per queue), so claiming a policy
+    // later is enough -- no schema change, no worker restart.
+    await boss.getDb().executeSql(
+      `UPDATE pgboss.queue SET policy = $2, updated_on = now()
+         WHERE name = $1 AND policy IS DISTINCT FROM $2`,
+      [name, "exclusive"],
+    );
+
+    const queue = await boss.getQueue(name);
+    expect(queue?.policy).toBe("exclusive");
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
+    expect(first).toBeTruthy();
+    expect(second).toBeNull();
+  });
+});
+
+describe("createAndSchedule applies the registrar's policy table", () => {
+  it("provisions a NEW queue with the decided policy", async () => {
+    const { createAndSchedule } =
+      await import("@/lib/jobs/reminder/registrar-shared");
+    const name = q("cas-new");
+
+    await createAndSchedule(boss, [name], [], {
+      [name]: { policy: "exclusive", reason: "test" },
+    });
+
+    const queue = await boss.getQueue(name);
+    expect(queue?.policy).toBe("exclusive");
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
+    expect(first).toBeTruthy();
+    expect(second).toBeNull();
+  });
+
+  it("reconciles an EXISTING standard queue onto the decided policy", async () => {
+    const { createAndSchedule } =
+      await import("@/lib/jobs/reminder/registrar-shared");
+    const name = q("cas-existing");
+
+    // Simulate a running instance: the queue already exists under `standard`,
+    // which is how every queue on every deployed instance was provisioned.
+    await boss.createQueue(name);
+    expect((await boss.getQueue(name))?.policy).toBe("standard");
+
+    await createAndSchedule(boss, [name], [], {
+      [name]: { policy: "short", reason: "test" },
+    });
+
+    expect((await boss.getQueue(name))?.policy).toBe("short");
+
+    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
+    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
+    expect(first).toBeTruthy();
+    expect(second).toBeNull();
+  });
+
+  it("leaves a queue absent from the table on standard", async () => {
+    const { createAndSchedule } =
+      await import("@/lib/jobs/reminder/registrar-shared");
+    const name = q("cas-absent");
+
+    await createAndSchedule(boss, [name], [], {});
+
+    expect((await boss.getQueue(name))?.policy).toBe("standard");
+  });
+});
+
+describe("boot-discovery skipped counter", () => {
+  it("can actually be non-zero once the queue carries a policy", async () => {
+    const name = q("skipped");
+    await boss.createQueue(name, { policy: "exclusive" });
+
+    // Mirror the shape every boot-discovery helper uses: iterate a candidate
+    // set, count a null job id as `skipped`. Two of these three users are
+    // already represented, which under `standard` could never happen -- which
+    // is why `skipped` was structurally always zero in the boot log lines.
+    const candidates = ["u1", "u1", "u2", "u2", "u3"];
+    let enqueued = 0;
+    let skipped = 0;
+    for (const userId of candidates) {
+      const jobId = await boss.send(name, { userId }, { singletonKey: userId });
+      if (jobId) {
+        enqueued += 1;
+      } else {
+        skipped += 1;
+      }
+    }
+
+    expect(enqueued).toBe(3);
+    expect(skipped).toBe(2);
+  });
+});