feat(workouts): write-time canonical-row dedup for cross-source batches

iOS HealthKit observer queue routinely carries the same logical workout
from two paired sensors (Apple Watch + Withings ScanWatch) in a single
ingest batch. Without write-time dedup both rows persist (their
externalIds differ, so the (userId, source, externalId) unique index
never fires) and the read-time picker has to carry the duplicate
forward on every dashboard render.

Add pickCanonicalWorkoutRows() at src/lib/workouts/canonical-rows.ts as
a pure pre-createMany pass invoked by POST /api/workouts/batch. Groups
by (userId, activityType, startedAt +-90 s); prefers the canonical
source ladder (APPLE_HEALTH > WITHINGS > MANUAL > IMPORT) shared with
the read-time helper; breaks ties on calories > earliest createdAt >
input order. Losers surface to the iOS client as duplicate so the sync
cursor advances past them identically to the externalId case.

src/app/api/workouts/__tests__/batch-create.test.ts

@@ -225,8 +225,17 @@ describe("POST /api/workouts/batch — per-entry status envelope", () => {
     const res = await POST(
       makeRequest({
         workouts: [
+          // Two distinct runs separated by 2 h so the v1.4.42 W5
+          // write-time canonical-row picker treats them as separate
+          // workouts. Without the spacing the picker would collapse
+          // them to one survivor (same userId + sportType + startedAt
+          // inside the 90 s dedup window) and the envelope-count
+          // assertion would mask the per-entry status pin.
           validWorkout("uuid-fresh-1"),
-          validWorkout("uuid-fresh-2"),
+          validWorkout("uuid-fresh-2", {
+            startedAt: "2026-05-14T08:30:00.000Z",
+            endedAt: "2026-05-14T09:15:00.000Z",
+          }),
         ],
       }),
     );
@@ -253,8 +262,14 @@ describe("POST /api/workouts/batch — per-entry status envelope", () => {
     const res = await POST(
       makeRequest({
         workouts: [
+          // Same spacing rationale as the prior test — keep the two
+          // entries' `startedAt` outside the W5 dedup window so the
+          // pre-`createMany` pass doesn't collapse them.
           validWorkout("uuid-existing"),
-          validWorkout("uuid-fresh"),
+          validWorkout("uuid-fresh", {
+            startedAt: "2026-05-14T08:30:00.000Z",
+            endedAt: "2026-05-14T09:15:00.000Z",
+          }),
         ],
       }),
     );
@@ -284,8 +299,15 @@ describe("POST /api/workouts/batch — per-entry status envelope", () => {
     const res = await POST(
       makeRequest({
         workouts: [
+          // Same spacing rationale as the per-entry tests — keep both
+          // entries outside the W5 dedup window so the race-reconcile
+          // contract is exercised directly without the write-time
+          // canonical-picker absorbing one of the rows first.
           validWorkout("uuid-race-1"),
-          validWorkout("uuid-race-2"),
+          validWorkout("uuid-race-2", {
+            startedAt: "2026-05-14T08:30:00.000Z",
+            endedAt: "2026-05-14T09:15:00.000Z",
+          }),
         ],
       }),
     );

src/app/api/workouts/batch/route.ts

@@ -68,7 +68,8 @@ import {
   createBatchWorkoutSchema,
   MAX_WORKOUTS_PER_BATCH,
 } from "@/lib/validations/workout";
-import { Prisma } from "@/generated/prisma/client";
+import { pickCanonicalWorkoutRows } from "@/lib/workouts/canonical-rows";
+import { Prisma, type MeasurementSource } from "@/generated/prisma/client";
 
 // v1.4.25 W16c — push-suppression threshold for workout PRs. A batch
 // larger than this fires the detection job with `silent: true` so a
@@ -254,11 +255,56 @@ async function postBatch(request: NextRequest): Promise<Response> {
     };
   });
 
+  // v1.4.42 W5 — write-time cross-source dedup. The v1.5 iOS app
+  // drains a HealthKit observer queue that frequently carries the
+  // SAME logical workout from two paired sensors (Apple Watch +
+  // Withings ScanWatch). Without this pass, both rows persist (their
+  // `externalId`s differ → the `(userId, source, externalId)` unique
+  // index never fires) and the read-time canonical-picker has to
+  // carry the duplicate forward on every dashboard render.
+  //
+  // The helper is pure (no Prisma, no user lookups). It groups by
+  // `(userId, activityType, startedAt ± 90 s)`, prefers the canonical
+  // source ladder (APPLE_HEALTH > WITHINGS > MANUAL > IMPORT), and
+  // breaks ties on calories > earliest createdAt > input order. Rows
+  // dropped here surface to the iOS client as `duplicate` so the
+  // sync cursor advances past them identically to the externalId
+  // case below.
+  const canonicalPrepared = pickCanonicalWorkoutRows(
+    prepared.map((p) => ({
+      userId: p.row.userId,
+      activityType: p.row.sportType,
+      startedAt: p.row.startedAt as Date,
+      // Zod's `measurementSourceEnum.optional().default("MANUAL")`
+      // guarantees `source` is set post-parse, but Prisma's
+      // `WorkoutCreateManyInput` types it optional. Default through
+      // the same fallback the validator uses so the picker's source
+      // ladder still resolves on the rare row where TypeScript
+      // can't prove the union.
+      source: (p.row.source ?? "MANUAL") as MeasurementSource,
+      caloriesKcal: p.row.totalEnergyKcal ?? null,
+      // No createdAt yet — the row is pre-insert; the tie-breaker
+      // falls through to input order via the `index` field below.
+      index: p.index,
+    })),
+  );
+  const survivingIndices = new Set(canonicalPrepared.map((r) => r.index ?? -1));
+  const droppedByWriteDedup: number[] = [];
+  for (const p of prepared) {
+    if (!survivingIndices.has(p.index)) {
+      results[p.index] = { index: p.index, status: "duplicate" };
+      droppedByWriteDedup.push(p.index);
+    }
+  }
+  const survivors = prepared.filter((p) => survivingIndices.has(p.index));
+  let duplicateCount = droppedByWriteDedup.length;
   let insertedCount = 0;
-  let duplicateCount = 0;
 
-  if (prepared.length > 0) {
-    const dedupCandidates = prepared
+  if (survivors.length > 0) {
+    // v1.4.42 W5 — only the write-time dedup survivors are probed
+    // against existing rows; the dropped twins already carry a
+    // `duplicate` status and never reach the DB.
+    const dedupCandidates = survivors
       .map((p) => p.dedupKey)
       .filter((k): k is { source: string; externalId: string } => k !== null);
 
@@ -280,7 +326,7 @@ async function postBatch(request: NextRequest): Promise<Response> {
     );
 
     const toInsert: Prepared[] = [];
-    for (const p of prepared) {
+    for (const p of survivors) {
       const keyTuple = p.dedupKey
         ? `${p.dedupKey.source}::${p.dedupKey.externalId}`
         : null;
@@ -461,7 +507,10 @@ async function postBatch(request: NextRequest): Promise<Response> {
       const racedDuplicates = toInsert.length - insertedCount;
       if (racedDuplicates > 0) {
         let downgraded = 0;
-        for (const p of prepared) {
+        // v1.4.42 W5 — walk `survivors` (not `prepared`) so the
+        // already-downgraded-by-write-dedup losers don't get
+        // double-counted toward the race-reconcile budget.
+        for (const p of survivors) {
           if (downgraded >= racedDuplicates) break;
           if (results[p.index]?.status === "inserted") {
             results[p.index] = { index: p.index, status: "duplicate" };

src/lib/workouts/__tests__/canonical-rows.test.ts

@@ -0,0 +1,293 @@
+/**
+ * v1.4.42 W5 — write-time canonical-row picker tests.
+ *
+ * Pairs with `src/lib/workouts/canonical-rows.ts`. The helper is the
+ * pre-`createMany` dedup pass invoked by `POST /api/workouts/batch`
+ * when the v1.5 iOS sprint posts a batch carrying both an Apple Watch
+ * row and a Withings ScanWatch row for the same Sunday-morning run.
+ * The tests pin the algorithm contract so a future tightening of the
+ * window or a tie-breaker tweak surfaces a failing assertion rather
+ * than a silent regression in the duplicate-drop count.
+ */
+import { describe, expect, it } from "vitest";
+
+import {
+  pickCanonicalWorkoutRows,
+  WORKOUT_DEDUP_WINDOW_MS,
+  type WorkoutRow,
+} from "../canonical-rows";
+
+import type { MeasurementSource } from "@/generated/prisma/client";
+
+interface TestRow extends WorkoutRow {
+  /** Marker so the test can assert which input row survived. */
+  id: string;
+}
+
+function row(partial: Partial<TestRow> & { id: string }): TestRow {
+  return {
+    id: partial.id,
+    userId: partial.userId ?? "user-1",
+    activityType: partial.activityType ?? "running",
+    startedAt: partial.startedAt ?? new Date("2026-05-21T08:00:00Z"),
+    source: partial.source ?? "APPLE_HEALTH",
+    caloriesKcal: partial.caloriesKcal ?? null,
+    createdAt: partial.createdAt ?? null,
+    index: partial.index,
+  };
+}
+
+describe("pickCanonicalWorkoutRows", () => {
+  it("returns an empty array on empty input", () => {
+    expect(pickCanonicalWorkoutRows([])).toEqual([]);
+  });
+
+  it("collapses exact duplicates from the same source", () => {
+    // Two rows from the SAME source at the SAME instant: a re-posted
+    // batch with the externalId differing (e.g. the iOS app split the
+    // workout into two HKWorkout entries by mistake) would otherwise
+    // double-count. The helper keeps one survivor — the higher-cal /
+    // earliest-createdAt row.
+    const a = row({
+      id: "a",
+      source: "APPLE_HEALTH",
+      caloriesKcal: 420,
+      createdAt: new Date("2026-05-21T08:30:00Z"),
+    });
+    const b = row({
+      id: "b",
+      source: "APPLE_HEALTH",
+      caloriesKcal: 420,
+      createdAt: new Date("2026-05-21T08:31:00Z"),
+    });
+
+    const out = pickCanonicalWorkoutRows([a, b]);
+    expect(out).toHaveLength(1);
+    // Same calories → earliest createdAt wins (`a`).
+    expect(out[0]!.id).toBe("a");
+  });
+
+  it("prefers the higher-priority source on a cross-source overlap inside the window", () => {
+    // Apple Watch + Withings ScanWatch both record the same run. The
+    // window is satisfied (rows are 60 s apart, well under the 90 s
+    // ceiling) and the activityType matches → group dedups to one
+    // survivor. APPLE_HEALTH outranks WITHINGS per the ladder.
+    const apple = row({
+      id: "apple",
+      source: "APPLE_HEALTH",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+      caloriesKcal: 400,
+    });
+    const withings = row({
+      id: "withings",
+      source: "WITHINGS",
+      startedAt: new Date("2026-05-21T08:01:00Z"),
+      caloriesKcal: 500,
+    });
+
+    const out = pickCanonicalWorkoutRows([apple, withings]);
+    expect(out).toHaveLength(1);
+    // Source-axis wins outright — APPLE_HEALTH beats WITHINGS even
+    // when WITHINGS has higher calories. The calories tie-break only
+    // fires WITHIN a single source tier.
+    expect(out[0]!.id).toBe("apple");
+  });
+
+  it("returns every row unchanged when no overlap exists", () => {
+    // Two runs 10 min apart from the SAME source — well outside the
+    // 90 s window. Both must survive; the helper is not a generic
+    // "one run per day" filter.
+    const first = row({
+      id: "first",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+    });
+    const second = row({
+      id: "second",
+      startedAt: new Date("2026-05-21T08:10:00Z"),
+    });
+
+    const out = pickCanonicalWorkoutRows([first, second]);
+    expect(out.map((r) => r.id)).toEqual(["first", "second"]);
+  });
+
+  it("returns rows from different users untouched even on identical timestamps", () => {
+    // A multi-user batch (theoretically possible from an admin path)
+    // must never collapse rows across `userId`. The window check
+    // includes the userId in the grouping key.
+    const u1 = row({
+      id: "u1",
+      userId: "user-1",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+    });
+    const u2 = row({
+      id: "u2",
+      userId: "user-2",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+    });
+
+    const out = pickCanonicalWorkoutRows([u1, u2]);
+    expect(out.map((r) => r.id).sort()).toEqual(["u1", "u2"]);
+  });
+
+  it("returns rows of different activity types untouched at the same instant", () => {
+    // A user opens a "Walk" workout on the Watch and a "Strength"
+    // workout on the iPhone at the same instant. The window matches
+    // but the activityType doesn't — two distinct rows, both survive.
+    const walk = row({
+      id: "walk",
+      activityType: "walking",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+    });
+    const strength = row({
+      id: "strength",
+      activityType: "strength",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+    });
+
+    const out = pickCanonicalWorkoutRows([walk, strength]);
+    expect(out.map((r) => r.id).sort()).toEqual(["strength", "walk"]);
+  });
+
+  it("uses caloriesKcal as the tie-breaker when sources tie", () => {
+    // Apple Watch in the bag (no HR contact, calories=200) vs Apple
+    // Watch on the wrist (full HR series, calories=500). Same source,
+    // same instant, same activityType — calories wins.
+    const lo = row({
+      id: "lo",
+      source: "APPLE_HEALTH",
+      caloriesKcal: 200,
+    });
+    const hi = row({
+      id: "hi",
+      source: "APPLE_HEALTH",
+      caloriesKcal: 500,
+      // Stamp a later createdAt to prove the calories axis wins
+      // over the createdAt fallback.
+      createdAt: new Date("2026-05-21T09:00:00Z"),
+    });
+    const loWithEarlierCreated = row({
+      id: "lo-early",
+      source: "APPLE_HEALTH",
+      caloriesKcal: 200,
+      createdAt: new Date("2026-05-21T08:30:00Z"),
+    });
+
+    // Pair 1: hi vs lo → hi wins on calories.
+    expect(pickCanonicalWorkoutRows([lo, hi]).map((r) => r.id)).toEqual(["hi"]);
+
+    // Pair 2: hi vs lo-early → hi STILL wins because calories trump
+    // createdAt.
+    expect(
+      pickCanonicalWorkoutRows([loWithEarlierCreated, hi]).map((r) => r.id),
+    ).toEqual(["hi"]);
+  });
+
+  it("falls back to earliest createdAt when source + calories both tie", () => {
+    // Marc's ScanWatch on the wrist + ScanWatch in a desk drawer:
+    // unrealistic, but the case isolates the tie-breaker hierarchy.
+    const early = row({
+      id: "early",
+      source: "WITHINGS",
+      caloriesKcal: 350,
+      createdAt: new Date("2026-05-21T08:30:00Z"),
+    });
+    const late = row({
+      id: "late",
+      source: "WITHINGS",
+      caloriesKcal: 350,
+      createdAt: new Date("2026-05-21T08:45:00Z"),
+    });
+
+    expect(pickCanonicalWorkoutRows([late, early]).map((r) => r.id)).toEqual([
+      "early",
+    ]);
+  });
+
+  it("keeps determinism via input order when every field ties", () => {
+    // Two manually-entered "running" workouts at the same instant
+    // with no calories and no createdAt — the helper falls back to
+    // the original input ordinal so re-running over the same input
+    // always picks the same survivor.
+    const first = row({ id: "first", source: "MANUAL" });
+    const second = row({ id: "second", source: "MANUAL" });
+
+    expect(pickCanonicalWorkoutRows([first, second]).map((r) => r.id)).toEqual([
+      "first",
+    ]);
+    // Reversed input → reversed survivor. Confirms the tie-breaker
+    // genuinely consults input order rather than a hidden field
+    // (e.g. `id` lexicographic).
+    expect(pickCanonicalWorkoutRows([second, first]).map((r) => r.id)).toEqual([
+      "second",
+    ]);
+  });
+
+  it("treats a row exactly at the ±90 s boundary as in-window", () => {
+    // The window is inclusive (`<=`). A row landing on the boundary
+    // joins the group; one millisecond past stays separate.
+    const anchor = row({
+      id: "anchor",
+      source: "APPLE_HEALTH",
+      startedAt: new Date("2026-05-21T08:00:00Z"),
+    });
+    const onBoundary = row({
+      id: "boundary",
+      source: "WITHINGS",
+      startedAt: new Date(
+        anchor.startedAt.getTime() + WORKOUT_DEDUP_WINDOW_MS,
+      ),
+    });
+    const justOutside = row({
+      id: "outside",
+      source: "WITHINGS",
+      startedAt: new Date(
+        anchor.startedAt.getTime() + WORKOUT_DEDUP_WINDOW_MS + 1,
+      ),
+    });
+
+    // anchor + onBoundary → one group; APPLE_HEALTH wins.
+    expect(
+      pickCanonicalWorkoutRows([anchor, onBoundary]).map((r) => r.id),
+    ).toEqual(["anchor"]);
+
+    // anchor + justOutside → two distinct groups, both survive.
+    expect(
+      pickCanonicalWorkoutRows([anchor, justOutside]).map((r) => r.id),
+    ).toEqual(["anchor", "outside"]);
+  });
+
+  it("walks the full source ladder (APPLE_HEALTH > WITHINGS > MANUAL > IMPORT)", () => {
+    const sources: MeasurementSource[] = [
+      "IMPORT",
+      "MANUAL",
+      "WITHINGS",
+      "APPLE_HEALTH",
+    ];
+    const rows = sources.map((source, idx) =>
+      row({
+        id: source,
+        source,
+        // Stagger by 10 s so each row sits inside the 90 s window
+        // relative to the anchor (the first row inserted).
+        startedAt: new Date(
+          new Date("2026-05-21T08:00:00Z").getTime() + idx * 10_000,
+        ),
+      }),
+    );
+
+    expect(pickCanonicalWorkoutRows(rows).map((r) => r.id)).toEqual([
+      "APPLE_HEALTH",
+    ]);
+  });
+
+  it("does not mutate caller rows", () => {
+    // The picker is pure — the caller's array reference and row
+    // identity survive the call so a future audit-overlay can
+    // re-inspect the dropped rows.
+    const a = row({ id: "a", source: "APPLE_HEALTH", caloriesKcal: 100 });
+    const b = row({ id: "b", source: "WITHINGS", caloriesKcal: 100 });
+    const before = JSON.stringify([a, b]);
+    pickCanonicalWorkoutRows([a, b]);
+    expect(JSON.stringify([a, b])).toBe(before);
+  });
+});

src/lib/workouts/canonical-rows.ts

@@ -0,0 +1,259 @@
+/**
+ * v1.4.42 W5 — write-time cross-source canonical-row picker for
+ * incoming `Workout` ingest payloads.
+ *
+ * Why this exists alongside the read-time helper
+ * ──────────────────────────────────────────────
+ *
+ * Two related-but-distinct dedup paths exist in the codebase:
+ *
+ *   - `pickCanonicalWorkoutRows()` in `src/lib/measurements/` (v1.4.30)
+ *     — READ-time picker invoked by `GET /api/workouts` /
+ *     `GET /api/workouts/[id]`. It walks DB rows, looks up the user's
+ *     persisted `sourcePriorityJson`, buckets by 5-minute slot, and
+ *     filters the read result so the dashboard tile shows one row per
+ *     logical workout.
+ *
+ *   - This helper (v1.4.42) — WRITE-time picker invoked by
+ *     `POST /api/workouts/batch` BEFORE `prisma.workout.createMany`.
+ *     The v1.5 iOS sprint will ingest `HKWorkoutType` data in batches
+ *     against `/api/workouts/batch`; Apple Watch + Withings ScanWatch
+ *     paired to the same iPhone frequently capture the same workout
+ *     simultaneously (e.g. a Sunday-morning run) and the iOS app may
+ *     post both rows in one batch. Without write-time dedup, both
+ *     rows would persist (their `externalId`s differ, so the
+ *     `(userId, source, externalId)` unique index doesn't fire) and
+ *     the read-time picker would carry the duplicate forward at every
+ *     downstream read.
+ *
+ * The two pickers are layered, not redundant:
+ *   - Write-time picker drops PAYLOAD-internal duplicates inside a
+ *     single batch so we don't pay the storage cost.
+ *   - Read-time picker still runs to collapse cross-batch duplicates
+ *     (e.g. Apple Watch row from Monday's batch + ScanWatch row from
+ *     Tuesday's backfill batch).
+ *
+ * Algorithm
+ * ─────────
+ *
+ *   1. Group by `(userId, activityType, startedAt ± 90 s window)`.
+ *      The 90-second window is tighter than the read-time picker's
+ *      5-minute slot because a single batch from one iOS client has
+ *      no cross-device clock skew to absorb — both rows are stamped
+ *      with HealthKit's `startDate`, which Apple aggregates to the
+ *      same instant across paired sensors. A 90-second cushion
+ *      covers HK's documented ±60s smoothing plus a small buffer.
+ *   2. Within each group, prefer in order:
+ *      `APPLE_HEALTH ≻ WITHINGS ≻ MANUAL ≻ IMPORT`. The literal
+ *      source ladder is sourced from `DEFAULT_WORKOUT_SOURCE_PRIORITY`
+ *      in `src/lib/sources/pick-canonical-workout.ts` (the v1.4.25
+ *      W16a scaffolding) so both pickers consult a single source of
+ *      truth — adding a new source enum value is a one-place change.
+ *      Per the scope memo the legacy `STRAVA_IMPORT` reference maps
+ *      to the actual `IMPORT` enum value (the codebase has no
+ *      separate Strava source today; Strava XML imports ride the
+ *      generic `IMPORT` bucket).
+ *   3. Tie-breaker: higher `caloriesKcal` (more sensor data — a row
+ *      with HR-derived calories beats a row with a zero / null
+ *      total). Then earliest `createdAt` so re-running the picker
+ *      with the same inputs picks the same survivor every time.
+ *
+ * Purity contract
+ * ───────────────
+ *
+ * Zero Prisma imports, zero IO, zero user lookups. The helper takes
+ * an array, returns an array. Testable in isolation; safely
+ * invoked from a hot ingest path with no DB round-trip cost. The
+ * write-time path has the user's id on every row already and the
+ * source ladder is the canonical constant, so the helper doesn't
+ * need to consult `User.sourcePriorityJson` (that's a read-time
+ * concern: the user's display preference for the dashboard tile).
+ */
+
+import { DEFAULT_WORKOUT_SOURCE_PRIORITY } from "@/lib/sources/pick-canonical-workout";
+import type { MeasurementSource } from "@/generated/prisma/client";
+
+/**
+ * Minimum row shape consulted by the picker. Field names mirror the
+ * batch-route's prepared `Prisma.WorkoutCreateManyInput` plus a
+ * `createdAt` slot for the deterministic tie-breaker. Anything else
+ * the caller carried alongside the row (route geometry, HR series,
+ * metadata) survives untouched — the picker filters but never
+ * mutates and is generic in `T` so call sites keep their own
+ * carrier type.
+ *
+ * Naming aligned with the v1.5 iOS contract:
+ *   - `activityType` — the sport type (`"running"`, `"cycling"`,
+ *     ...). Matches `Workout.sportType` at the DB layer and
+ *     `WorkoutSportType` at the validator layer; cross-source
+ *     overlap requires identical sport so a "Walking" + "Running"
+ *     pair at the same instant stays as two distinct rows.
+ *   - `caloriesKcal` — the active-energy aggregate. The v1.5 iOS
+ *     contract uses `caloriesKcal` for parity with HealthKit's
+ *     `HKQuantityTypeIdentifierActiveEnergyBurned`; the server's
+ *     write path maps it to `Workout.totalEnergyKcal`.
+ *   - `createdAt` — when the row was minted. The batch path doesn't
+ *     have this yet (it's set by Prisma's `@default(now())` at
+ *     insert), so the helper falls back to insertion order via the
+ *     `index` field when `createdAt` is undefined on every member.
+ */
+export interface WorkoutRow {
+  userId: string;
+  activityType: string;
+  startedAt: Date;
+  source: MeasurementSource;
+  caloriesKcal?: number | null;
+  /**
+   * Optional row-mint timestamp. The write-time picker is called
+   * BEFORE `createMany`, so prepared rows usually have no
+   * `createdAt` yet — undefined falls through to the next
+   * tie-breaker (insertion order via the `index` field).
+   */
+  createdAt?: Date | null;
+  /**
+   * Stable per-batch ordinal. Set by the caller (e.g. the position
+   * of the entry in the inbound payload) so two rows with identical
+   * priority + calories + createdAt still resolve deterministically.
+   * Optional; defaults to 0 so single-row groups never need it.
+   */
+  index?: number;
+}
+
+/**
+ * Cross-source dedup window. Within a single ingest batch from one
+ * iOS client, two rows whose `startedAt` falls within ±90 s AND
+ * share the same `activityType` AND belong to the same user are
+ * treated as the same logical workout. The window is symmetric:
+ * the comparison uses `|a.startedAt - b.startedAt| ≤ 90 000 ms`.
+ */
+export const WORKOUT_DEDUP_WINDOW_MS = 90 * 1000;
+
+/**
+ * Source-priority rank lookup. Lower rank = higher priority. A
+ * source absent from the ladder gets a rank past the end of the
+ * array — the picker still keeps that row when no ladder-listed
+ * row competes for the slot (single-source group), but loses to
+ * any ladder-listed competitor.
+ */
+function sourceRank(source: MeasurementSource): number {
+  const idx = DEFAULT_WORKOUT_SOURCE_PRIORITY.indexOf(source);
+  return idx === -1 ? DEFAULT_WORKOUT_SOURCE_PRIORITY.length : idx;
+}
+
+/**
+ * Comparator: returns negative when `a` should win over `b`.
+ * Priority order: source rank → higher calories → earliest
+ * createdAt → earliest input index.
+ */
+function compareCandidates<T extends WorkoutRow>(a: T, b: T): number {
+  const sourceDelta = sourceRank(a.source) - sourceRank(b.source);
+  if (sourceDelta !== 0) return sourceDelta;
+
+  // Tie-breaker 1 — higher caloriesKcal wins (more sensor data).
+  // `null` / `undefined` calories sort below any numeric value.
+  const aCal = a.caloriesKcal ?? -1;
+  const bCal = b.caloriesKcal ?? -1;
+  if (aCal !== bCal) return bCal - aCal;
+
+  // Tie-breaker 2 — earliest `createdAt` wins (stable across
+  // re-runs). Undefined sorts below any concrete date so a
+  // pre-insert row (no createdAt yet) doesn't beat an already-
+  // persisted row carrying a real timestamp.
+  const aCreated = a.createdAt?.getTime() ?? Number.POSITIVE_INFINITY;
+  const bCreated = b.createdAt?.getTime() ?? Number.POSITIVE_INFINITY;
+  if (aCreated !== bCreated) return aCreated - bCreated;
+
+  // Tie-breaker 3 — input order. Guarantees determinism even when
+  // every other field ties (rare; same source, same calories,
+  // same / no createdAt). Default of 0 keeps single-row groups
+  // untouched.
+  return (a.index ?? 0) - (b.index ?? 0);
+}
+
+/**
+ * Pick canonical rows across sources for an inbound ingest batch.
+ *
+ * Group by `(userId, activityType, startedAt ± 90 s)`; within each
+ * group the comparator selects one survivor (the rest drop). No
+ * cross-source overlap → the row passes through. The returned
+ * array preserves the input order of the surviving rows so a
+ * caller that pre-sorted by `startedAt ASC` keeps that ordering
+ * downstream.
+ *
+ * Implementation walks the input once with a union-find-style
+ * pass: each row joins an existing group whose anchor falls
+ * within the dedup window AND matches `(userId, activityType)`,
+ * or seeds a new group. The anchor is the FIRST row that landed
+ * in the group; subsequent rows compare against the anchor so the
+ * window can't drift across an arbitrarily long chain of
+ * within-window-of-neighbour rows. This is O(n²) worst case but
+ * a typical batch is ≤ 100 entries so the constant is fine.
+ */
+export function pickCanonicalWorkoutRows<T extends WorkoutRow>(
+  rows: readonly T[],
+): T[] {
+  if (rows.length === 0) return [];
+
+  // Stamp each input with its original index so the comparator
+  // has a stable tie-breaker without mutating the caller's row.
+  type Tagged = { row: T; origIndex: number };
+  const tagged: Tagged[] = rows.map((row, origIndex) => ({ row, origIndex }));
+
+  // Group anchors: each group is a list of (row, origIndex). The
+  // anchor row sets the window comparison reference.
+  const groups: Tagged[][] = [];
+
+  for (const entry of tagged) {
+    let placed = false;
+    for (const group of groups) {
+      const anchor = group[0]!.row;
+      if (
+        anchor.userId === entry.row.userId &&
+        anchor.activityType === entry.row.activityType &&
+        Math.abs(anchor.startedAt.getTime() - entry.row.startedAt.getTime()) <=
+          WORKOUT_DEDUP_WINDOW_MS
+      ) {
+        group.push(entry);
+        placed = true;
+        break;
+      }
+    }
+    if (!placed) {
+      groups.push([entry]);
+    }
+  }
+
+  // Pick one survivor per group, then re-emit in original input
+  // order. The comparator sees `index` via the original input
+  // ordinal (overrides any caller-supplied `index` so the helper
+  // is self-contained — but a caller who set `index` explicitly
+  // still wins because we layer the explicit value on top below).
+  const survivors: Tagged[] = [];
+  for (const group of groups) {
+    if (group.length === 1) {
+      survivors.push(group[0]!);
+      continue;
+    }
+    // Annotate each member with its origIndex as the row's `index`
+    // for the comparator, but only when the caller didn't already
+    // set one. Caller-supplied indices outrank insertion order so
+    // a deliberate "I want this row first" hint survives.
+    const annotated = group.map((entry) => ({
+      ...entry,
+      row: {
+        ...entry.row,
+        index: entry.row.index ?? entry.origIndex,
+      } as T,
+    }));
+    annotated.sort((a, b) => compareCandidates(a.row, b.row));
+    // Preserve the picker's input-order survival by remembering the
+    // ORIGINAL row reference (not the annotated copy) so the caller
+    // gets back exactly what it passed in.
+    const winner = annotated[0]!;
+    const original = group.find((g) => g.origIndex === winner.origIndex)!;
+    survivors.push(original);
+  }
+
+  survivors.sort((a, b) => a.origIndex - b.origIndex);
+  return survivors.map((entry) => entry.row);
+}