Give the status, integration-sync and maintenance queues explicit policies

The per-connection backfills and per-user one-shots take 'exclusive': a
worker restarting during a heavy account's multi-hour pass used to append
another identical full-history job per restart, because the discovery
predicate only stops offering a job once it has completed.

The morning-digest refresh takes 'exclusive' so its user+localDate key
becomes a real at-most-once-per-morning contract; a wall-clock
singletonSeconds window cannot express a local date.

The per-document queues take 'short' -- they have no discovery pass to
re-converge, so only queued duplicates may be collapsed or a re-process
after a content change would be lost.

Leave the environment-fetch queue on 'standard' and say why: its
explicit-range backfill deliberately sends without a key, so any policy
would silently merge two different requested ranges into one.

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-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,