refactor(measurements): move the batch ingest out of its route

postBatch was one 1255-line function. The route keeps authentication, the
rate limit, the body, the entry cap, the schema and the bridge-source
refusal; the ingest moves verbatim into lib/measurements/batch-ingest/, one
module per stage in the order they always ran: the first pass (prepare),
the measurement writes with their prefilters and set-based insert, the
hourly sums, the sync checkpoint, the audit and wide-event report, and the
cache and arrival tail. The route is 274 lines; no function in the folder
is over 200.

The parity walk recorded before the move (measurements-batch.parity) is
unchanged, as is the set-insert golden. The guards that read the route for
the per-entry external-id check, the plausibility validator, the write
tail, the dense-read bound and the unit audit now read the service files.

src/__tests__/apple-health-unit-audit-guard.test.ts

@@ -48,7 +48,7 @@ import {
   convertHkValue,
   isConvertibleHkUnit,
 } from "@/lib/measurements/hk-units";
-import { readSource } from "@/__tests__/helpers/source-files";
+import { readSource, readSourceFolder } from "@/__tests__/helpers/source-files";
 
 const ENTRIES = Object.values(APPLE_HEALTH_TYPE_MAP);
 
@@ -194,10 +194,17 @@ describe("Apple Health mapping — the record's own unit (issue #944)", () => {
       "utf8",
     );
     expect(archive).toContain(optIn);
-    const batch = readSource("src/app/api/measurements/batch/route.ts", [
-      "export const POST",
-      "hkIdentifier",
-    ]);
+    // The route parses; its ingest maps the entries in the service folder.
+    const batch = [
+      readSource("src/app/api/measurements/batch/route.ts", [
+        "export const POST",
+        "hkIdentifier",
+      ]),
+      ...readSourceFolder("src/lib/measurements/batch-ingest", {
+        floor: 5,
+      }).map(([, text]) => text),
+    ].join("\n");
+    expect(batch).toContain("mapAppleHealthEntry(");
     expect(
       batch,
       "the batch route opted into record-unit conversion",

src/__tests__/dense-measurement-read-guard.test.ts

@@ -64,6 +64,9 @@ const SCOPED_DIRS = [
   "lib/export",
   "lib/targets/",
   "lib/doctor-report",
+  // Request-handler logic moved out of its route into a service keeps the
+  // route's scope.
+  "lib/measurements/batch-ingest/",
 ];
 
 /** Types a device writes at sampling rate rather than a few times a day. */
@@ -103,7 +106,7 @@ const ALLOWED: Record<string, string> = {
     "Health-score inputs: steps (drained to one row per day nightly), sleep, waist, weight, blood pressure, fasting glucose only.",
   'app/api/analytics/route.ts::"BLOOD_GLUCOSE"':
     "The 30-day per-context glucose summary, the same window and bound as the dashboard and coach glucose panels: at a sensor's fixed rate, below 45 000 rows.",
-  "app/api/measurements/batch/route.ts::p.row.type as MeasurementType":
+  "lib/measurements/batch-ingest/write-measurements.ts::p.row.type as MeasurementType":
     "Cross-source merge probe: one ±tolerance window around each posted reading's instant, so the rows found are bounded by the batch size.",
   "app/api/measurements/route.ts::(shorthand)":
     "Cross-source merge probe for one posted sleep segment: SLEEP_DURATION in a ±tolerance window around its end.",

src/__tests__/external-id-stability-guard.test.ts

@@ -394,17 +394,28 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => {
   // Each of these drains a client sync queue. A schema-level refusal here
   // would stop a whole batch of good rows because of one bad one, which is
   // worse than the duplicate it prevents.
-  const BATCH_ROUTES = [
-    "app/api/measurements/batch/route.ts",
-    "app/api/mood-entries/bulk/route.ts",
-    "app/api/medications/intake/bulk/route.ts",
-    "app/api/workouts/batch/route.ts",
-    "app/api/cycle/day-logs/bulk/route.ts",
-    "app/api/import/route.ts",
-  ];
+  //
+  // A surface whose ingest moved into a service names the service files
+  // beside the route: the route parses, the service refuses per entry.
+  const BATCH_SURFACES: Record<string, readonly string[]> = {
+    "app/api/measurements/batch/route.ts": [
+      "lib/measurements/batch-ingest/prepare.ts",
+      "lib/measurements/batch-ingest/ingest.ts",
+    ],
+    "app/api/mood-entries/bulk/route.ts": [],
+    "app/api/medications/intake/bulk/route.ts": [],
+    "app/api/workouts/batch/route.ts": [],
+    "app/api/cycle/day-logs/bulk/route.ts": [],
+    "app/api/import/route.ts": [],
+  };
+  const BATCH_ROUTES = Object.keys(BATCH_SURFACES);
+
+  /** The route and the service files that carry its per-entry ingest. */
+  const surface = (rel: string): string =>
+    [rel, ...BATCH_SURFACES[rel]].map((file) => body(file)).join("\n");
 
   it.each(BATCH_ROUTES)("%s classifies each entry itself", (rel) => {
-    const source = body(rel);
+    const source = surface(rel);
     expect(source).toContain("classifyExternalId");
     expect(source).toContain("unstableExternalIdMeta");
   });
@@ -414,7 +425,7 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => {
     (rel) => {
       // The JSON importer has no per-entry status envelope — it reports a
       // skipped COUNT — so it is the one batch surface without the code.
-      expect(body(rel)).toContain("UNSTABLE_EXTERNAL_ID_REASON");
+      expect(surface(rel)).toContain("UNSTABLE_EXTERNAL_ID_REASON");
     },
   );
 });

src/__tests__/measurement-write-plausibility-guard.test.ts

@@ -73,10 +73,13 @@ const WRITERS: Record<string, Disposition> = {
     validatedIn: "lib/validations/measurement.ts",
   },
   "app/api/import/route.ts": { kind: "validator" },
-  // v1.44 — the set-based insert writes rows that already passed the
-  // route's `validateMeasurementRange` per entry; everything else goes
-  // through the reconciler, which gates itself.
-  "app/api/measurements/batch/route.ts": { kind: "validator" },
+  // The batch ingest's set-based insert writes rows that already passed
+  // `validateMeasurementRange` per entry in its first pass; everything else
+  // goes through the reconciler, which gates itself.
+  "lib/measurements/batch-ingest/write-measurements.ts": {
+    kind: "upstream",
+    validatedIn: "lib/measurements/batch-ingest/prepare.ts",
+  },
   "lib/mcp/writes.ts": { kind: "validator" },
   "lib/measurements/create-from-telegram.ts": { kind: "validator" },
   "lib/measurements/import-apple-health-export.ts": { kind: "validator" },

src/__tests__/measurement-write-tail-guard.test.ts

@@ -32,6 +32,13 @@ const WRITE_RE =
 /** Runs `afterMeasurementMutation` on the rows it wrote. */
 const TAIL = "runs afterMeasurementMutation";
 
+/**
+ * Writes here; the module named runs `afterMeasurementMutation` on the same
+ * rows (a service split into stages, whose write and tail are siblings).
+ */
+const TAIL_VIA_PREFIX = "runs afterMeasurementMutation via ";
+const TAIL_VIA = (file: string) => `${TAIL_VIA_PREFIX}${file}`;
+
 /**
  * Every file with a measurement write, and how it keeps the two readers
  * current. A file listed here must still contain a write (a stale entry
@@ -44,7 +51,9 @@ const WRITE_SITES: Record<string, string> = {
   "app/api/measurements/bulk-delete/route.ts": TAIL,
   "app/api/measurements/by-external-ids/route.ts": TAIL,
   "app/api/measurements/restore/route.ts": TAIL,
-  "app/api/measurements/batch/route.ts": TAIL,
+  "lib/measurements/batch-ingest/write-measurements.ts": TAIL_VIA(
+    "lib/measurements/batch-ingest/tail.ts",
+  ),
   "app/api/measurements/route.ts": TAIL,
   "app/api/mental-health/assessments/route.ts": TAIL,
   "lib/insights/score-row.ts": TAIL,
@@ -157,8 +166,14 @@ describe("measurement write tail", () => {
 
   it("the files that claim the shared tail call it", () => {
     const missing = Object.entries(WRITE_SITES)
-      .filter(([, reason]) => reason === TAIL)
-      .map(([file]) => file)
+      .filter(
+        ([, reason]) => reason === TAIL || reason.startsWith(TAIL_VIA_PREFIX),
+      )
+      .map(([file, reason]) =>
+        reason.startsWith(TAIL_VIA_PREFIX)
+          ? reason.slice(TAIL_VIA_PREFIX.length)
+          : file,
+      )
       .filter(
         (file) =>
           !/afterMeasurementMutation\s*\(/.test(

src/lib/measurements/batch-ingest/entries.ts

@@ -0,0 +1,115 @@
+/**
+ * The vocabulary of the native batch ingest (`POST /api/measurements/batch`):
+ * what one posted entry carries, what the ingest answers for it, and the two
+ * small rules more than one stage needs.
+ *
+ * The wire schema itself stays in the route, which owns the contract; the
+ * entry type here is what that schema parses to, and the route assigns one to
+ * the other so the compiler holds them together.
+ */
+import type { Prisma } from "@/generated/prisma/client";
+import type { HealthKitRequestSyncTrigger } from "@/lib/integrations/healthkit-type-sync";
+import { getUnitForType, VALUE_RANGES } from "@/lib/validations/measurement";
+import type { DeviceType } from "@/lib/validations/source-priority";
+
+/** One HealthKit sample (or `stats:` aggregate) as the route parsed it. */
+export interface BatchIngestEntry {
+  hkIdentifier: string;
+  value: number;
+  unit: string;
+  startDate: string;
+  endDate: string;
+  sleepStage?: number;
+  categoryValue?: number;
+  valueMin?: number;
+  valueMax?: number;
+  externalId: string;
+  externalSourceVersion?: string;
+  source?: "APPLE_HEALTH" | "MANUAL";
+  deviceType?: DeviceType | null;
+}
+
+/** The parsed request: the entries and the client's declared wake reason. */
+export interface BatchIngestInput {
+  user: { id: string; timezone: string | null };
+  /** A narrow single-purpose Bearer (a bridge), not the native client. */
+  scoped: boolean;
+  entries: BatchIngestEntry[];
+  syncTrigger: HealthKitRequestSyncTrigger | undefined;
+  ipAddress: string | null;
+}
+
+/**
+ * Per-entry outcome the iOS client uses to advance its sync cursor.
+ * `inserted` and `duplicate` both indicate the row landed (or was
+ * already present), so the client can checkpoint past them. `updated`
+ * is the per-day-aggregate overwrite path: when the iOS client re-posts
+ * a `stats:*` cumulative row (Steps, Active Energy, Sleep Duration,
+ * Walking/Running Distance, Flights Climbed — see issue #213) for a day
+ * we already have, the server overwrites the row's `value` rather than
+ * dropping the new payload as a duplicate. Sample-class rows (every
+ * other HK metric) keep the strict `duplicate` semantics because each
+ * sample is a canonical, immutable reading. `skipped` is a validation no-op;
+ * `failed` is a hard database verdict and must not be checkpointed as landed.
+ */
+export type EntryStatus =
+  "inserted" | "updated" | "duplicate" | "skipped" | "failed";
+
+export interface EntryResult {
+  index: number;
+  status: EntryStatus;
+  reason?: string;
+  /**
+   * Only on a `value_out_of_range` skip: the value after conversion into the
+   * stored unit, which is what the band was checked against.
+   */
+  convertedValue?: number;
+  /** Only on a `value_out_of_range` skip: the band applied, in the stored unit. */
+  range?: { min: number; max: number; unit: string };
+}
+
+/** An entry that passed the first pass, with the row it becomes. */
+export interface PreparedEntry {
+  index: number;
+  entry: BatchIngestEntry;
+  row: Prisma.MeasurementCreateManyInput;
+}
+
+// Issue #213 — `stats:*` externalIds carry per-day cumulative
+// totals that the iOS HealthKit observer re-posts as the day progresses.
+// Treating those as immutable duplicates freezes the day's tile at the
+// first-sync value. Sample-class externalIds (every other prefix) keep
+// the strict insert-only contract.
+const STATS_EXTERNAL_ID_PREFIX = "stats:";
+
+export function isStatsExternalId(
+  externalId: string | null | undefined,
+): boolean {
+  return (
+    typeof externalId === "string" &&
+    externalId.startsWith(STATS_EXTERNAL_ID_PREFIX)
+  );
+}
+
+/**
+ * The `value_out_of_range` skip, with what a client needs to say which limit
+ * applied. The value is the entry's own reading after conversion, handed back
+ * to the account that sent it; a fraction sent where a percent was expected
+ * shows up here as a converted value a hundred times too small or too large.
+ */
+export function outOfRangeResult(
+  index: number,
+  type: string,
+  convertedValue: number,
+): EntryResult {
+  const band = VALUE_RANGES[type];
+  return {
+    index,
+    status: "skipped",
+    reason: "value_out_of_range",
+    ...(Number.isFinite(convertedValue) ? { convertedValue } : {}),
+    ...(band
+      ? { range: { min: band.min, max: band.max, unit: getUnitForType(type) } }
+      : {}),
+  };
+}

src/lib/measurements/batch-ingest/ingest.ts

@@ -0,0 +1,125 @@
+/**
+ * The native batch ingest behind `POST /api/measurements/batch`, from a
+ * parsed and admitted request to the per-entry answer.
+ *
+ * The stages run in a fixed order, and the order is part of the contract:
+ * the first pass maps every entry; the measurements are written, then the
+ * hourly sums; the sync checkpoint is stamped from the verdicts; the audit
+ * row, the wide-event breakdowns and the record jobs follow; the caches and
+ * the arrival spine hear last. The route owns authentication, the rate
+ * limit, the body and the response envelope.
+ */
+import { annotate } from "@/lib/logging/context";
+import { unstableExternalIdMeta } from "@/lib/validations/external-id";
+
+import type { BatchIngestInput, EntryResult } from "./entries";
+import { prepareBatch } from "./prepare";
+import {
+  annotateIngest,
+  enqueueRecordJobs,
+  reportBatchOutcome,
+  summariseSkips,
+  type BatchTallies,
+  type SkippedEntry,
+} from "./report";
+import { stampSyncCheckpoint } from "./sync-checkpoint";
+import { afterBatchWrite } from "./tail";
+import { writeBatchHours } from "./write-hours";
+import { MeasurementBatchWriter } from "./write-measurements";
+
+/** The response body of the batch route. */
+export interface BatchIngestResult {
+  processed: number;
+  inserted: number;
+  updated: number;
+  duplicates: number;
+  failed: number;
+  skipped: SkippedEntry[];
+  entries: EntryResult[];
+}
+
+export async function ingestMeasurementBatch(
+  input: BatchIngestInput,
+): Promise<BatchIngestResult> {
+  const { user, scoped, entries, syncTrigger, ipAddress } = input;
+
+  const batch = await prepareBatch(input);
+  const { prepared, results } = batch;
+  if (batch.unstableShapes.length > 0) {
+    annotate({
+      meta: unstableExternalIdMeta("measurement.batch", batch.unstableShapes),
+    });
+  }
+
+  const writer = new MeasurementBatchWriter(user.id, results);
+  await writer.write(prepared);
+
+  const hours = await writeBatchHours({
+    userId: user.id,
+    hourWrites: batch.hourWrites,
+    accountTz: batch.accountTz,
+    results,
+    processed: entries.length,
+  });
+  writer.duplicateCount += hours.duplicates;
+  writer.failedCount += hours.failed;
+
+  await stampSyncCheckpoint({
+    userId: user.id,
+    scoped,
+    syncTrigger,
+    prepared,
+    results,
+    failedCount: writer.failedCount,
+    unchangedStatsRows: writer.unchangedStatsRows,
+    hourLanded: hours.hourLanded,
+    hourChanged: hours.hourChanged,
+  });
+
+  const tallies: BatchTallies = writer;
+  const { skipped, skippedByReason } = summariseSkips(results);
+  await reportBatchOutcome({
+    userId: user.id,
+    ipAddress,
+    entries,
+    results,
+    tallies,
+    skipped,
+  });
+  await enqueueRecordJobs({
+    userId: user.id,
+    processed: entries.length,
+    tallies,
+  });
+  annotateIngest({
+    processed: entries.length,
+    tallies,
+    skipped,
+    skippedByReason,
+    syncTrigger,
+  });
+
+  await afterBatchWrite({
+    userId: user.id,
+    syncTrigger,
+    prepared,
+    results,
+    insertedCount: writer.insertedCount,
+    updatedCount: writer.updatedCount,
+    insertedPrepared: writer.insertedPrepared,
+    writtenIdentities: writer.writtenIdentities,
+    hourInsertedCount: hours.hourInsertedCount,
+    hourUpdatedCount: hours.hourUpdatedCount,
+  });
+
+  return {
+    processed: entries.length,
+    inserted: writer.insertedCount + hours.hourInsertedCount,
+    updated:
+      writer.updatedCount + writer.unchangedStatsCount + hours.hourUpdatedCount,
+    duplicates: writer.duplicateCount,
+    failed: writer.failedCount,
+    skipped,
+    entries: results,
+  };
+}

src/lib/measurements/batch-ingest/prepare.ts

@@ -0,0 +1,310 @@
+/**
+ * First pass of the batch ingest: map each posted entry to the row it would
+ * become, or to the per-entry refusal that keeps it out. Nothing is written
+ * here; hourly sums are validated and set aside for their own write.
+ */
+import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source";
+import {
+  isAggregatedBucketExternalId,
+  mapAppleHealthEntry,
+  targetsAggregatedBucket,
+} from "@/lib/measurements/apple-health-mapping";
+import {
+  classifyHourBucket,
+  HOUR_BUCKET_SKIP_REASONS,
+  parseHourBucketId,
+  type HourBucketEntry,
+  type HourBucketId,
+} from "@/lib/measurements/intraday-hour-buckets";
+import { resolveStoredTimezone } from "@/lib/tz/resolver";
+import { isPlausibleEntryInstant } from "@/lib/validations/entry-instant";
+import {
+  classifyExternalId,
+  UNSTABLE_EXTERNAL_ID_REASON,
+  type UnstableExternalIdShape,
+} from "@/lib/validations/external-id";
+import { validateMeasurementRange } from "@/lib/validations/measurement";
+
+import {
+  outOfRangeResult,
+  type BatchIngestEntry,
+  type BatchIngestInput,
+  type EntryResult,
+  type PreparedEntry,
+} from "./entries";
+
+export interface PreparedBatch {
+  prepared: PreparedEntry[];
+  results: EntryResult[];
+  unstableShapes: UnstableExternalIdShape[];
+  /** Hourly sums for the intraday profile, written after the measurements. */
+  hourWrites: HourBucketEntry[];
+  /** The zone the hours are cut in, resolved when the batch carried one. */
+  accountTz: string | undefined;
+}
+
+export async function prepareBatch(
+  input: Pick<BatchIngestInput, "user" | "scoped" | "entries">,
+): Promise<PreparedBatch> {
+  const { user, scoped, entries } = input;
+  const batch: PreparedBatch = {
+    prepared: [],
+    results: new Array(entries.length),
+    unstableShapes: [],
+    hourWrites: [],
+    accountTz: undefined,
+  };
+  const batchNow = new Date();
+
+  for (let index = 0; index < entries.length; index++) {
+    const entry = entries[index];
+
+    // The `(user_id, type, source, external_id)` dedup key above is only
+    // idempotent while the id is STABLE across client launches. An id
+    // that rotates per process (an object description carrying a memory
+    // address) never matches its own earlier row, so every sync sweep
+    // mints a fresh measurement. Refused per entry, like the range and
+    // timestamp guards below — one malformed row must not stop the other
+    // 499 from landing.
+    const unstable = classifyExternalId(entry.externalId);
+    if (unstable) {
+      batch.unstableShapes.push(unstable);
+      batch.results[index] = {
+        index,
+        status: "skipped",
+        reason: UNSTABLE_EXTERNAL_ID_REASON,
+      };
+      continue;
+    }
+
+    // An hourly sum of a summed type (`stats:<HK>:<YYYY-MM-DD>T<HH>`, ios #77)
+    // is never a measurement: it lands in the intraday profile behind the
+    // same-time baseline, and the day's total stays the day row. Validated
+    // here, written after the measurements.
+    const hourId = parseHourBucketId(entry.externalId);
+    if (hourId) {
+      await prepareHourEntry(batch, {
+        index,
+        entry,
+        hourId,
+        user,
+        scoped,
+        batchNow,
+      });
+      continue;
+    }
+
+    const prepared = prepareMeasurementEntry(index, entry, user.id, scoped);
+    if ("row" in prepared) batch.prepared.push(prepared);
+    else batch.results[index] = prepared;
+  }
+  return batch;
+}
+
+/** Validate one hourly sum and set it aside, or record why it is refused. */
+async function prepareHourEntry(
+  batch: PreparedBatch,
+  args: {
+    index: number;
+    entry: BatchIngestEntry;
+    hourId: HourBucketId;
+    user: BatchIngestInput["user"];
+    scoped: boolean;
+    batchNow: Date;
+  },
+): Promise<void> {
+  const { index, entry, hourId, user, scoped, batchNow } = args;
+  const skip = (reason: string) => {
+    batch.results[index] = { index, status: "skipped", reason };
+  };
+  // The profile has no source dimension and is the phone's: a bridge or a
+  // hand-entered row cannot post hours into it.
+  if (scoped || (entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") {
+    skip(HOUR_BUCKET_SKIP_REASONS.unsupportedType);
+    return;
+  }
+  batch.accountTz ??= await resolveStoredTimezone(user.timezone);
+  const startDate = new Date(entry.startDate);
+  const verdict = classifyHourBucket(
+    hourId,
+    startDate,
+    batch.accountTz,
+    batchNow,
+  );
+  if (!verdict.ok) {
+    skip(verdict.reason);
+    return;
+  }
+  if (!isPlausibleEntryInstant(startDate)) {
+    skip("implausible_timestamp");
+    return;
+  }
+  const hourMapped = mapAppleHealthEntry({
+    hkIdentifier: entry.hkIdentifier,
+    value: entry.value,
+    unit: entry.unit,
+    startDate: entry.startDate,
+    endDate: entry.endDate,
+  });
+  if (!hourMapped || hourMapped.type !== verdict.type) {
+    // The entry's identifier must be the one its id names.
+    skip(
+      hourMapped
+        ? HOUR_BUCKET_SKIP_REASONS.unsupportedType
+        : "unmappable_identifier",
+    );
+    return;
+  }
+  if (
+    hourMapped.value < 0 ||
+    validateMeasurementRange(hourMapped.type, hourMapped.value) !== null
+  ) {
+    batch.results[index] = outOfRangeResult(
+      index,
+      hourMapped.type,
+      hourMapped.value,
+    );
+    return;
+  }
+  batch.hourWrites.push({
+    index,
+    type: verdict.type,
+    hkIdentifier: hourId.hkIdentifier,
+    dateKey: hourId.dateKey,
+    hour: hourId.hour,
+    value: hourMapped.value,
+    startMs: startDate.getTime(),
+  });
+}
+
+/** Map one entry to its measurement row, or to the refusal that keeps it out. */
+function prepareMeasurementEntry(
+  index: number,
+  entry: BatchIngestEntry,
+  userId: string,
+  scoped: boolean,
+): PreparedEntry | EntryResult {
+  const mapped = mapAppleHealthEntry({
+    hkIdentifier: entry.hkIdentifier,
+    value: entry.value,
+    unit: entry.unit,
+    startDate: entry.startDate,
+    endDate: entry.endDate,
+    sleepStage: entry.sleepStage,
+    categoryValue: entry.categoryValue,
+    valueMin: entry.valueMin,
+    valueMax: entry.valueMax,
+  });
+
+  if (!mapped) {
+    return {
+      index,
+      status: "skipped",
+      reason: "unmappable_identifier",
+    };
+  }
+
+  // Plausibility-range guard. We tolerate "out of plausible range"
+  // by skipping rather than failing the batch — a single bogus
+  // sensor reading shouldn't poison an otherwise clean ingest.
+  const rangeError = validateMeasurementRange(mapped.type, mapped.value);
+  if (rangeError !== null) {
+    return outOfRangeResult(index, mapped.type, mapped.value);
+  }
+
+  // Time-plausibility guard, the bound the single-entry twin already applies
+  // through `validateEntryInstant` on `createMeasurementSchema.measuredAt`.
+  // Without it a future-dated row permanently won "latest reading" on every
+  // dashboard and poisoned the rollup buckets and the canonical picker. The
+  // past floor stays at 1900 (no `maxAgeMs`) because a historical Apple
+  // Health backfill is a legitimate, expected shape on this route; only the
+  // future side is a bug. Skipped per entry, like the range guard above.
+  if (!isPlausibleEntryInstant(mapped.takenAt)) {
+    return {
+      index,
+      status: "skipped",
+      reason: "implausible_timestamp",
+    };
+  }
+
+  // The go-forward aggregated heart-rate wire contract (iOS #34). A
+  // `stats:HKQuantityTypeIdentifierHeartRate:<10-min-ISO-Z>` row is the
+  // 10-minute-average PULSE bucket; it rides the generic `stats:` overwrite
+  // path so a within-bucket re-post (the running mean shifts) replaces the
+  // value instead of duplicating. Reject a row that targets a bucket prefix
+  // but carries a malformed suffix — a garbage suffix would mint an
+  // un-overwriteable row the next re-post can't collapse onto. A well-formed
+  // bucket (and every non-bucket `stats:` / per-sample `uuid` externalId)
+  // passes through unchanged.
+  if (
+    targetsAggregatedBucket(entry.externalId) &&
+    !isAggregatedBucketExternalId(entry.externalId)
+  ) {
+    return {
+      index,
+      status: "skipped",
+      reason: "malformed_hr_bucket_id",
+    };
+  }
+
+  // The per-bucket MIN / MAX are persisted ONLY on a well-formed 10-min
+  // aggregated bucket row. A per-sample reading, a per-day cumulative
+  // `stats:` total, or a manual entry never carries a spread, so we pin them
+  // to null there even if a client mis-sends the fields — `value` stays the
+  // single source of truth for those rows.
+  //
+  // The spread is validated the same way `value` is guarded (above): it
+  // drives the DAY min/max band + the intraday envelope, so an out-of-range
+  // or mis-ordered spread (a spurious `discreteMin` of 0, a sensor glitch)
+  // would poison the aggregate exactly like a bogus `value` would. Persist
+  // the spread ONLY when both extremes are present, in range, and correctly
+  // ordered around `value`; otherwise drop to null and keep the trustworthy
+  // average. Partial or invalid spread → avg-only, still valid.
+  const isHrBucket = isAggregatedBucketExternalId(entry.externalId);
+  const spreadValid =
+    isHrBucket &&
+    typeof mapped.valueMin === "number" &&
+    typeof mapped.valueMax === "number" &&
+    validateMeasurementRange(mapped.type, mapped.valueMin) === null &&
+    validateMeasurementRange(mapped.type, mapped.valueMax) === null &&
+    mapped.valueMin <= mapped.value &&
+    mapped.value <= mapped.valueMax;
+  const valueMin = spreadValid ? mapped.valueMin : null;
+  const valueMax = spreadValid ? mapped.valueMax : null;
+
+  return {
+    index,
+    entry,
+    row: {
+      userId,
+      type: mapped.type,
+      value: mapped.value,
+      valueMin,
+      valueMax,
+      unit: mapped.unit,
+      // Honour the per-entry source tag, defaulting to `APPLE_HEALTH` when
+      // absent so legacy callers are unchanged.
+      //
+      // A scoped credential gets `EXTERNAL`, and can hold no other value: the
+      // route refused any body that named one, so nothing asserted is being
+      // discarded here. The value has to move with the credential rather
+      // than with the route, because the route serves both the phone and a
+      // third-party bridge. It also moves the dedup key to
+      // `(userId, type, EXTERNAL, externalId)`, which is the isolated
+      // namespace this source exists to give.
+      source: scoped ? EXTERNAL_SOURCE : (entry.source ?? "APPLE_HEALTH"),
+      measuredAt: mapped.takenAt,
+      externalId: entry.externalId,
+      externalSourceVersion: entry.externalSourceVersion ?? null,
+      sleepStage: mapped.sleepStage ?? null,
+      // Categorical events: the device's own verdict / severity for an EVENT
+      // row; null for every continuous reading.
+      rhythmClassification: mapped.rhythmClassification ?? null,
+      // The iOS-supplied device-type, passed through to the row. Stays null
+      // for builds that do not send it; the canonical picker treats null as
+      // `unknown` so legacy ingest keeps working without a server-side
+      // default.
+      deviceType: entry.deviceType ?? null,
+    },
+  };
+}

src/lib/measurements/batch-ingest/report.ts

@@ -0,0 +1,258 @@
+/**
+ * What a batch reports once its rows are written: the audit row, the
+ * operator's wide-event breakdowns, and the record-detection and reminder
+ * jobs a written row starts.
+ */
+import { fireAndForget } from "@/lib/logging/fire-and-forget";
+import { annotate } from "@/lib/logging/context";
+import { auditLog } from "@/lib/auth/audit";
+import { targetsAggregatedBucket } from "@/lib/measurements/apple-health-mapping";
+import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
+import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import type { HealthKitRequestSyncTrigger } from "@/lib/integrations/healthkit-type-sync";
+
+import type { BatchIngestEntry, EntryResult } from "./entries";
+
+// Historical-backfill threshold for PR push suppression. A batch larger than
+// this fires the detection job with `silent: true` so a multi-year Apple
+// Health backfill writes records without spamming the user with hundreds of
+// pushes. Tuned generously — a healthy daily-sync batch is typically well
+// under 50 entries.
+const PR_DETECTION_SILENT_THRESHOLD = 50;
+
+/** The measurement-side tallies the report reads. */
+export interface BatchTallies {
+  insertedCount: number;
+  updatedCount: number;
+  unchangedStatsCount: number;
+  duplicateCount: number;
+  failedCount: number;
+  crossSourceMergedCount: number;
+  foldedWindowCount: number;
+  setInsertedCount: number;
+}
+
+export type SkippedEntry = { index: number; reason: string };
+
+/** The skipped entries, and how many tripped each reason. */
+export function summariseSkips(results: EntryResult[]): {
+  skipped: SkippedEntry[];
+  skippedByReason: Record<string, number>;
+} {
+  const skipped = results
+    .filter((r) => r.status === "skipped")
+    .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" }));
+
+  // Per-reason skip breakdown (issue #585). The ingest annotation only ever
+  // carried the total `skipped` count, so an operator investigating "why does
+  // this account's uploaded data look sparser than expected" had no way to
+  // see WHICH validation gate an entry tripped (malformed bucket id,
+  // out-of-range value, unmappable identifier, …) without pulling raw rows.
+  // Grouping by reason turns that into a one-glance wide-event field.
+  const skippedByReason: Record<string, number> = {};
+  for (const s of skipped) {
+    skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1;
+  }
+  return { skipped, skippedByReason };
+}
+
+/** The audit row and the per-feature wide-event breakdowns. */
+export async function reportBatchOutcome(args: {
+  userId: string;
+  ipAddress: string | null;
+  entries: BatchIngestEntry[];
+  results: EntryResult[];
+  tallies: BatchTallies;
+  skipped: SkippedEntry[];
+}): Promise<void> {
+  const { userId, ipAddress, entries, results, tallies, skipped } = args;
+  await auditLog("measurement.batch.ingest", {
+    userId,
+    ipAddress,
+    details: {
+      processed: entries.length,
+      inserted: tallies.insertedCount,
+      updated: tallies.updatedCount + tallies.unchangedStatsCount,
+      duplicates: tallies.duplicateCount,
+      skipped: skipped.length,
+      failed: tallies.failedCount,
+    },
+  });
+
+  // Issue #213 — a dedicated wide-event annotation so an operator can grep
+  // `measurement.batch.stats-overwrite` to see how often per-day cumulative
+  // rows are getting re-posted (a healthy ingest flow). Only fires when at
+  // least one row was overwritten so the baseline ingest trace is unchanged
+  // for batches with no `stats:*` duplicates.
+  if (tallies.updatedCount > 0) {
+    annotate({
+      action: { name: "measurement.batch.stats-overwrite" },
+      meta: {
+        updated: tallies.updatedCount,
+        processed: entries.length,
+      },
+    });
+  }
+
+  // iOS #2 — a dedicated annotation for the MANUAL↔APPLE_HEALTH same-reading
+  // collapse so an operator can grep `measurement.batch.cross-source-merge`
+  // to confirm the standalone-pair mirror duplicate is being absorbed (a
+  // healthy first-pair flow). Only fires when at least one row was merged so
+  // the baseline ingest trace is unchanged for the common case.
+  if (tallies.crossSourceMergedCount > 0) {
+    annotate({
+      action: { name: "measurement.batch.cross-source-merge" },
+      meta: {
+        merged: tallies.crossSourceMergedCount,
+        processed: entries.length,
+      },
+    });
+  }
+
+  annotateHeartRateBuckets(entries, results);
+}
+
+/**
+ * Issue #585 — a dedicated outcome breakdown for entries targeting the
+ * go-forward aggregated 10-min heart-rate bucket prefix
+ * (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday pulse
+ * chart reads exactly these rows for a live-synced day; a report of "the
+ * chart looks sparser after the ZIP import cutoff" is unanswerable from the
+ * baseline ingest trace alone, since it never distinguished an HR-bucket
+ * write from any other entry in the batch. Grepping
+ * `measurement.batch.hr-bucket` shows, per sync, how many bucket entries the
+ * client sent and what happened to each — confirming (or ruling out)
+ * server-side rejection/dedup as the density-gap cause. Only fires when the
+ * batch actually carried at least one HR-bucket entry so the baseline trace
+ * is unchanged for every other sync.
+ */
+function annotateHeartRateBuckets(
+  entries: BatchIngestEntry[],
+  results: EntryResult[],
+): void {
+  let hrBucketInserted = 0;
+  let hrBucketUpdated = 0;
+  let hrBucketDuplicate = 0;
+  let hrBucketSkipped = 0;
+  let hrBucketFailed = 0;
+  for (let index = 0; index < entries.length; index++) {
+    if (!targetsAggregatedBucket(entries[index].externalId)) continue;
+    switch (results[index]?.status) {
+      case "inserted":
+        hrBucketInserted++;
+        break;
+      case "updated":
+        hrBucketUpdated++;
+        break;
+      case "duplicate":
+        hrBucketDuplicate++;
+        break;
+      case "skipped":
+        hrBucketSkipped++;
+        break;
+      case "failed":
+        hrBucketFailed++;
+        break;
+    }
+  }
+  const hrBucketAttempted =
+    hrBucketInserted +
+    hrBucketUpdated +
+    hrBucketDuplicate +
+    hrBucketSkipped +
+    hrBucketFailed;
+  if (hrBucketAttempted > 0) {
+    annotate({
+      action: { name: "measurement.batch.hr-bucket" },
+      meta: {
+        attempted: hrBucketAttempted,
+        inserted: hrBucketInserted,
+        updated: hrBucketUpdated,
+        duplicate: hrBucketDuplicate,
+        skipped: hrBucketSkipped,
+        failed: hrBucketFailed,
+        processed: entries.length,
+      },
+    });
+  }
+}
+
+/**
+ * Kick off PR detection for this user when at least one row was written, so
+ * a single off-day reading still gets evaluated; the warm-up gate inside the
+ * detector decides whether it's a record. Suppress push notifications for
+ * historical backfills above the silent threshold. Updated `stats:*` rows
+ * also count — a per-day-total overwrite can flip the day's value past a
+ * personal record.
+ *
+ * A batch of duplicates only does not enqueue. It changed nothing a record
+ * or a reminder could read, and an anchor-reset backfill sends hundreds of
+ * them, each of which used to queue a detection pass, a reminder sweep and
+ * an audit row for no new reading.
+ */
+export async function enqueueRecordJobs(args: {
+  userId: string;
+  processed: number;
+  tallies: Pick<BatchTallies, "insertedCount" | "updatedCount">;
+}): Promise<void> {
+  const { userId, processed, tallies } = args;
+  if (!(tallies.insertedCount > 0 || tallies.updatedCount > 0)) return;
+  const silent = processed > PR_DETECTION_SILENT_THRESHOLD;
+  // Eventful Vorsorge satisfaction. A matching reading just landed; resolve
+  // the user's reminders now rather than waiting on the 15-min cron.
+  // Fire-and-forget — the cron reconciles on enqueue miss.
+  fireAndForget(enqueueReminderSatisfy(userId), {
+    action: "reminder.satisfy.enqueue",
+  });
+  try {
+    await enqueuePrDetection(userId, { silent });
+    // Telemetry, not an audit event: it rides the wide event.
+    annotate({
+      meta: { pr_detection_enqueued: true, pr_detection_silent: silent },
+    });
+  } catch (err) {
+    // Enqueue failure is non-fatal — the 30-minute fallback cron
+    // picks the user up in the next slot. Log so the operator
+    // notices repeated failures in Wide-Event traffic.
+    annotate({
+      action: { name: "personal_records.detection_enqueue_failed" },
+      meta: { error: err instanceof Error ? err.message : String(err) },
+    });
+  }
+}
+
+/** The batch's own wide-event summary, the last annotation of the request. */
+export function annotateIngest(args: {
+  processed: number;
+  tallies: BatchTallies;
+  skipped: SkippedEntry[];
+  skippedByReason: Record<string, number>;
+  syncTrigger: HealthKitRequestSyncTrigger | undefined;
+}): void {
+  const { processed, tallies, skipped, skippedByReason, syncTrigger } = args;
+  annotate({
+    action: { name: "measurement.batch.ingest" },
+    meta: {
+      processed,
+      inserted: tallies.insertedCount,
+      updated: tallies.updatedCount + tallies.unchangedStatsCount,
+      stats_unchanged: tallies.unchangedStatsCount,
+      duplicates: tallies.duplicateCount,
+      skipped: skipped.length,
+      // Per-reason skip breakdown (issue #585); see `summariseSkips`.
+      skipped_by_reason: skippedByReason,
+      failed: tallies.failedCount,
+      // Diagnostic sync-trigger tag (iOS #66: foreground / background / push
+      // / manual), or null when the client did not send one. It never
+      // influences any count above; it decides how hard the caches are
+      // invalidated.
+      syncTrigger: syncTrigger ?? null,
+      // Duplicates refused because a live `stats:` row already covers the
+      // sample's hour or day (`folded-window.ts`).
+      folded_window: tallies.foldedWindowCount,
+      // Inserts that took the set-based path rather than one reconciler
+      // transaction each.
+      set_inserted: tallies.setInsertedCount,
+    },
+  });
+}

src/lib/measurements/batch-ingest/sync-checkpoint.ts

@@ -0,0 +1,132 @@
+/**
+ * The sync checkpoint and the HealthKit stamps a batch sets on the account.
+ *
+ * This batch route is the native ingestion boundary, so it is also where the
+ * SyncMode checkpoint (`User.lastSyncedAt`, the same column
+ * `markSyncCheckpoint` writes) is established — `GET /api/sync/state` only
+ * reports it. The checkpoint covers MANUAL pushes too: the client synced
+ * whatever it was holding. A batch carrying a hard failure leaves the
+ * checkpoint where it was, so the undelivered window stays inside whatever
+ * the client asks for next.
+ *
+ * The HealthKit fields ride the same update and keep their narrower
+ * condition. Stamp only after every attempted write has reached a durable,
+ * non-failed verdict; MANUAL-only and skipped-only batches are not
+ * HealthKit syncs.
+ *
+ * The trigger the client declared is stamped alongside, and a background- or
+ * push-triggered batch additionally dates `healthKitLastBackgroundSyncAt`.
+ * Without that second timestamp there is no way to tell a phone that delivers
+ * on its own from one that only ever delivers while the app is open — both
+ * look like a moving `healthKitLastSyncedAt`. A batch that declares no
+ * trigger writes null over the previous value rather than leaving a stale one
+ * standing: the field describes THIS sync, and an older build's silence is
+ * not evidence about it.
+ *
+ * None of that is true of a scoped credential, which is why it is excluded
+ * from the whole block rather than only from the HealthKit half. The
+ * checkpoint's meaning is "the client synced what it was holding", and a
+ * third-party bridge is not the client and holds nothing on its behalf.
+ * `lastSyncedAt` is what `GET /api/sync/state` reports and what the phone
+ * uses to size its undelivered window, so letting a scale advance it can make
+ * the phone skip a window of its own samples — a silent data loss whose cause
+ * would be nowhere near the code that caused it.
+ *
+ * A `backfill` batch (ios #77) is a catch-up, not a sync. It records its
+ * types under `lastBackfillAt` only and moves none of the live stamps:
+ * thirty days of history must not read as thirty healthy daily syncs.
+ */
+import { prisma } from "@/lib/db";
+import type { MeasurementType } from "@/generated/prisma/client";
+import {
+  BACKFILL_SYNC_TRIGGER,
+  recordHealthKitTypeArrivals,
+  tallyHealthKitArrivals,
+  type HealthKitRequestSyncTrigger,
+} from "@/lib/integrations/healthkit-type-sync";
+import type { HourBucketWrite } from "@/lib/measurements/intraday-hour-buckets";
+
+import type { EntryResult, PreparedEntry } from "./entries";
+
+export async function stampSyncCheckpoint(args: {
+  userId: string;
+  scoped: boolean;
+  syncTrigger: HealthKitRequestSyncTrigger | undefined;
+  prepared: PreparedEntry[];
+  results: EntryResult[];
+  failedCount: number;
+  unchangedStatsRows: Set<PreparedEntry>;
+  hourLanded: HourBucketWrite[];
+  hourChanged: Set<HourBucketWrite>;
+}): Promise<void> {
+  const {
+    userId,
+    scoped,
+    syncTrigger,
+    prepared,
+    results,
+    failedCount,
+    unchangedStatsRows,
+    hourLanded,
+    hourChanged,
+  } = args;
+
+  const healthKitSyncSucceeded =
+    failedCount === 0 &&
+    (hourLanded.length > 0 ||
+      prepared.some((p) => {
+        if ((p.entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") return false;
+        const status = results[p.index]?.status;
+        return (
+          status === "inserted" ||
+          status === "updated" ||
+          status === "duplicate"
+        );
+      }));
+
+  const backfill = syncTrigger === BACKFILL_SYNC_TRIGGER;
+  if (failedCount === 0 && !scoped) {
+    const syncedAt = new Date();
+    const deliveredWithoutTheApp =
+      syncTrigger === "background" || syncTrigger === "push";
+    // Per-type arrivals for the Apple Health card (#1173), Apple Health rows
+    // and hours only, under the checkpoint's conditions. Since #1259 only a
+    // new or changed one brings something (`brought`).
+    const arrivals = tallyHealthKitArrivals([
+      ...prepared.flatMap((p) => {
+        const status = results[p.index]?.status;
+        if (p.row.source !== "APPLE_HEALTH" || !status) return [];
+        if (status === "skipped" || status === "failed") return [];
+        const unchanged = unchangedStatsRows.has(p);
+        const brought = status !== "duplicate" && !unchanged;
+        return [{ type: p.row.type as MeasurementType, brought }];
+      }),
+      ...hourLanded.map((w) => ({ type: w.type, brought: hourChanged.has(w) })),
+    ]);
+    if (arrivals.size > 0) {
+      await recordHealthKitTypeArrivals(prisma, {
+        userId,
+        counts: arrivals,
+        trigger: syncTrigger ?? null,
+        at: syncedAt,
+      });
+    }
+    if (!backfill) {
+      await prisma.user.update({
+        where: { id: userId },
+        data: {
+          lastSyncedAt: syncedAt,
+          ...(healthKitSyncSucceeded
+            ? {
+                healthKitLastSyncedAt: syncedAt,
+                healthKitLastSyncTrigger: syncTrigger ?? null,
+                ...(deliveredWithoutTheApp
+                  ? { healthKitLastBackgroundSyncAt: syncedAt }
+                  : {}),
+              }
+            : {}),
+        },
+      });
+    }
+  }
+}

src/lib/measurements/batch-ingest/tail.ts

@@ -0,0 +1,152 @@
+/**
+ * The consumers a written batch has to tell: the measurement caches, the
+ * rollup and status tail, the morning refresh and the data-arrival spine.
+ */
+import type { MeasurementType } from "@/generated/prisma/client";
+import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
+import { groupRowsByArrivalKind } from "@/lib/arrivals/measurement-kind";
+import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
+import type { HealthKitRequestSyncTrigger } from "@/lib/integrations/healthkit-type-sync";
+
+import type { EntryResult, PreparedEntry } from "./entries";
+
+type Identity = { type: MeasurementType; measuredAt: Date };
+
+export async function afterBatchWrite(args: {
+  userId: string;
+  syncTrigger: HealthKitRequestSyncTrigger | undefined;
+  prepared: PreparedEntry[];
+  results: EntryResult[];
+  insertedCount: number;
+  updatedCount: number;
+  insertedPrepared: PreparedEntry[];
+  writtenIdentities: Identity[];
+  hourInsertedCount: number;
+  hourUpdatedCount: number;
+}): Promise<void> {
+  const { userId, insertedCount, updatedCount } = args;
+  const freshlyInsertedRows = args.insertedPrepared.map((p) => ({
+    type: p.row.type as MeasurementType,
+    measuredAt:
+      p.row.measuredAt instanceof Date
+        ? p.row.measuredAt
+        : new Date(p.row.measuredAt as string),
+  }));
+
+  // Bust per-user analytics + achievements + workouts caches when at least
+  // one row landed so the next read picks up the ingested batch. Skipped-only
+  // ingests don't change state, so the invalidation is gated on
+  // `insertedCount > 0 || updatedCount > 0` — overwriting a per-day `stats:*`
+  // row changes the day's reading and must invalidate every consumer that
+  // reads through this user's measurements.
+  if (insertedCount > 0 || updatedCount > 0) {
+    await afterMeasurementsWritten({ ...args, freshlyInsertedRows });
+  }
+
+  // The hours are not measurements, so they touched none of the measurement
+  // consumers above; only the caches that read the intraday profile (the
+  // same-time baseline) are marked stale.
+  if (
+    args.hourInsertedCount + args.hourUpdatedCount > 0 &&
+    insertedCount === 0 &&
+    updatedCount === 0
+  ) {
+    invalidateUserMeasurements(userId);
+  }
+}
+
+async function afterMeasurementsWritten(args: {
+  userId: string;
+  syncTrigger: HealthKitRequestSyncTrigger | undefined;
+  prepared: PreparedEntry[];
+  results: EntryResult[];
+  writtenIdentities: Identity[];
+  freshlyInsertedRows: Identity[];
+}): Promise<void> {
+  const {
+    userId,
+    syncTrigger,
+    prepared,
+    results,
+    writtenIdentities,
+    freshlyInsertedRows,
+  } = args;
+  // Evict only when someone is looking (#38 evicted on every batch so a
+  // focus-refetch after a sync showed post-sync data, but a phone syncing in
+  // the background posts a batch every few minutes all day, and each
+  // eviction sent the next dashboard read into a cold rebuild). Every other
+  // background sync marks stale, which `cachedSwr` serves once while one
+  // rebuild runs; this route does the same unless the app is open
+  // (`foreground`), the person pressed sync (`manual`), or the batch wrote
+  // one of their own hand-entered readings (`MANUAL`), whose entry has to
+  // show on the very next read.
+  const wroteManualRow = prepared.some(
+    (p) =>
+      p.row.source === "MANUAL" &&
+      (results[p.index]?.status === "inserted" ||
+        results[p.index]?.status === "updated"),
+  );
+  const evict =
+    syncTrigger === "foreground" || syncTrigger === "manual" || wroteManualRow;
+  // Every row this batch created, changed or retired is in
+  // `writtenIdentities`, so its earliest `measuredAt` bounds what changed:
+  // closed blocks of older readings (the pre-fold remainder, the day
+  // view's past history blocks) stay cached across the sync.
+  const touchedFrom = new Date(
+    Math.min(...writtenIdentities.map((w) => w.measuredAt.getTime())),
+  );
+  invalidateUserMeasurements(userId, {
+    ...(evict ? { evict: true } : {}),
+    ...(Number.isFinite(touchedFrom.getTime()) ? { touchedFrom } : {}),
+  });
+
+  // The shared post-mutation tail for every distinct (type, day) the batch
+  // touched. Updates participate as well as inserts, so this deliberately
+  // retains the prepared-row scope. A batch is the phone's sync, held to the
+  // sync gap, unless it wrote one of the person's own hand-entered readings,
+  // which is narrated like any other hand entry.
+  await afterMeasurementMutation(userId, writtenIdentities, {
+    source: wroteManualRow ? "manual" : "sync",
+  });
+
+  // Apple-Health sleep is the third sleep transport. If any SLEEP_DURATION
+  // row for last night just landed, kick the debounced morning refresh so the
+  // digest/score finalise with the current sleep. The trigger judges "last
+  // night" in the user's profile tz, so a historical Apple export replaying
+  // months of nights never re-triggers for an old night. Fire-and-forget.
+  void maybeEnqueueMorningRefresh(
+    userId,
+    freshlyInsertedRows
+      .filter((row) => row.type === "SLEEP_DURATION")
+      .map((row) => row.measuredAt),
+  ).catch(() => {});
+
+  // The measurement arm of the data-arrival spine.
+  //
+  // ONE emit per kind per request, never one per row. The batch route is the
+  // Apple-Health / iOS ingest path, so it is also the single biggest backfill
+  // path in the product: a ten-year export arrives here in 500-row batches.
+  // Emitting per row would put thousands of classifier calls (and thousands
+  // of dropped-event annotations) on the hot path for no gain, since the
+  // day-scoped singleton key collapses them to one job anyway.
+  //
+  // Blood pressure is deliberately one arrival, not two: it is stored as two
+  // rows (SYS + DIA) sharing a `measuredAt`, and a reader cares about the
+  // reading, not the arms.
+  //
+  // Only rows the write actually INSERTED count — updated rows are the
+  // `stats:` overwrite contract re-stating a value the record already had,
+  // which is not news. If nothing was inserted the classifier returns `noop`
+  // and nothing is enqueued.
+  for (const [kind, group] of groupRowsByArrivalKind(freshlyInsertedRows)) {
+    void emitDataArrival({
+      userId,
+      kind,
+      newestSampleAt: group.newestAt,
+      insertedCount: group.count,
+      source: "batch",
+    }).catch(() => {});
+  }
+}

src/lib/measurements/batch-ingest/write-hours.ts

@@ -0,0 +1,99 @@
+/**
+ * The hourly sums of a batch, written to the intraday profile after the
+ * measurements. One write per slot: the last entry for one start wins (an
+ * earlier one in the same batch is a superseded duplicate), and the two
+ * starts of the hour a fall-back day repeats are summed
+ * (`combineHourWrites`).
+ */
+import { prisma } from "@/lib/db";
+import { annotate } from "@/lib/logging/context";
+import {
+  combineHourWrites,
+  writeHourBuckets,
+  type HourBucketEntry,
+  type HourBucketWrite,
+} from "@/lib/measurements/intraday-hour-buckets";
+
+import type { EntryResult } from "./entries";
+
+export interface HourOutcome {
+  hourInsertedCount: number;
+  hourUpdatedCount: number;
+  hourLanded: HourBucketWrite[];
+  /** #1259: the slots whose value changed, as opposed to an unchanged re-post. */
+  hourChanged: Set<HourBucketWrite>;
+  /** Superseded entries, counted with the batch's duplicates. */
+  duplicates: number;
+  /** Entries whose write failed, counted with the batch's failures. */
+  failed: number;
+}
+
+export async function writeBatchHours(args: {
+  userId: string;
+  hourWrites: HourBucketEntry[];
+  accountTz: string | undefined;
+  results: EntryResult[];
+  processed: number;
+}): Promise<HourOutcome> {
+  const { userId, hourWrites, accountTz, results, processed } = args;
+  const outcome: HourOutcome = {
+    hourInsertedCount: 0,
+    hourUpdatedCount: 0,
+    hourLanded: [],
+    hourChanged: new Set<HourBucketWrite>(),
+    duplicates: 0,
+    failed: 0,
+  };
+  if (!(hourWrites.length > 0 && accountTz)) return outcome;
+
+  const { writes, superseded } = combineHourWrites(hourWrites);
+  for (const index of superseded) {
+    results[index] = {
+      index,
+      status: "duplicate",
+      reason: "superseded_in_batch",
+    };
+    outcome.duplicates++;
+  }
+  const toWrite = writes.map(({ write }) => write);
+  try {
+    const statuses = await writeHourBuckets(prisma, userId, accountTz, toWrite);
+    for (const { write, contributors } of writes) {
+      const status = statuses.get(write.index) ?? "inserted";
+      // An unchanged hour still reads `updated` on the wire (#1259).
+      const wire = status === "inserted" ? "inserted" : "updated";
+      for (const index of contributors) {
+        results[index] = { index, status: wire };
+        if (status === "inserted") outcome.hourInsertedCount++;
+        else outcome.hourUpdatedCount++;
+      }
+      outcome.hourLanded.push(write);
+      if (status !== "unchanged") outcome.hourChanged.add(write);
+    }
+  } catch (err) {
+    for (const { contributors } of writes) {
+      for (const index of contributors) {
+        results[index] = {
+          index,
+          status: "failed",
+          reason: "write_failed",
+        };
+        outcome.failed++;
+      }
+    }
+    annotate({
+      action: { name: "measurement.batch.hour-bucket.failed" },
+      meta: { error: err instanceof Error ? err.message : String(err) },
+    });
+  }
+  annotate({
+    action: { name: "measurement.batch.hour-bucket" },
+    meta: {
+      attempted: hourWrites.length,
+      inserted: outcome.hourInsertedCount,
+      updated: outcome.hourUpdatedCount,
+      processed,
+    },
+  });
+  return outcome;
+}

src/lib/measurements/batch-ingest/write-measurements.ts

@@ -0,0 +1,566 @@
+/**
+ * Second pass of the batch ingest: write the prepared rows and record each
+ * one's verdict.
+ *
+ * Every row ends in one of the reconciler's verdicts, reached one of four
+ * ways: superseded by a later `stats:` entry in the same batch, an unchanged
+ * `stats:` re-post answered without a write, a duplicate found without a
+ * transaction (the cross-source merge, the exact stored duplicate, the
+ * folded window), or a write: the set-based insert for rows that share no
+ * identity with another entry, the per-entry reconciler for the rest.
+ */
+import { prisma } from "@/lib/db";
+import {
+  MEASURED_AT_TOLERANCE_MS,
+  isMergeableSource,
+  isSameReadingAcrossSource,
+  oppositeMergeSource,
+  type MergeCandidate,
+} from "@/lib/measurements/cross-source-merge";
+import {
+  externalKeyOf,
+  onceInBatch,
+  readStoredSampleIdentities,
+  readStoredStatsRows,
+  sampleIsStoredDuplicate,
+  statsWriteUnchanged,
+} from "@/lib/measurements/batch-stored-rows";
+import {
+  lockMeasurementUser,
+  reconcileExternalMeasurement,
+  type MeasurementReconciliationVerdict,
+} from "@/lib/measurements/reconcile-external-measurement";
+import { classifyRangeRejection } from "@/lib/measurements/plausibility-gate";
+import {
+  FOLDED_WINDOW_REASON,
+  findFoldedWindowDuplicates,
+} from "@/lib/measurements/folded-window";
+import { annotate } from "@/lib/logging/context";
+import type {
+  MeasurementSource,
+  MeasurementType,
+  Prisma,
+} from "@/generated/prisma/client";
+
+import {
+  isStatsExternalId,
+  outOfRangeResult,
+  type EntryResult,
+  type PreparedEntry,
+} from "./entries";
+
+/**
+ * Rows per set-based insert statement. 500 entries fit two statements; each
+ * row binds about fifteen parameters, far under PostgreSQL's 65 535.
+ */
+const SET_INSERT_CHUNK = 250;
+
+type Verdict = Pick<MeasurementReconciliationVerdict, "status"> &
+  Partial<{
+    dirtyIdentities: Array<{ type: MeasurementType; measuredAt: Date }>;
+    error: { code?: string };
+  }>;
+
+/** What the reads before the write loop found for this batch. */
+interface StoredState {
+  crossSourceCandidates: MergeCandidate[];
+  onceInThisBatch: ReturnType<typeof onceInBatch>;
+  existingByExternalKey: Awaited<ReturnType<typeof readStoredSampleIdentities>>;
+  storedStats: Awaited<ReturnType<typeof readStoredStatsRows>>;
+  foldedPrepared: Set<PreparedEntry>;
+}
+
+/** The reconciler's input for one prepared row. */
+function desiredFor(userId: string, p: PreparedEntry) {
+  const statsRow = isStatsExternalId(p.row.externalId as string);
+  return {
+    userId,
+    type: p.row.type as MeasurementType,
+    value: p.row.value as number,
+    valueMin: p.row.valueMin as number | null,
+    valueMax: p.row.valueMax as number | null,
+    unit: p.row.unit as string,
+    source: p.row.source as Prisma.MeasurementUncheckedCreateInput["source"],
+    measuredAt: p.row.measuredAt as Date,
+    externalId: p.row.externalId as string,
+    externalSourceVersion: p.row.externalSourceVersion as string | null,
+    glucoseContext: p.row
+      .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"],
+    sleepStage: p.row
+      .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"],
+    rhythmClassification: p.row
+      .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"],
+    deviceType: p.row.deviceType as string | null,
+    aggregationProvenance: statsRow ? ("HEALTHKIT_STATISTICS" as const) : null,
+  };
+}
+
+/**
+ * The reads the verdicts need, in the order the route has always made them:
+ * the cross-source candidates, the stored sample identities, the stored
+ * `stats:` rows, the folded windows.
+ */
+async function readStoredState(
+  userId: string,
+  prepared: PreparedEntry[],
+): Promise<StoredState> {
+  // Same-reading merge for the MANUAL ↔ APPLE_HEALTH mirror duplicate
+  // (iOS #2). When the iOS app logs a manual reading offline it
+  // mirrors that reading into HealthKit; on pairing, adopt-on-pair
+  // uploads it as MANUAL and HealthKit background sync independently
+  // re-ingests the mirrored sample as APPLE_HEALTH. The two carry
+  // different externalIds + sources, so the composite unique index lets
+  // both land — duplicating the whole hand-logged history on first pair.
+  // We collapse them at ingest: an incoming MANUAL / APPLE_HEALTH row is
+  // dropped as a `duplicate` when a same-reading row of the OPPOSITE
+  // source already exists (same type + value + measuredAt within a tight
+  // ±2 s window). `stats:*` cumulative rows are excluded — those are
+  // per-day aggregates the observer overwrites in place, never
+  // hand-entered point readings. See `cross-source-merge.ts` for the
+  // full rule + why first-physical-reading-wins (the value is identical
+  // between the two rows, so only the source label differs).
+  //
+  // Pull the opposite-source candidate rows in one widened query: for
+  // every mergeable, non-`stats:*` incoming row, look for an opposite-
+  // source row of the same type whose measuredAt falls inside the
+  // tolerance window. Compared in JS because the value match needs a
+  // float epsilon the SQL `IN` can't express.
+  const mergeProbes = prepared.filter(
+    (p) =>
+      isMergeableSource(p.row.source as string) &&
+      !isStatsExternalId(p.row.externalId as string),
+  );
+  const crossSourceCandidates: MergeCandidate[] = [];
+  if (mergeProbes.length > 0) {
+    const candidateRows = await prisma.measurement.findMany({
+      where: {
+        userId,
+        deletedAt: null,
+        OR: mergeProbes.map((p) => {
+          const measuredAt = p.row.measuredAt as Date;
+          return {
+            type: p.row.type as MeasurementType,
+            source: oppositeMergeSource(
+              p.row.source as "MANUAL" | "APPLE_HEALTH",
+            ),
+            measuredAt: {
+              gte: new Date(measuredAt.getTime() - MEASURED_AT_TOLERANCE_MS),
+              lte: new Date(measuredAt.getTime() + MEASURED_AT_TOLERANCE_MS),
+            },
+          };
+        }),
+      },
+      select: { type: true, source: true, value: true, measuredAt: true },
+    });
+    crossSourceCandidates.push(...candidateRows);
+  }
+
+  // Exact-duplicate prefilter. A re-sent sample used to cost one
+  // interactive transaction of about ten round-trips (savepoint, three
+  // advisory locks, two candidate reads, a row lock) only for the
+  // reconciler to answer `duplicate` without writing. A backfill after an
+  // anchor reset re-sends whole 500-entry batches of exactly that.
+  //
+  // The reconciler answers `duplicate` for a non-`stats:` row exactly when
+  // an existing row, live OR tombstoned, has the same
+  // `(type, source, externalId)` AND the same `measuredAt` and
+  // `sleepStage`: the natural-identity index is unique NULLS NOT DISTINCT,
+  // so that row is both its external and its natural match, and with no
+  // aggregation provenance on a sample nothing else can win. One read over
+  // the batch's sample ids, without a `deletedAt` filter, answers that for
+  // every row at once.
+  //
+  // The verdict is taken at the very point the reconciler would have been
+  // called, after `superseded_in_batch` and the cross-source merge, so
+  // statuses, reasons, counters and the in-batch merge candidates are the
+  // same as before. A row is only prefiltered when neither its external
+  // key nor its natural identity occurs a second time in the batch: an
+  // earlier entry sharing either could move or retire the matched row
+  // before this one is reached, and then the reconciler has to decide.
+  // `stats:` rows are never prefiltered (overwrite semantics), and neither
+  // is a value the reconciler's own plausibility gate would refuse.
+  const onceInThisBatch = onceInBatch(prepared.map((p) => p.row));
+  const existingByExternalKey = await readStoredSampleIdentities(
+    prisma,
+    userId,
+    [
+      ...new Set(
+        prepared
+          .map((p) => p.row.externalId as string)
+          .filter((externalId) => !isStatsExternalId(externalId)),
+      ),
+    ],
+  );
+
+  // An unchanged `stats:` re-post is reported `updated` without a write
+  // (see `batch-stored-rows.ts`), under the same once-per-batch key rule.
+  const storedStats = await readStoredStatsRows(prisma, userId, [
+    ...new Set(
+      prepared
+        .map((p) => p.row.externalId as string)
+        .filter((externalId) => isStatsExternalId(externalId)),
+    ),
+  ]);
+
+  // The `folded_window` guard (`folded-window.ts`). A raw Apple Health
+  // sample of a dense or mean type, old enough to have been folded, whose
+  // hour or day a live `stats:` row already covers, is a duplicate: the fold
+  // no longer leaves the raw rows behind as tombstones, so this is what
+  // keeps a re-upload after an anchor reset from standing beside the mean
+  // that already accounts for it. One read for the batch, and none at all
+  // when no row is old enough.
+  const foldedIndexes = await findFoldedWindowDuplicates(
+    prisma,
+    userId,
+    prepared.map((p) => ({
+      type: p.row.type as MeasurementType,
+      source: p.row.source as MeasurementSource,
+      externalId: p.row.externalId as string,
+      measuredAt: p.row.measuredAt as Date,
+    })),
+  );
+  const foldedPrepared = new Set(
+    [...foldedIndexes].map((index) => prepared[index]),
+  );
+
+  return {
+    crossSourceCandidates,
+    onceInThisBatch,
+    existingByExternalKey,
+    storedStats,
+    foldedPrepared,
+  };
+}
+
+/** The measurement half of one batch: its verdicts and what they started. */
+export class MeasurementBatchWriter {
+  insertedCount = 0;
+  updatedCount = 0;
+  /**
+   * `stats:` re-posts reported `updated` with nothing to write (see
+   * `isUnchangedStatsRow`). Counted in the response's `updated` like before,
+   * and in nothing that starts work.
+   */
+  unchangedStatsCount = 0;
+  readonly unchangedStatsRows = new Set<PreparedEntry>(); // #1259: not an arrival
+  duplicateCount = 0;
+  failedCount = 0;
+  /** Subset of `insertedCount` written by the set-based insert. */
+  setInsertedCount = 0;
+  /**
+   * Subset of `duplicateCount` collapsed by the MANUAL↔APPLE_HEALTH
+   * same-reading merge (iOS #2), as opposed to a plain composite-key
+   * duplicate. Surfaced as a dedicated wide-event count so an operator can
+   * see how often the standalone-pair mirror collapses.
+   */
+  crossSourceMergedCount = 0;
+  /** Subset of `duplicateCount` refused by the `folded_window` guard. */
+  foldedWindowCount = 0;
+  /** Only rows the reconciler actually inserted drive arrival side effects. */
+  readonly insertedPrepared: PreparedEntry[] = [];
+  readonly writtenIdentities: Array<{
+    type: MeasurementType;
+    measuredAt: Date;
+  }> = [];
+
+  /**
+   * Rows already chosen for insert earlier in THIS batch — so a MANUAL
+   * and an APPLE_HEALTH same-reading arriving in the SAME batch also
+   * collapse (the DB probe only sees rows from prior batches).
+   */
+  private readonly inBatchCandidates: MergeCandidate[] = [];
+  private readonly pendingInserts: PreparedEntry[] = [];
+  private stored!: StoredState;
+
+  constructor(
+    private readonly userId: string,
+    private readonly results: EntryResult[],
+  ) {}
+
+  /** Write every prepared row, in order, recording each verdict. */
+  async write(prepared: PreparedEntry[]): Promise<void> {
+    if (prepared.length === 0) return;
+    this.stored = await readStoredState(this.userId, prepared);
+
+    // `stats:` rows carry overwrite semantics and the last entry in one batch
+    // is authoritative. Earlier snapshots are safe duplicates.
+    const lastStatsIndexByKey = new Map<string, number>();
+    for (const p of prepared) {
+      if (!isStatsExternalId(p.row.externalId as string)) continue;
+      lastStatsIndexByKey.set(
+        `${p.row.type}::${p.row.source}::${p.row.externalId}`,
+        p.index,
+      );
+    }
+
+    for (const p of prepared) {
+      await this.writeRow(p, lastStatsIndexByKey);
+    }
+    await this.flushPendingInserts();
+  }
+
+  private async writeRow(
+    p: PreparedEntry,
+    lastStatsIndexByKey: Map<string, number>,
+  ): Promise<void> {
+    const externalId = p.row.externalId as string;
+    const statsRow = isStatsExternalId(externalId);
+    const key = `${p.row.type}::${p.row.source}::${externalId}`;
+    if (statsRow && lastStatsIndexByKey.get(key) !== p.index) {
+      this.results[p.index] = {
+        index: p.index,
+        status: "duplicate",
+        reason: "superseded_in_batch",
+      };
+      this.duplicateCount++;
+      return;
+    }
+
+    if (statsRow && this.isUnchangedStatsRow(p)) {
+      this.results[p.index] = { index: p.index, status: "updated" };
+      this.unchangedStatsCount++;
+      this.unchangedStatsRows.add(p);
+      this.pushMergeCandidate(p);
+      return;
+    }
+
+    if (!statsRow && this.hasSameReadingSibling(p)) {
+      this.results[p.index] = {
+        index: p.index,
+        status: "duplicate",
+        reason: "cross_source_merge",
+      };
+      this.duplicateCount++;
+      this.crossSourceMergedCount++;
+      return;
+    }
+
+    // Both verdicts below are the reconciler's `duplicate`, reached without
+    // its transaction; like every duplicate they fall through to the
+    // in-batch merge candidates.
+    const prefiltered = !statsRow && this.isExactStoredDuplicate(p);
+    const folded =
+      !prefiltered && !statsRow && this.stored.foldedPrepared.has(p);
+    if (prefiltered || folded) {
+      this.results[p.index] = folded
+        ? {
+            index: p.index,
+            status: "duplicate",
+            reason: FOLDED_WINDOW_REASON,
+          }
+        : { index: p.index, status: "duplicate" };
+      this.duplicateCount++;
+      if (folded) this.foldedWindowCount++;
+      this.pushMergeCandidate(p);
+      return;
+    }
+
+    if (this.stored.onceInThisBatch(p.row)) {
+      this.pendingInserts.push(p);
+      this.pushMergeCandidate(p);
+      return;
+    }
+
+    await this.flushPendingInserts();
+    this.applyVerdict(p, await this.reconcileRow(p), true);
+  }
+
+  private hasSameReadingSibling(p: PreparedEntry): boolean {
+    const incoming = {
+      type: p.row.type as MeasurementType,
+      source: p.row.source as string,
+      value: p.row.value as number,
+      measuredAt: p.row.measuredAt as Date,
+    };
+    if (!isMergeableSource(incoming.source)) return false;
+    for (const candidate of this.stored.crossSourceCandidates) {
+      if (isSameReadingAcrossSource(incoming, candidate)) return true;
+    }
+    for (const candidate of this.inBatchCandidates) {
+      if (isSameReadingAcrossSource(incoming, candidate)) return true;
+    }
+    return false;
+  }
+
+  private isExactStoredDuplicate(p: PreparedEntry): boolean {
+    return (
+      sampleIsStoredDuplicate(
+        this.stored.existingByExternalKey.get(
+          externalKeyOf(p.row.type, p.row.source, p.row.externalId),
+        ),
+        p.row,
+      ) &&
+      this.stored.onceInThisBatch(p.row) &&
+      classifyRangeRejection(
+        p.row.type as MeasurementType,
+        p.row.value as number,
+      ) === null
+    );
+  }
+
+  private isUnchangedStatsRow(p: PreparedEntry): boolean {
+    return (
+      this.stored.onceInThisBatch(p.row) &&
+      classifyRangeRejection(
+        p.row.type as MeasurementType,
+        p.row.value as number,
+      ) === null &&
+      statsWriteUnchanged(
+        this.stored.storedStats.get(
+          externalKeyOf(p.row.type, p.row.source, p.row.externalId),
+        ),
+        desiredFor(this.userId, p),
+      )
+    );
+  }
+
+  private pushMergeCandidate(p: PreparedEntry): void {
+    if (isMergeableSource(p.row.source as string)) {
+      this.inBatchCandidates.push({
+        type: p.row.type as MeasurementType,
+        source: p.row.source as string,
+        value: p.row.value as number,
+        measuredAt: p.row.measuredAt as Date,
+      });
+    }
+  }
+
+  /**
+   * Record one row's verdict. `candidate` is false for a row whose merge
+   * candidate was already pushed when it was buffered (see below).
+   */
+  private applyVerdict(
+    p: PreparedEntry,
+    verdict: Verdict,
+    candidate: boolean,
+  ): void {
+    switch (verdict.status) {
+      case "inserted":
+        this.results[p.index] = { index: p.index, status: "inserted" };
+        this.insertedCount++;
+        this.insertedPrepared.push(p);
+        this.writtenIdentities.push({
+          type: p.row.type as MeasurementType,
+          measuredAt: p.row.measuredAt as Date,
+        });
+        break;
+      case "updated":
+      case "resurrected":
+        this.results[p.index] = { index: p.index, status: "updated" };
+        this.updatedCount++;
+        this.writtenIdentities.push(
+          {
+            type: p.row.type as MeasurementType,
+            measuredAt: p.row.measuredAt as Date,
+          },
+          ...(verdict.dirtyIdentities ?? []),
+        );
+        break;
+      case "duplicate":
+        this.results[p.index] = { index: p.index, status: "duplicate" };
+        this.duplicateCount++;
+        break;
+      case "failed":
+        this.results[p.index] = {
+          index: p.index,
+          status: "failed",
+          reason: verdict.error?.code ?? "write_failed",
+        };
+        this.failedCount++;
+        return;
+      case "rejected_range":
+        // The reconciler carries the same plausibility gate the first pass
+        // already applied, so reaching here means the two disagreed —
+        // report it exactly as the earlier guard would have, and let the
+        // wide-event tally record which end refused.
+        this.results[p.index] = outOfRangeResult(
+          p.index,
+          p.row.type as string,
+          p.row.value as number,
+        );
+        return;
+    }
+    if (candidate) this.pushMergeCandidate(p);
+  }
+
+  private reconcileRow(p: PreparedEntry) {
+    return prisma.$transaction((tx) =>
+      reconcileExternalMeasurement(tx, desiredFor(this.userId, p), {
+        exactExternalMatch: isStatsExternalId(p.row.externalId as string)
+          ? "update"
+          : "duplicate",
+      }),
+    );
+  }
+
+  // Set-based inserts. A fresh sample used to cost one interactive
+  // transaction of about ten round trips (savepoint, three advisory locks,
+  // two candidate reads, the insert), so a 500-entry batch of new samples
+  // was 500 of those in a row: the 2–15 s outliers.
+  //
+  // The reconciler inserts exactly when no row, live or tombstoned, holds
+  // the entry's external identity `(type, source, externalId)` or its
+  // natural identity `(type, measuredAt, source, sleepStage)`. Both are
+  // full unique indexes (the natural one NULLS NOT DISTINCT), so
+  // `INSERT … ON CONFLICT DO NOTHING` inserts exactly those rows and skips
+  // every other; a skipped row then goes through the reconciler as before
+  // and gets its own verdict (update, resurrect, duplicate). The insert
+  // runs under the same per-account advisory lock the reconciler takes, so
+  // the two serialise like two reconcilers would.
+  //
+  // Only a row whose external and natural key each occur once in the batch
+  // is buffered, and the buffer is flushed before any other row reaches the
+  // reconciler. Such a row shares no identity with another entry, so no
+  // other entry's reconcile can create, move or retire a row at its
+  // identity, and inserting it out of turn changes nothing. Rows the insert
+  // skips are reconciled in their original order. A buffered row's merge
+  // candidate is pushed when it is buffered, as the sequential loop pushed
+  // it after a successful write; the only case that differs is a write that
+  // fails outright, where the candidate stays.
+  private async flushPendingInserts(): Promise<void> {
+    if (this.pendingInserts.length === 0) return;
+    const batch = this.pendingInserts.splice(0, this.pendingInserts.length);
+    const landed = new Set<PreparedEntry>();
+    try {
+      for (let i = 0; i < batch.length; i += SET_INSERT_CHUNK) {
+        const slice = batch.slice(i, i + SET_INSERT_CHUNK);
+        const rows = await prisma.$transaction(async (tx) => {
+          await lockMeasurementUser(tx, this.userId);
+          return tx.measurement.createManyAndReturn({
+            data: slice.map((p) => ({
+              ...desiredFor(this.userId, p),
+              sleepStage: (p.row.sleepStage ??
+                null) as Prisma.MeasurementUncheckedCreateInput["sleepStage"],
+              deletedAt: null,
+            })),
+            skipDuplicates: true,
+            select: { type: true, source: true, externalId: true },
+          });
+        });
+        const keys = new Set(
+          rows.map((r) => `${r.type}::${r.source}::${r.externalId}`),
+        );
+        for (const p of slice) {
+          if (keys.has(`${p.row.type}::${p.row.source}::${p.row.externalId}`))
+            landed.add(p);
+        }
+      }
+    } catch (err) {
+      // Whatever did not land goes through the reconciler below, which
+      // turns a real database fault into a per-entry `failed` verdict.
+      annotate({
+        action: { name: "measurement.batch.set_insert_fallback" },
+        meta: {
+          error_class: err instanceof Error ? err.name : "unknown",
+          landed: landed.size,
+          pending: batch.length,
+        },
+      });
+    }
+    this.setInsertedCount += landed.size;
+    for (const p of batch) {
+      if (landed.has(p)) this.applyVerdict(p, { status: "inserted" }, false);
+      else this.applyVerdict(p, await this.reconcileRow(p), false);
+    }
+  }
+}