fix: harden imports and daily medication state (#577)

CHANGELOG.md

@@ -2,6 +2,24 @@
 
 ## [Unreleased]
 
+## [1.31.7] — 2026-07-20
+
+- **Current OpenAI reasoning models use their supported request contract.**
+  Official GPT-5, o1, o3, and o4 models receive `max_completion_tokens`
+  without unsupported `temperature` or `seed` fields; compatible custom
+  gateways retain the legacy parameters they advertise.
+- **Apple Health XML cumulative totals no longer add overlapping sources.**
+  Imports aggregate each source/day independently and select the highest
+  source subtotal, while explicit provenance prevents XML estimates or late
+  legacy samples from overwriting authoritative native HealthKit statistics.
+  Existing legacy aggregates can be repaired by re-uploading their archive.
+- **Every currently actionable medication reaches Today.** Equal-time doses no
+  longer collapse to one row, overdue doses win deterministic ties, and weekly
+  or custom intake windows use the scheduling engine's real availability
+  boundary. The shared digest fixes both the web start page and iOS Home.
+
+Migration `0263_apple_health_aggregate_authority`. No breaking changes.
+
 ## [1.31.6] — 2026-07-21
 
 - **The hardened runtime image can run database migrations again.** The

docs/api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: 3.1.0
 info:
   title: HealthLog API
-  version: 1.31.6
+  version: 1.31.7
   description: >-
     Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
 
@@ -25024,6 +25024,28 @@ components:
               anyOf:
                 - type: string
                 - type: "null"
+            dueCandidates:
+              type: array
+              items:
+                type: object
+                properties:
+                  medicationId:
+                    type: string
+                  medicationName:
+                    type: string
+                  dueAt:
+                    type: string
+                  overdue:
+                    type: boolean
+                  availableFrom:
+                    type: string
+                required:
+                  - medicationId
+                  - medicationName
+                  - dueAt
+                  - overdue
+                  - availableFrom
+                additionalProperties: false
           required:
             - activeCount
             - scheduledToday
@@ -25034,12 +25056,11 @@ components:
             - nextDueMedicationName
             - nextDueMedicationId
           additionalProperties: false
-          description: "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), and
-            the earliest next-due slot across active medications. `nextDueOverdue: true` marks an OPEN overdue slot
-            (anchor passed, still inside its catch-up band). `nextDueMedicationId` deep-links a consumer straight to the
-            overdue medication's card. Staleness contract: the snapshot is cache-served, so a `nextDueAt` in the past
-            with `nextDueOverdue: false` means the anchor passed after the snapshot was built — render the plain day
-            summary, never an overdue state."
+          description: "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), every
+            active medication's current display-due candidate, and legacy scalar fields projected from candidate zero.
+            `overdue: true` marks an OPEN overdue slot (anchor passed, still inside its catch-up band); `availableFrom`
+            is the canonical cadence- and dose-window-derived attribution-band start. `dueCandidates` is optional so
+            older cached snapshots remain valid. Medication ids let consumers deep-link straight to the relevant card."
         healthScore:
           anyOf:
             - type: object

docs/integrations/apple-health.md

@@ -112,29 +112,50 @@ mapped set (workout routes, ECG voltage traces) is counted in
 ### Cumulative HK type → daily aggregate
 
 Cumulative quantity types — steps, active energy, walking/running
-distance, flights climbed — get **collapsed into one row per
-user-local day** to match the iOS daily-aggregation convention.
+distance, flights climbed, time in daylight, and fall count — get
+**collapsed into one estimated row per user-local day**. `export.xml`
+contains source-specific records, not HealthKit's canonical daily
+statistics, so the importer does not sum different devices or apps
+into a source-blind total.
 
+- Records are grouped by type, local day, and a SHA-256 hash of the
+  available source name, source version, and device attributes.
+  Values sum within each identified source-day; the largest
+  source-day subtotal is selected deterministically. If every source
+  attribute is absent, records remain separate rather than being
+  trusted as one source. Raw source and device labels are not
+  persisted.
+- The row records `EXPORT_XML_SOURCE_MAX` provenance, its distinct
+  contributor count, and the selected non-identifying source hash.
 - External ID format: `stats:<HKType>:<YYYY-MM-DD>` (e.g.
   `stats:HKQuantityTypeIdentifierStepCount:2026-05-15`).
 - The day boundary respects the user's timezone preference. The
   worker defaults to `Europe/Berlin` when no preference is set
-  (`src/lib/jobs/apple-health-import-worker.ts:132-133`).
-- A second import of the same archive re-folds the same days onto
-  the same external IDs — the values update in place rather than
-  duplicating.
+  (`src/lib/jobs/apple-health-import-worker.ts`).
+- XML intervals are not reconstructed or split. A record crossing
+  midnight is assigned using its mapped timestamp and remains covered
+  by the explicit estimate contract.
+- A second import re-folds the same days onto the same external IDs.
+  Equal-authority estimates update idempotently rather than duplicate.
+
+The terminal job result exposes
+`cumulativeEstimates: { days, rows }`; the web import card warns when
+one or more cumulative days used this estimate path.
 
 ### Re-import idempotency
 
 The two-axis idempotency story:
 
 1. **File-level** — re-uploading the same `export.zip` (same SHA-256
-   of bytes) short-circuits to the previous job before the parser
-   even starts.
+   of bytes) short-circuits to the previous job only when that job
+   used the current parser revision. Existing jobs are revision 1;
+   new jobs use revision 2, so a pre-fix successful archive can be
+   deliberately processed again under the corrected aggregation.
 2. **Record-level** — when an older `export.zip` is re-exported from
    iOS (same history, slightly newer device timestamps), every
-   record's external ID stays stable. The upsert path matches on
-   `(userId, externalId)` and updates instead of inserting.
+   record's external ID stays stable. The serialized reconciliation
+   path matches the canonical identity and updates instead of
+   inserting.
 
 Practically: a user who exports monthly and re-imports each archive
 sees fresh data merged in without duplicates. The job result line
@@ -150,14 +171,34 @@ is identical to the user-facing flow.
 
 The `ImportJob` row carries `triggeredByAdminId = admin.id` so the
 status endpoint admits both the target user and the triggering
-admin. Idempotency is scoped by target user — an admin re-uploading
-the same archive for the same user resolves to the previous job.
+admin. Idempotency is scoped by target user and parser revision — an
+admin re-uploading the same archive for the same user resolves to the
+previous current-revision job.
 
 Typical use: migrating a friend's history into their HealthLog
 account when they cannot run the import themselves, or backfilling
 a household member's data after the admin received the `export.zip`
 out of band.
 
+## Aggregate authority and recovery
+
+Native HealthKit daily statistics submitted by the iOS batch endpoint
+are canonical and carry `HEALTHKIT_STATISTICS` provenance. XML
+source-maximum estimates carry `EXPORT_XML_SOURCE_MAX`. Both use the
+same serialized reconciler with this ordering:
+
+`LEGACY_UNKNOWN < EXPORT_XML_SOURCE_MAX < HEALTHKIT_STATISTICS`
+
+A later XML import cannot overwrite native statistics. Native
+statistics repair an XML estimate or legacy row in either arrival
+order, and every material reconciled update or resurrection increments
+`syncVersion`. Existing Apple Health `stats:*` rows are migrated to
+`LEGACY_UNKNOWN`; they are never guessed or heuristically promoted.
+To recover an already-corrupted legacy total, the user must re-upload
+the original archive under parser revision 2 or sync native HealthKit
+statistics. Without either input, the correct per-source total cannot
+be reconstructed from the stored aggregate.
+
 ## Source-priority interaction
 
 HealthLog's source ladder is **APPLE_HEALTH ≻ WITHINGS ≻ MANUAL ≻
@@ -169,12 +210,12 @@ max). The defaults live in `src/lib/validations/source-priority.ts:205-220`.
 
 Concrete consequences:
 
-- If you already sync steps from a Withings ScanWatch and also import
-  Apple Health, the Apple-Health-aggregated stream wins for the day —
-  iOS HealthKit folds the watch sensor + iPhone motion data into a
-  single canonical stream, so it is the most complete cumulative
-  source. The Withings rows stay in the database as an audit trail
-  but drop out of the per-day aggregation.
+- Native HealthKit cumulative statistics are the canonical Apple
+  Health stream for a day and outrank both XML estimates and Withings
+  cumulative rows. An XML upload chooses the largest source subtotal
+  only; it never claims to reproduce HealthKit's device de-duplication.
+  Lower-priority rows stay in the database as an audit trail but drop
+  out of the displayed per-day aggregation.
 - If you weigh yourself on a Withings scale and also import Apple
   Health (which received the same reading via the Health Mate iOS
   app), the Withings row wins for display. Apple Health is

messages/de.json

@@ -4519,6 +4519,7 @@
             "failed": "Der Import ist fehlgeschlagen.",
             "rowsImported": "Bisher {count} Einträge importiert",
             "doneSummary": "Fertig — {imported} von {read} Datensätzen importiert, {skipped} klinische Datensätze übersprungen.",
+            "estimateWarning": "Kumulative Summen für {count} Tag(e) wurden anhand der größten Exportquelle geschätzt. Native Apple-Health-Statistiken bleiben maßgeblich.",
             "phase": {
               "queued": "In Warteschlange…",
               "unpacking": "Archiv wird entpackt…",

messages/en.json

@@ -4519,6 +4519,7 @@
             "failed": "The import failed.",
             "rowsImported": "{count} rows imported so far",
             "doneSummary": "Done — {imported} of {read} records imported, {skipped} clinical records skipped.",
+            "estimateWarning": "Cumulative totals for {count} day(s) were estimated from the largest export source. Native Apple Health statistics remain authoritative.",
             "phase": {
               "queued": "Queued…",
               "unpacking": "Unpacking the archive…",

messages/es.json

@@ -4519,6 +4519,7 @@
             "failed": "La importación falló.",
             "rowsImported": "{count} filas importadas hasta ahora",
             "doneSummary": "Listo: {imported} de {read} registros importados, {skipped} registros clínicos omitidos.",
+            "estimateWarning": "Los totales acumulados de {count} día(s) se estimaron a partir de la mayor fuente de la exportación. Las estadísticas nativas de Apple Health siguen siendo la referencia.",
             "phase": {
               "queued": "En cola…",
               "unpacking": "Descomprimiendo el archivo…",

messages/fr.json

@@ -4519,6 +4519,7 @@
             "failed": "L'import a échoué.",
             "rowsImported": "{count} lignes importées jusqu'ici",
             "doneSummary": "Terminé — {imported} enregistrements sur {read} importés, {skipped} enregistrements cliniques ignorés.",
+            "estimateWarning": "Les totaux cumulés de {count} jour(s) ont été estimés à partir de la source d’export la plus élevée. Les statistiques Apple Santé natives restent la référence.",
             "phase": {
               "queued": "En file d'attente…",
               "unpacking": "Décompression de l'archive…",

messages/it.json

@@ -4519,6 +4519,7 @@
             "failed": "L'importazione non è riuscita.",
             "rowsImported": "{count} righe importate finora",
             "doneSummary": "Fatto: {imported} di {read} record importati, {skipped} record clinici saltati.",
+            "estimateWarning": "I totali cumulativi di {count} giorno/i sono stati stimati dalla fonte di esportazione più alta. Le statistiche native di Apple Salute restano autorevoli.",
             "phase": {
               "queued": "In coda…",
               "unpacking": "Estrazione dell'archivio…",

messages/pl.json

@@ -4519,6 +4519,7 @@
             "failed": "Import nie powiódł się.",
             "rowsImported": "Zaimportowano dotąd {count} wierszy",
             "doneSummary": "Gotowe — zaimportowano {imported} z {read} rekordów, pominięto {skipped} rekordów klinicznych.",
+            "estimateWarning": "Łączne wartości dla {count} dni oszacowano na podstawie największego źródła w eksporcie. Natywne statystyki Apple Health pozostają nadrzędne.",
             "phase": {
               "queued": "W kolejce…",
               "unpacking": "Rozpakowywanie archiwum…",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "healthlog",
-  "version": "1.31.6",
+  "version": "1.31.7",
   "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.",
   "license": "PolyForm-Noncommercial-1.0.0",
   "homepage": "https://healthlog.dev",

prisma/migrations/0263_apple_health_aggregate_authority/migration.sql

@@ -0,0 +1,27 @@
+-- Aggregate provenance makes the shared Apple Health `stats:*` identity
+-- authority-aware. Existing rows cannot be classified safely from their stored
+-- shape, so they remain repairable but are never promoted heuristically.
+CREATE TYPE "measurement_aggregation_provenance" AS ENUM (
+  'HEALTHKIT_STATISTICS',
+  'EXPORT_XML_SOURCE_MAX',
+  'LEGACY_UNKNOWN'
+);
+
+ALTER TABLE "measurements"
+  ADD COLUMN "aggregation_provenance" "measurement_aggregation_provenance",
+  ADD COLUMN "aggregation_contributor_count" INTEGER,
+  ADD COLUMN "aggregation_selected_source_hash" TEXT;
+
+UPDATE "measurements"
+SET "aggregation_provenance" = 'LEGACY_UNKNOWN'
+WHERE "source" = 'APPLE_HEALTH'
+  AND "external_id" LIKE 'stats:%';
+
+-- Revision 1 denotes rows created under the pre-authority parser. New binaries
+-- write revision 2 explicitly; the default stays legacy-safe during rollouts.
+ALTER TABLE "import_jobs"
+  ADD COLUMN "parser_revision" INTEGER NOT NULL DEFAULT 1;
+
+UPDATE "import_jobs"
+SET "parser_revision" = 1;
+

prisma/schema.prisma

@@ -1592,6 +1592,14 @@ enum SleepStage {
   @@map("sleep_stage")
 }
 
+enum MeasurementAggregationProvenance {
+  HEALTHKIT_STATISTICS
+  EXPORT_XML_SOURCE_MAX
+  LEGACY_UNKNOWN
+
+  @@map("measurement_aggregation_provenance")
+}
+
 model Measurement {
   id                    String                @id @default(cuid())
   userId                String                @map("user_id")
@@ -1625,6 +1633,18 @@ model Measurement {
   // sample's `metadata.HKWasUserEntered`, Withings model code, …).
   // Stored as a string so the ingest path can replay it back unchanged.
   externalSourceVersion String?               @map("external_source_version")
+  // Authority of an aggregate stored under a `stats:*` identity. NULL for
+  // ordinary point/sample rows. Native HealthKit statistics outrank the
+  // source-day estimate derived from export.xml; migrated ambiguous rows are
+  // explicitly LEGACY_UNKNOWN until a native sync or archive reprocess repairs
+  // them.
+  aggregationProvenance         MeasurementAggregationProvenance? @map("aggregation_provenance")
+  // Distinct source identities contributing to an export.xml source-day
+  // estimate. Source labels themselves are never persisted.
+  aggregationContributorCount   Int?                             @map("aggregation_contributor_count")
+  // SHA-256 of the deterministically selected source identity. Non-identifying
+  // provenance for repeatability without retaining device/source labels.
+  aggregationSelectedSourceHash String?                          @map("aggregation_selected_source_hash")
   // Only set for BLOOD_GLUCOSE measurements. NULL for all others.
   glucoseContext        GlucoseContext?       @map("glucose_context")
   // v1.4.23 — per-stage sleep label. NULL for non-sleep rows; non-NULL
@@ -4284,6 +4304,9 @@ model ImportJob {
   /// kick-off endpoint short-circuit a re-upload of the exact same
   /// file back to the existing job id without re-queueing.
   uploadSha256       String?   @map("upload_sha256")
+  /// Parser semantics revision. Existing archives migrated at revision 1 can
+  /// be deliberately reprocessed when the current parser revision advances.
+  parserRevision     Int       @default(1) @map("parser_revision")
   /// Apple's `ExportDate` attribute string-formatted, parsed during
   /// the unpack phase. NULL until the parser sees it.
   exportedAt         DateTime? @map("exported_at")

public/sw.js

@@ -36,7 +36,7 @@ try {
 // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild.
 const CACHE_VERSION =
   (typeof self !== "undefined" && self.__APP_VERSION__) ||
-  /* @sw-version-fallback */ "v1.31.6";
+  /* @sw-version-fallback */ "v1.31.7";
 const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`;
 const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`;
 // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*`

src/__tests__/i18n-reverse-coverage.test.ts

@@ -245,5 +245,5 @@ describe("i18n reverse coverage", () => {
           "Wire a call site, or add a DB-driven prefix to DYNAMIC_ALLOWLIST_PREFIXES.",
       );
     }
-  });
+  }, 15_000);
 });

src/app/api/admin/import-apple-health-export/__tests__/route.test.ts

@@ -188,9 +188,29 @@ describe("POST /api/admin/import-apple-health-export", () => {
     expect(body.data.jobId).not.toBe("old-failed-job");
     expect(mocks.bossSend).toHaveBeenCalledTimes(1);
     expect(mocks.importJobCreate).toHaveBeenCalledTimes(1);
+    expect(mocks.importJobCreate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: expect.objectContaining({ parserRevision: 2 }),
+      }),
+    );
+    expect(mocks.bossSend).toHaveBeenCalledWith(
+      "apple-health-import-v2",
+      expect.objectContaining({
+        userId: "user-2",
+        triggeredByAdminId: "admin-1",
+      }),
+      expect.objectContaining({ retryLimit: 0 }),
+    );
+    expect(mocks.importJobUpdate).toHaveBeenCalledWith({
+      where: { id: "ij-1" },
+      data: { pgBossJobId: "boss-job-1" },
+    });
     expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
       expect.objectContaining({
-        where: expect.objectContaining({ status: { not: "failed" } }),
+        where: expect.objectContaining({
+          status: { not: "failed" },
+          parserRevision: 2,
+        }),
       }),
     );
     expect(mocks.unlink).not.toHaveBeenCalled();
@@ -211,5 +231,10 @@ describe("POST /api/admin/import-apple-health-export", () => {
     expect(mocks.bossSend).not.toHaveBeenCalled();
     expect(mocks.importJobCreate).not.toHaveBeenCalled();
     expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
+    expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({ parserRevision: 2 }),
+      }),
+    );
   });
 });

src/app/api/admin/import-apple-health-export/route.ts

@@ -21,7 +21,8 @@ import { auditLog } from "@/lib/auth/audit";
 import { checkRateLimit } from "@/lib/rate-limit";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
-  APPLE_HEALTH_IMPORT_QUEUE,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
   APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
@@ -121,6 +122,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
     where: {
       userId: targetUser.id,
       uploadSha256: uploaded.sha256,
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       status: { not: "failed" },
     },
     orderBy: { startedAt: "desc" },
@@ -157,6 +159,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
       status: "queued",
       uploadBytes: uploaded.bytes,
       uploadSha256: uploaded.sha256,
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
     },
   });
 
@@ -171,7 +174,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
   // first run, so a redelivery could only fail on the deleted `/tmp`
   // file and mask the real outcome (see the worker's send-options doc).
   const bossJobId = await boss.send(
-    APPLE_HEALTH_IMPORT_QUEUE,
+    APPLE_HEALTH_IMPORT_V2_QUEUE,
     payload,
     APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   );

src/app/api/import/apple-health-export/__tests__/route.test.ts

@@ -170,10 +170,27 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue
     expect(body.data.jobId).not.toBe("old-failed-job");
     expect(mocks.bossSend).toHaveBeenCalledTimes(1);
     expect(mocks.importJobCreate).toHaveBeenCalledTimes(1);
+    expect(mocks.importJobCreate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: expect.objectContaining({ parserRevision: 2 }),
+      }),
+    );
+    expect(mocks.bossSend).toHaveBeenCalledWith(
+      "apple-health-import-v2",
+      expect.objectContaining({ userId: "user-1" }),
+      expect.objectContaining({ retryLimit: 0 }),
+    );
+    expect(mocks.importJobUpdate).toHaveBeenCalledWith({
+      where: { id: "ij-1" },
+      data: { pgBossJobId: "boss-job-1" },
+    });
     // The dedup lookup must exclude failed rows.
     expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
       expect.objectContaining({
-        where: expect.objectContaining({ status: { not: "failed" } }),
+        where: expect.objectContaining({
+          status: { not: "failed" },
+          parserRevision: 2,
+        }),
       }),
     );
     // A fresh job was staged — nothing to unlink on this path.
@@ -196,5 +213,10 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue
     expect(mocks.importJobCreate).not.toHaveBeenCalled();
     // The freshly staged upload is redundant and must be cleaned up.
     expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
+    expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({ parserRevision: 2 }),
+      }),
+    );
   });
 });

src/app/api/import/apple-health-export/route.ts

@@ -4,7 +4,7 @@
  *
  * Streams the upload directly to disk (`/tmp/healthlog-upload-*.bin`)
  * without buffering the full body in memory, creates an `ImportJob`
- * row in `queued`, and enqueues an `apple-health-import` pg-boss job.
+ * row in `queued`, and enqueues an `apple-health-import-v2` pg-boss job.
  * Returns `{ jobId }` so the client can poll
  * `GET /api/import/apple-health-export/[jobId]/status`.
  *
@@ -29,7 +29,8 @@ import { auditLog } from "@/lib/auth/audit";
 import { checkRateLimit } from "@/lib/rate-limit";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
-  APPLE_HEALTH_IMPORT_QUEUE,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
   APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
@@ -118,6 +119,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
     where: {
       userId: user.id,
       uploadSha256: uploaded.sha256,
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       status: { not: "failed" },
     },
     orderBy: { startedAt: "desc" },
@@ -154,6 +156,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
       status: "queued",
       uploadBytes: uploaded.bytes,
       uploadSha256: uploaded.sha256,
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
     },
   });
 
@@ -167,7 +170,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
   // first run, so a redelivery could only fail on the deleted `/tmp`
   // file and mask the real outcome (see the worker's send-options doc).
   const bossJobId = await boss.send(
-    APPLE_HEALTH_IMPORT_QUEUE,
+    APPLE_HEALTH_IMPORT_V2_QUEUE,
     payload,
     APPLE_HEALTH_IMPORT_SEND_OPTIONS,
   );

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

@@ -516,6 +516,9 @@ async function postBatch(request: NextRequest): Promise<Response> {
             rhythmClassification: p.row
               .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"],
             deviceType: p.row.deviceType as string | null,
+            aggregationProvenance: statsRow ? "HEALTHKIT_STATISTICS" : null,
+            aggregationContributorCount: null,
+            aggregationSelectedSourceHash: null,
           },
           { exactExternalMatch: statsRow ? "update" : "duplicate" },
         ),

src/components/settings/__tests__/import-panel.test.tsx

@@ -26,6 +26,8 @@ vi.mock("next/navigation", () => ({
   usePathname: () => "/settings/export",
 }));
 
+import type { Locale } from "@/lib/i18n/config";
+import { AppleHealthEstimateWarning } from "../import-panel/apple-health-import-card";
 import { I18nProvider } from "@/lib/i18n/context";
 import {
   ImportPanel,
@@ -33,9 +35,10 @@ import {
   EXAMPLE_CSV,
   parseImportJson,
 } from "../import-panel";
+
 import { parseCsvMeasurements } from "@/lib/import/csv-measurements";
 
-function render(node: React.ReactElement, locale: "en" | "de" = "en") {
+function render(node: React.ReactElement, locale: Locale = "en") {
   const queryClient = new QueryClient({
     defaultOptions: { queries: { retry: false, staleTime: Infinity } },
   });
@@ -98,6 +101,20 @@ describe("<ImportPanel> — SSR smoke", () => {
   });
 });
 
+describe("<AppleHealthEstimateWarning>", () => {
+  it.each(["en", "de", "es", "fr", "it", "pl"] as const)(
+    "renders localized estimate disclosure for %s",
+    (locale) => {
+      const html = render(<AppleHealthEstimateWarning days={3} />, locale);
+      expect(html).toContain('data-testid="apple-health-estimate-warning"');
+      expect(html).toContain("3");
+      expect(html).not.toContain(
+        "settings.sections.export.import.appleHealth.estimateWarning",
+      );
+    },
+  );
+});
+
 describe("EXAMPLE_IMPORT payload", () => {
   it("round-trips through JSON and stays parseable", () => {
     const serialised = JSON.stringify(EXAMPLE_IMPORT, null, 2);

src/components/settings/import-panel/apple-health-import-card.tsx

@@ -27,10 +27,31 @@ interface JobStatus {
   result: {
     totals?: { recordsRead?: number; rowsUpserted?: number };
     clinical?: { skipped?: number };
+    cumulativeEstimates?: { days?: number; rows?: number };
   } | null;
   failureReason: string | null;
 }
 
+export function AppleHealthEstimateWarning({ days }: { days: number }) {
+  const { t } = useTranslations();
+  return (
+    <p
+      data-testid="apple-health-estimate-warning"
+      className="text-foreground flex items-start gap-2 text-xs"
+    >
+      <AlertCircle
+        className="text-warning mt-0.5 size-3.5 shrink-0"
+        aria-hidden="true"
+      />
+      <span>
+        {t("settings.sections.export.import.appleHealth.estimateWarning", {
+          count: days,
+        })}
+      </span>
+    </p>
+  );
+}
+
 export function AppleHealthImportCard() {
   const { t } = useTranslations();
   const fileInputRef = useRef<HTMLInputElement>(null);
@@ -233,6 +254,11 @@ export function AppleHealthImportCard() {
             </span>
           </div>
         )}
+        {isDone && (status?.result?.cumulativeEstimates?.days ?? 0) > 0 && (
+          <AppleHealthEstimateWarning
+            days={status?.result?.cumulativeEstimates?.days ?? 0}
+          />
+        )}
         {(isFailed || uploadError) && (
           <p
             role="alert"

src/lib/ai/__tests__/openai-client.test.ts

@@ -60,6 +60,9 @@ describe("OpenAIClient", () => {
     expect(body.response_format).toEqual({ type: "json_object" });
     // No seed passed → field omitted from the body entirely.
     expect(body).not.toHaveProperty("seed");
+    expect(body.max_tokens).toBe(1000);
+    expect(body.temperature).toBe(0.3);
+    expect(body).not.toHaveProperty("max_completion_tokens");
   });
 
   it("forces JSON mode only when responseFormat is json AND no tools (M-1)", async () => {
@@ -147,6 +150,118 @@ describe("OpenAIClient", () => {
     expect(body.seed).toBe(1234);
   });
 
+  it.each([
+    ["gpt-5.4-nano", 1000, "https://api.openai.com/v1"],
+    ["gpt-5-2025-08-07", 321, "https://api.openai.com/v1"],
+    ["o1", 321, "https://api.openai.com/v1"],
+    ["o3-mini", 321, "https://api.openai.com/v1"],
+    ["o4-mini-2025-04-16", 321, "https://api.openai.com/v1"],
+    ["gpt-5.4-mini", 321, "https://API.OPENAI.COM:443/v1/"],
+  ])(
+    "uses the canonical modern token wire for %s",
+    async (model, expectedBudget, baseUrl) => {
+      const mockFetch = vi.fn().mockResolvedValue({
+        ok: true,
+        json: () =>
+          Promise.resolve({
+            choices: [{ message: { content: "ok" } }],
+            usage: { total_tokens: 5 },
+          }),
+      });
+      vi.stubGlobal("fetch", mockFetch);
+
+      const client = new OpenAIClient({
+        apiKey: "sk-test",
+        model,
+        baseUrl,
+      });
+
+      await client.generateCompletion(
+        singleUserTurn({
+          system: "test",
+          user: "test",
+          temperature: 0.7,
+          ...(expectedBudget === 1000 ? {} : { maxTokens: expectedBudget }),
+          seed: 1234,
+        }),
+      );
+
+      const body = JSON.parse(mockFetch.mock.calls[0][1].body);
+      expect(body.max_completion_tokens).toBe(expectedBudget);
+      expect(body).not.toHaveProperty("max_tokens");
+      expect(body).not.toHaveProperty("temperature");
+      expect(body).not.toHaveProperty("seed");
+    },
+  );
+
+  it("keeps the legacy sampling wire for canonical GPT-4 models", async () => {
+    const mockFetch = vi.fn().mockResolvedValue({
+      ok: true,
+      json: () =>
+        Promise.resolve({
+          choices: [{ message: { content: "ok" } }],
+          usage: { total_tokens: 5 },
+        }),
+    });
+    vi.stubGlobal("fetch", mockFetch);
+
+    const client = new OpenAIClient({
+      apiKey: "sk-test",
+      model: "gpt-4o-mini",
+      baseUrl: "https://api.openai.com/v1",
+    });
+
+    await client.generateCompletion(
+      singleUserTurn({
+        system: "test",
+        user: "test",
+        temperature: 0.7,
+        maxTokens: 321,
+        seed: 1234,
+      }),
+    );
+
+    const body = JSON.parse(mockFetch.mock.calls[0][1].body);
+    expect(body.max_tokens).toBe(321);
+    expect(body.temperature).toBe(0.7);
+    expect(body.seed).toBe(1234);
+    expect(body).not.toHaveProperty("max_completion_tokens");
+  });
+
+  it("keeps the legacy sampling wire for GPT-5-named models on compatible gateways", async () => {
+    const mockFetch = vi.fn().mockResolvedValue({
+      ok: true,
+      json: () =>
+        Promise.resolve({
+          choices: [{ message: { content: "ok" } }],
+          usage: { total_tokens: 5 },
+        }),
+    });
+    vi.stubGlobal("fetch", mockFetch);
+
+    const client = new OpenAIClient({
+      apiKey: "sk-test",
+      model: "gpt-5.4-nano",
+      baseUrl: "https://gateway.example.com/v1",
+    });
+
+    await client.generateCompletion(
+      singleUserTurn({
+        system: "test",
+        user: "test",
+        temperature: 0.7,
+        maxTokens: 321,
+        seed: 1234,
+      }),
+    );
+
+    const body = JSON.parse(mockFetch.mock.calls[0][1].body);
+    expect(body.max_tokens).toBe(321);
+    expect(body.temperature).toBe(0.7);
+    expect(body.seed).toBe(1234);
+    expect(body).not.toHaveProperty("max_completion_tokens");
+  });
+
   it("throws on non-ok response", async () => {
     vi.stubGlobal(
       "fetch",

src/lib/ai/openai-capabilities.ts

@@ -0,0 +1,67 @@
+export type OpenAIChatCompletionsCapabilities =
+  | {
+      tokenBudgetField: "max_completion_tokens";
+      supportsSamplingControls: false;
+    }
+  | {
+      tokenBudgetField: "max_tokens";
+      supportsSamplingControls: true;
+    };
+
+const MODERN_CANONICAL_CAPABILITIES = {
+  tokenBudgetField: "max_completion_tokens",
+  supportsSamplingControls: false,
+} as const;
+
+const LEGACY_COMPATIBLE_CAPABILITIES = {
+  tokenBudgetField: "max_tokens",
+  supportsSamplingControls: true,
+} as const;
+
+function isModernCanonicalModel(model: string): boolean {
+  const normalizedModel = model.toLowerCase();
+
+  return (
+    normalizedModel === "gpt-5" ||
+    normalizedModel.startsWith("gpt-5-") ||
+    normalizedModel.startsWith("gpt-5.") ||
+    normalizedModel === "o1" ||
+    normalizedModel.startsWith("o1-") ||
+    normalizedModel === "o3" ||
+    normalizedModel.startsWith("o3-") ||
+    normalizedModel === "o4" ||
+    normalizedModel.startsWith("o4-")
+  );
+}
+
+function isCanonicalOpenAIEndpoint(baseUrl: string): boolean {
+  try {
+    const url = new URL(baseUrl);
+    return (
+      url.protocol === "https:" &&
+      url.hostname === "api.openai.com" &&
+      url.port === "" &&
+      url.pathname.replace(/\/+$/, "") === "/v1" &&
+      url.username === "" &&
+      url.password === "" &&
+      url.search === "" &&
+      url.hash === ""
+    );
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * Select Chat Completions request capabilities from both endpoint and model.
+ * Custom OpenAI-compatible gateways retain the broadly supported legacy wire,
+ * even when their model name resembles a modern OpenAI model.
+ */
+export function selectOpenAIChatCompletionsCapabilities(
+  baseUrl: string,
+  model: string,
+): OpenAIChatCompletionsCapabilities {
+  return isCanonicalOpenAIEndpoint(baseUrl) && isModernCanonicalModel(model)
+    ? MODERN_CANONICAL_CAPABILITIES
+    : LEGACY_COMPATIBLE_CAPABILITIES;
+}

src/lib/ai/openai-client.ts

@@ -1,5 +1,6 @@
 import { safeFetch } from "@/lib/safe-fetch";
 import type { AIProvider, CompletionParams, CompletionResult } from "./types";
+import { selectOpenAIChatCompletionsCapabilities } from "./openai-capabilities";
 import {
   buildOpenAIMessages,
   buildOpenAITools,
@@ -60,6 +61,18 @@ export class OpenAIClient implements AIProvider {
     // therefore stay out of JSON mode, and every tool round is non-JSON by
     // construction. Insight/extraction callers opt in with `responseFormat:"json"`.
     const useJsonFormat = params.responseFormat === "json" && !hasTools;
+    const capabilities = selectOpenAIChatCompletionsCapabilities(
+      this.config.baseUrl,
+      this.config.model,
+    );
+    const tokenBudget = params.maxTokens ?? 1000;
+    const capabilityParams = capabilities.supportsSamplingControls
+      ? {
+          [capabilities.tokenBudgetField]: tokenBudget,
+          temperature: params.temperature ?? 0.3,
+          ...(params.seed !== undefined ? { seed: params.seed } : {}),
+        }
+      : { [capabilities.tokenBudgetField]: tokenBudget };
 
     const res = await safeFetch(
       url,
@@ -72,15 +85,10 @@ export class OpenAIClient implements AIProvider {
         body: JSON.stringify({
           model: this.config.model,
           messages,
-          temperature: params.temperature ?? 0.3,
-          max_tokens: params.maxTokens ?? 1000,
+          ...capabilityParams,
           ...(useJsonFormat
             ? { response_format: { type: "json_object" } }
             : {}),
-          // Deterministic seed for reproducible reference output; omitted
-          // (undefined → dropped by JSON.stringify) when the caller does
-          // not pin one (e.g. the seedless daily-briefing re-roll).
-          ...(params.seed !== undefined ? { seed: params.seed } : {}),
           ...(tools ? { tools } : {}),
           ...(params.toolChoice ? { tool_choice: params.toolChoice } : {}),
         }),

src/lib/daily/__tests__/digest.test.ts

@@ -236,6 +236,49 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => {
     expect(dose?.actions[0].href).toBe("/medications");
   });
 
+  it("emits one medication-specific card and link per actionable candidate", () => {
+    const d = buildDailyDigest(
+      input({
+        medsToday: meds({
+          activeCount: 2,
+          dueCandidates: [
+            {
+              medicationId: "med-ramipril",
+              medicationName: "Ramipril",
+              dueAt: new Date(NOW.getTime() - 30 * 60_000).toISOString(),
+              overdue: true,
+              availableFrom: new Date(
+                NOW.getTime() - 2 * 60 * 60_000,
+              ).toISOString(),
+            },
+            {
+              medicationId: "med-mounjaro",
+              medicationName: "Mounjaro",
+              dueAt: new Date(NOW.getTime() - 30 * 60_000).toISOString(),
+              overdue: false,
+              availableFrom: new Date(
+                NOW.getTime() - 24 * 60 * 60_000,
+              ).toISOString(),
+            },
+          ],
+        }),
+      }),
+      t,
+    );
+
+    const doses = d.worthALook.filter((item) => item.kind === "dose_window");
+    expect(doses).toHaveLength(2);
+    expect(doses.map((item) => item.status)).toEqual(["warning", "info"]);
+    expect(doses.map((item) => item.body)).toEqual([
+      t("daily.item.doseWindow.overdueBodyNamed", { name: "Ramipril" }),
+      t("daily.item.doseWindow.bodyNamed", { name: "Mounjaro" }),
+    ]);
+    expect(doses.map((item) => item.actions[0]?.href)).toEqual([
+      "/medications?highlight=med-ramipril",
+      "/medications?highlight=med-mounjaro",
+    ]);
+  });
+
   it("does NOT emit a dose-window item when the medications module is off", () => {
     const d = buildDailyDigest(
       input({
@@ -334,6 +377,58 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => {
     expect(d.worthALook.some((i) => i.kind === "dose_window")).toBe(false);
   });
 
+  it("shows a weekly future slot once its engine-derived availability window opens", () => {
+    const d = buildDailyDigest(
+      input({
+        medsToday: meds({
+          dueCandidates: [
+            {
+              medicationId: "med-weekly",
+              medicationName: "Weekly dose",
+              dueAt: new Date(NOW.getTime() + 24 * 60 * 60_000).toISOString(),
+              overdue: false,
+              availableFrom: new Date(NOW.getTime() - 60_000).toISOString(),
+            },
+          ],
+        }),
+      }),
+      t,
+    );
+
+    expect(
+      d.worthALook.filter((item) => item.kind === "dose_window"),
+    ).toHaveLength(1);
+  });
+
+  it("respects an explicit weekly availability boundary exactly", () => {
+    const candidate = {
+      medicationId: "med-weekly",
+      medicationName: "Weekly dose",
+      dueAt: new Date(NOW.getTime() + 24 * 60 * 60_000).toISOString(),
+      overdue: false,
+      availableFrom: NOW.toISOString(),
+    };
+
+    const atBoundary = buildDailyDigest(
+      input({ medsToday: meds({ dueCandidates: [candidate] }) }),
+      t,
+    );
+    const beforeBoundary = buildDailyDigest(
+      input({
+        now: new Date(NOW.getTime() - 1),
+        medsToday: meds({ dueCandidates: [candidate] }),
+      }),
+      t,
+    );
+
+    expect(
+      atBoundary.worthALook.some((item) => item.kind === "dose_window"),
+    ).toBe(true);
+    expect(
+      beforeBoundary.worthALook.some((item) => item.kind === "dose_window"),
+    ).toBe(false);
+  });
+
   it("keeps a dose beyond the takeable lead off the rail", () => {
     const d = buildDailyDigest(
       input({
@@ -350,9 +445,7 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => {
     expect(d.worthALook.some((i) => i.kind === "dose_window")).toBe(false);
   });
 
-  it("renders neither face for a cached block whose anchor has since passed", () => {
-    // `MedsTodayBlock` contract: a past `nextDueAt` with `nextDueOverdue:
-    // false` means the anchor passed AFTER the block was built.
+  it("keeps a stale cached non-overdue scalar calm instead of dropping or escalating it", () => {
     const d = buildDailyDigest(
       input({
         medsToday: meds({
@@ -363,7 +456,11 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => {
       }),
       t,
     );
-    expect(d.worthALook.some((i) => i.kind === "dose_window")).toBe(false);
+    const dose = d.worthALook.find((item) => item.kind === "dose_window");
+    expect(dose?.status).toBe("info");
+    expect(dose?.body).toBe(
+      t("daily.item.doseWindow.bodyNamed", { name: "Ramipril" }),
+    );
   });
 
   it("maps one sync-issue item per broken integration", () => {

src/lib/daily/digest.ts

@@ -320,76 +320,85 @@ export const DOSE_DUE_LOOKAHEAD_MS =
   60_000;
 
 /**
- * Dose-window item — the day's medication action, in two distinct faces.
+ * Compose one card for every medication candidate the canonical scheduling
+ * engine says is currently available. The additive collection is authoritative
+ * when present; scalar fields remain the fallback for older cached snapshots.
  *
- * OVERDUE: an open slot whose anchor has passed. `warning` status, past-due
- * wording — the one genuinely time-critical daily action.
- *
- * DUE: a slot the user can already take (within `DOSE_DUE_LOOKAHEAD_MS`) but
- * that is not yet late. `info` status, "due soon" wording — present on the
- * card because Today is a surface the user OPENS to see what the day asks of
- * them, not a notification that arrives uninvited. The no-nagging rule governs
- * PUSH (the reminder cron still fires only on overdue); a deliberately-opened
- * card that stays silent about a dose the user can take right now is simply
- * withholding the answer the surface exists to give.
- *
- * Neither face is manufactured on a settled day: once every scheduled dose is
- * resolved, the next display-due slot lies beyond the lookahead and nothing is
- * emitted. A `nextDueAt` in the PAST with `nextDueOverdue: false` is the
- * documented cached-block shape (`MedsTodayBlock`) — the anchor passed after
- * the block was built — and renders as neither face until a fresh block lands.
- *
- * Gated on the `medications` module; carries a single log-dose action.
+ * `overdue` is never inferred from the wall clock. A cached non-overdue
+ * candidate whose anchor has passed stays calm/informational until refresh.
  */
-function buildDoseWindowItem(
+function buildDoseWindowItems(
   meds: MedsTodayBlock,
   modules: DigestModuleMap,
   now: Date,
   t: Translate,
-): PriorityItem | null {
-  if (!moduleEnabled(modules, "medications")) return null;
+): PriorityItem[] {
+  if (!moduleEnabled(modules, "medications")) return [];
 
-  const name = meds.nextDueMedicationName;
-  const id = meds.nextDueMedicationId;
-  const action = {
-    labelKey: "daily.action.logDose",
-    intent: "dose.log",
-    // Deep-link straight to the due medication's card so the tap lands on the
-    // right one instead of a generic list the user then has to scan (the id is
-    // known server-side whenever the name is). The bare list stays the honest
-    // fallback for an older cached block that predates the id field.
-    href: id ? `/medications?highlight=${id}` : "/medications",
-  };
+  const candidates =
+    meds.dueCandidates !== undefined
+      ? meds.dueCandidates
+      : meds.nextDueOverdue || meds.nextDueAt
+        ? [
+            {
+              medicationId: meds.nextDueMedicationId,
+              medicationName: meds.nextDueMedicationName,
+              dueAt: meds.nextDueAt,
+              overdue: meds.nextDueOverdue,
+              availableFrom: null,
+            },
+          ]
+        : [];
+  const items: PriorityItem[] = [];
 
-  if (meds.nextDueOverdue) {
-    return {
-      kind: "dose_window",
-      title: t("daily.item.doseWindow.overdueTitle"),
-      body: name
-        ? t("daily.item.doseWindow.overdueBodyNamed", { name })
-        : t("daily.item.doseWindow.overdueBody"),
-      status: "warning",
-      actions: [action],
-      moduleKey: "medications",
+  for (const candidate of candidates) {
+    if (!candidate.overdue) {
+      const dueAt = candidate.dueAt ? Date.parse(candidate.dueAt) : NaN;
+      if (!Number.isFinite(dueAt)) continue;
+
+      if (candidate.availableFrom !== null) {
+        const availableFrom = Date.parse(candidate.availableFrom);
+        if (!Number.isFinite(availableFrom) || now.getTime() < availableFrom) {
+          continue;
+        }
+      } else if (dueAt - now.getTime() > DOSE_DUE_LOOKAHEAD_MS) {
+        continue;
+      }
+    }
+
+    const name = candidate.medicationName;
+    const id = candidate.medicationId;
+    const action = {
+      labelKey: "daily.action.logDose",
+      intent: "dose.log",
+      href: id ? `/medications?highlight=${id}` : "/medications",
     };
+    items.push(
+      candidate.overdue
+        ? {
+            kind: "dose_window",
+            title: t("daily.item.doseWindow.overdueTitle"),
+            body: name
+              ? t("daily.item.doseWindow.overdueBodyNamed", { name })
+              : t("daily.item.doseWindow.overdueBody"),
+            status: "warning",
+            actions: [action],
+            moduleKey: "medications",
+          }
+        : {
+            kind: "dose_window",
+            title: t("daily.item.doseWindow.title"),
+            body: name
+              ? t("daily.item.doseWindow.bodyNamed", { name })
+              : t("daily.item.doseWindow.body"),
+            status: "info",
+            actions: [action],
+            moduleKey: "medications",
+          },
+    );
   }
 
-  if (!meds.nextDueAt) return null;
-  const dueAt = Date.parse(meds.nextDueAt);
-  if (!Number.isFinite(dueAt)) return null;
-  const lead = dueAt - now.getTime();
-  if (lead < 0 || lead > DOSE_DUE_LOOKAHEAD_MS) return null;
-
-  return {
-    kind: "dose_window",
-    title: t("daily.item.doseWindow.title"),
-    body: name
-      ? t("daily.item.doseWindow.bodyNamed", { name })
-      : t("daily.item.doseWindow.body"),
-    status: "info",
-    actions: [action],
-    moduleKey: "medications",
-  };
+  return items;
 }
 
 /** One rail item per broken integration — reconnect to keep data current. */
@@ -735,13 +744,9 @@ export function buildDailyDigest(
   // check-up least urgent. Bounded to 3 — a check-in the cap crowds out
   // resurfaces on a following day (within its window), never lost.
   const worthALook: PriorityItem[] = [];
-  const dose = buildDoseWindowItem(
-    input.medsToday,
-    input.modules,
-    input.now,
-    t,
+  worthALook.push(
+    ...buildDoseWindowItems(input.medsToday, input.modules, input.now, t),
   );
-  if (dose) worthALook.push(dose);
   // A freshly-reached milestone is rare and one-per-day — surface it ahead of
   // the ambient sync / check-in items so the calm reward is not buried, but
   // below an overdue dose (the one genuinely time-critical daily action).

src/lib/dashboard/__tests__/meds-today.test.ts

@@ -208,6 +208,7 @@ describe("buildMedsTodayBlock — tally", () => {
       nextDueOverdue: false,
       nextDueMedicationName: null,
       nextDueMedicationId: null,
+      dueCandidates: [],
     });
   });
 
@@ -317,6 +318,63 @@ describe("buildMedsTodayBlock — next due (real engine)", () => {
     expect(block.nextDueOverdue).toBe(false);
   });
 
+  it("preserves equal-time medications and orders the overdue candidate first", async () => {
+    medications = [
+      medication({
+        id: "med-mounjaro",
+        name: "Mounjaro",
+        startsOn: new Date("2026-06-10T00:00:00.000Z"),
+        schedules: [
+          dailySchedule({
+            id: "sched-mounjaro",
+            medicationId: "med-mounjaro",
+            windowStart: "11:00",
+            windowEnd: "11:00",
+            timesOfDay: ["11:00"],
+            rrule: null,
+            rollingIntervalDays: 7,
+          }),
+        ],
+      }),
+      medication({
+        id: "med-ramipril",
+        name: "Ramipril",
+        schedules: [
+          dailySchedule({
+            id: "sched-ramipril",
+            medicationId: "med-ramipril",
+            windowStart: "11:00",
+            windowEnd: "11:00",
+            timesOfDay: ["11:00"],
+          }),
+        ],
+      }),
+    ];
+
+    const block = await buildMedsTodayBlock(fakePrisma, "user-1", BERLIN, NOW);
+
+    expect(block.dueCandidates).toEqual([
+      {
+        medicationId: "med-ramipril",
+        medicationName: "Ramipril",
+        dueAt: "2026-06-10T09:00:00.000Z",
+        overdue: true,
+        availableFrom: "2026-06-10T07:00:00.000Z",
+      },
+      {
+        medicationId: "med-mounjaro",
+        medicationName: "Mounjaro",
+        dueAt: "2026-06-10T09:00:00.000Z",
+        overdue: false,
+        availableFrom: "2026-06-09T08:00:00.000Z",
+      },
+    ]);
+    expect(block.nextDueAt).toBe("2026-06-10T09:00:00.000Z");
+    expect(block.nextDueOverdue).toBe(true);
+    expect(block.nextDueMedicationName).toBe("Ramipril");
+    expect(block.nextDueMedicationId).toBe("med-ramipril");
+  });
+
   it("returns null next-due when no schedule has an upcoming slot", async () => {
     // Course ended last month — the engine yields nothing.
     medications = [

src/lib/dashboard/meds-today.ts

@@ -44,6 +44,15 @@ import {
 } from "@/lib/medications/scheduling/next-due";
 import { getUserTodayBounds } from "@/lib/tz/local-day";
 
+export interface MedsTodayDueCandidate {
+  medicationId: string;
+  medicationName: string;
+  dueAt: string;
+  overdue: boolean;
+  /** Canonical attribution-band start for this slot (ISO8601). */
+  availableFrom: string;
+}
+
 export interface MedsTodayBlock {
   /** Active medications (paused courses excluded). */
   activeCount: number;
@@ -75,6 +84,11 @@ export interface MedsTodayBlock {
    * list.
    */
   nextDueMedicationId: string | null;
+  /**
+   * Every active medication's current display-due candidate. Optional only so
+   * older cached snapshots remain valid; the live builder always emits it.
+   */
+  dueCandidates?: MedsTodayDueCandidate[];
 }
 
 export async function buildMedsTodayBlock(
@@ -183,13 +197,7 @@ export async function buildMedsTodayBlock(
       eraStartByMedId.set(f.medicationId, f._max.validUntil);
   }
 
-  // Earliest display-due across active medications. An open overdue
-  // slot's anchor is in the past, so the minimum naturally prefers it
-  // over any future slot.
-  let nextDueAt: Date | null = null;
-  let nextDueOverdue = false;
-  let nextDueMedicationName: string | null = null;
-  let nextDueMedicationId: string | null = null;
+  const dueCandidates: MedsTodayDueCandidate[] = [];
   for (const m of medications) {
     const display = computeDisplayDue({
       medication: {
@@ -207,22 +215,32 @@ export async function buildMedsTodayBlock(
       eraStart: eraStartByMedId.get(m.id) ?? null,
     });
     if (!display) continue;
-    if (nextDueAt === null || display.at.getTime() < nextDueAt.getTime()) {
-      nextDueAt = display.at;
-      nextDueOverdue = display.overdue;
-      nextDueMedicationName = m.name;
-      nextDueMedicationId = m.id;
-    }
+    dueCandidates.push({
+      medicationId: m.id,
+      medicationName: m.name,
+      dueAt: display.at.toISOString(),
+      overdue: display.overdue,
+      availableFrom: (display.availableFrom ?? display.at).toISOString(),
+    });
   }
+  dueCandidates.sort(
+    (a, b) =>
+      Number(b.overdue) - Number(a.overdue) ||
+      a.availableFrom.localeCompare(b.availableFrom) ||
+      a.dueAt.localeCompare(b.dueAt) ||
+      a.medicationId.localeCompare(b.medicationId),
+  );
+  const next = dueCandidates[0] ?? null;
 
   return {
     activeCount: medications.length,
     scheduledToday: todayEvents.length,
     takenToday,
     skippedToday,
-    nextDueAt: nextDueAt ? nextDueAt.toISOString() : null,
-    nextDueOverdue,
-    nextDueMedicationName,
-    nextDueMedicationId,
+    nextDueAt: next?.dueAt ?? null,
+    nextDueOverdue: next?.overdue ?? false,
+    nextDueMedicationName: next?.medicationName ?? null,
+    nextDueMedicationId: next?.medicationId ?? null,
+    dueCandidates,
   };
 }

src/lib/jobs/__tests__/apple-health-import-legacy-bridge.test.ts

@@ -0,0 +1,92 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { Job } from "pg-boss";
+
+const mocks = vi.hoisted(() => ({
+  findUnique: vi.fn(),
+  update: vi.fn(),
+  send: vi.fn(),
+  getGlobalBoss: vi.fn(),
+}));
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    importJob: {
+      findUnique: mocks.findUnique,
+      update: mocks.update,
+    },
+  },
+  toJson: (value: unknown) => value,
+}));
+
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: mocks.getGlobalBoss,
+}));
+
+import {
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
+  APPLE_HEALTH_IMPORT_SEND_OPTIONS,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+  migrateLegacyAppleHealthImport,
+  type AppleHealthImportPayload,
+} from "../apple-health-import-worker";
+
+const payload: AppleHealthImportPayload = {
+  userId: "user-1",
+  uploadPath: "/tmp/export.zip",
+  uploadBytes: 100,
+  enqueuedAt: "2026-07-21T10:00:00.000Z",
+};
+
+function legacyJob(): Job<AppleHealthImportPayload> {
+  return {
+    id: "legacy-boss-1",
+    data: payload,
+  } as Job<AppleHealthImportPayload>;
+}
+
+beforeEach(() => {
+  vi.resetAllMocks();
+  mocks.getGlobalBoss.mockReturnValue({ send: mocks.send });
+  mocks.send.mockResolvedValue("v2-boss-1");
+  mocks.findUnique.mockResolvedValue({
+    id: "import-1",
+    status: "queued",
+    parserRevision: 1,
+  });
+  mocks.update.mockResolvedValue({ id: "import-1" });
+});
+
+describe("migrateLegacyAppleHealthImport", () => {
+  it("moves a legacy job onto the v2 queue while preserving its mirror id", async () => {
+    await migrateLegacyAppleHealthImport(legacyJob());
+
+    expect(mocks.send).toHaveBeenCalledWith(
+      APPLE_HEALTH_IMPORT_V2_QUEUE,
+      payload,
+      APPLE_HEALTH_IMPORT_SEND_OPTIONS,
+    );
+    expect(mocks.update).toHaveBeenCalledWith({
+      where: { id: "import-1" },
+      data: {
+        pgBossJobId: "v2-boss-1",
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+        status: "queued",
+        failureReason: null,
+        completedAt: null,
+      },
+    });
+  });
+
+  it("does not migrate a terminal legacy mirror", async () => {
+    mocks.findUnique.mockResolvedValue({
+      id: "import-1",
+      status: "done",
+      parserRevision: 1,
+    });
+
+    await migrateLegacyAppleHealthImport(legacyJob());
+
+    expect(mocks.send).not.toHaveBeenCalled();
+    expect(mocks.update).not.toHaveBeenCalled();
+  });
+});

src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts

@@ -19,13 +19,38 @@ vi.mock("@/lib/db", () => ({
   toJson: (value: unknown) => value,
 }));
 
-import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "../apple-health-import-worker";
+import {
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
+  APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
+  APPLE_HEALTH_IMPORT_SEND_OPTIONS,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+} from "../apple-health-import-worker";
 
 const ROUTE_FILES = [
   "src/app/api/import/apple-health-export/route.ts",
   "src/app/api/admin/import-apple-health-export/route.ts",
 ];
 
+const maintenanceSource = readFileSync(
+  join(process.cwd(), "src/lib/jobs/reminder/register-maintenance.ts"),
+  "utf8",
+);
+const workerSource = readFileSync(
+  join(process.cwd(), "src/lib/jobs/apple-health-import-worker.ts"),
+  "utf8",
+);
+const migrationSource = readFileSync(
+  join(
+    process.cwd(),
+    "prisma/migrations/0263_apple_health_aggregate_authority/migration.sql",
+  ),
+  "utf8",
+);
+const schemaSource = readFileSync(
+  join(process.cwd(), "prisma/schema.prisma"),
+  "utf8",
+);
+
 describe("apple health import — pg-boss send policy", () => {
   it("disables retries and widens the expiration window", () => {
     // retryLimit 0: the staged upload is unlinked by the first run, so
@@ -42,8 +67,52 @@ describe("apple health import — pg-boss send policy", () => {
     for (const file of ROUTE_FILES) {
       const source = readFileSync(join(process.cwd(), file), "utf8");
       expect(source).toMatch(
-        /boss\.send\(\s*APPLE_HEALTH_IMPORT_QUEUE,\s*payload,\s*APPLE_HEALTH_IMPORT_SEND_OPTIONS,?\s*\)/,
+        /boss\.send\(\s*APPLE_HEALTH_IMPORT_V2_QUEUE,\s*payload,\s*APPLE_HEALTH_IMPORT_SEND_OPTIONS,?\s*\)/,
       );
     }
   });
 });
+
+describe("apple health import — parser revision boundary", () => {
+  it("isolates revision-2 sends from legacy workers", () => {
+    expect(APPLE_HEALTH_IMPORT_PARSER_REVISION).toBe(2);
+    expect(APPLE_HEALTH_IMPORT_V2_QUEUE).toBe("apple-health-import-v2");
+    expect(APPLE_HEALTH_IMPORT_LEGACY_QUEUE).toBe("apple-health-import");
+  });
+
+  it("drains revision-2 jobs separately while bridging legacy backlog", () => {
+    const allQueues = maintenanceSource.match(
+      /const allQueues\s*=\s*\[([\s\S]*?)\];/,
+    );
+    expect(allQueues).not.toBeNull();
+    expect(allQueues![1]).toMatch(/\bAPPLE_HEALTH_IMPORT_V2_QUEUE\b/);
+    expect(allQueues![1]).toMatch(/\bAPPLE_HEALTH_IMPORT_LEGACY_QUEUE\b/);
+    expect(maintenanceSource).toMatch(
+      /boss\.work[\s\S]{0,200}APPLE_HEALTH_IMPORT_V2_QUEUE[\s\S]{0,400}handleAppleHealthImport/,
+    );
+    expect(maintenanceSource).toMatch(
+      /boss\.work[\s\S]{0,200}APPLE_HEALTH_IMPORT_LEGACY_QUEUE[\s\S]{0,400}migrateLegacyAppleHealthImport/,
+    );
+  });
+
+  it("keeps the database default at revision 1 for legacy binaries", () => {
+    expect(migrationSource).toMatch(
+      /ADD COLUMN "parser_revision" INTEGER NOT NULL DEFAULT 1/,
+    );
+    expect(migrationSource).not.toMatch(
+      /ALTER COLUMN "parser_revision" SET DEFAULT 2/,
+    );
+    expect(schemaSource).toMatch(
+      /parserRevision\s+Int\s+@default\(1\)\s+@map\("parser_revision"\)/,
+    );
+  });
+
+  it("explicitly creates revision-2 stand-ins and reconciles only revision-2 mirrors", () => {
+    expect(workerSource).toMatch(
+      /importJob\.create\([\s\S]{0,400}parserRevision:\s*APPLE_HEALTH_IMPORT_PARSER_REVISION/,
+    );
+    expect(workerSource).toMatch(
+      /importJob\.findMany\(\{\s*where:\s*\{[\s\S]{0,200}parserRevision:\s*APPLE_HEALTH_IMPORT_PARSER_REVISION/,
+    );
+  });
+});

src/lib/jobs/apple-health-import-worker.ts

@@ -1,5 +1,5 @@
 /**
- * v1.4.34 — pg-boss handler for the `apple-health-import` queue.
+ * v1.4.34 — pg-boss handler for the revision-2 Apple Health import queue.
  *
  * The kick-off endpoint (`POST /api/import/apple-health-export`,
  * `POST /api/admin/import-apple-health-export`) writes the upload to
@@ -38,8 +38,14 @@ import {
 } from "@/lib/measurements/import-apple-health-export";
 import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
 
-/** Queue name for the Apple Health export ingest. */
-export const APPLE_HEALTH_IMPORT_QUEUE = "apple-health-import";
+/** Queue name isolated from revision-1 workers during rolling deployments. */
+export const APPLE_HEALTH_IMPORT_V2_QUEUE = "apple-health-import-v2";
+
+/** Queue used by pre-revision-2 binaries; new workers only bridge its backlog. */
+export const APPLE_HEALTH_IMPORT_LEGACY_QUEUE = "apple-health-import";
+
+/** Parser semantics carried by every newly-created ImportJob. */
+export const APPLE_HEALTH_IMPORT_PARSER_REVISION = 2;
 
 /** Concurrency cap per worker host. */
 export const APPLE_HEALTH_IMPORT_CONCURRENCY = 1;
@@ -96,6 +102,58 @@ function getWorkerPrisma(): PrismaClient {
   return workerPrismaSingleton ?? prisma;
 }
 
+/**
+ * Move a job claimed from the legacy queue onto the revision-2 queue without
+ * parsing it under the wrong revision. The existing ImportJob id remains the
+ * status handle; only its pg-boss mirror id and truthful parser revision move.
+ */
+export async function migrateLegacyAppleHealthImport(
+  job: Job<AppleHealthImportPayload>,
+): Promise<void> {
+  const prisma = getWorkerPrisma();
+  let importJob = await prisma.importJob.findUnique({
+    where: { pgBossJobId: job.id },
+  });
+  if (!importJob) {
+    const { promise, resolve } = Promise.withResolvers<void>();
+    setTimeout(resolve, 250);
+    await promise;
+    importJob = await prisma.importJob.findUnique({
+      where: { pgBossJobId: job.id },
+    });
+  }
+  if (!importJob) {
+    throw new Error(`No ImportJob mirror for legacy pg-boss job ${job.id}`);
+  }
+  if (importJob.status === "done" || importJob.status === "failed") {
+    return;
+  }
+
+  const boss = getGlobalBoss();
+  if (!boss) {
+    throw new Error("Cannot bridge legacy Apple Health import without pg-boss");
+  }
+  const v2BossJobId = await boss.send(
+    APPLE_HEALTH_IMPORT_V2_QUEUE,
+    job.data,
+    APPLE_HEALTH_IMPORT_SEND_OPTIONS,
+  );
+  if (!v2BossJobId) {
+    throw new Error("Failed to enqueue bridged Apple Health import");
+  }
+
+  await prisma.importJob.update({
+    where: { id: importJob.id },
+    data: {
+      pgBossJobId: v2BossJobId,
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+      status: "queued",
+      failureReason: null,
+      completedAt: null,
+    },
+  });
+}
+
 /**
  * Persist a progress snapshot onto the mirror `ImportJob` row. The
  * worker calls this every `PROGRESS_TICK_RECORDS` records parsed +
@@ -151,6 +209,7 @@ export async function handleAppleHealthImport(
         pgBossJobId: job.id,
         status: "queued",
         uploadBytes,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       },
     });
   }
@@ -348,6 +407,8 @@ const LIVE_PG_BOSS_STATES = new Set(["active", "created", "retry"]);
  * it last shut down — flip it to `failed` with `interrupted_by_restart`
  * so the operator can re-upload (and, post-issue-#486, so the kick-off
  * dedup no longer short-circuits future re-uploads onto a dead job).
+ * Revision-1 rows are deliberately excluded: during a rolling deployment
+ * their backing jobs live on the legacy queue and remain owned by old workers.
  *
  * The reconcile is deliberately NOT unconditional. In a multi-replica
  * or rolling-deploy topology a booting worker must not flip a row that
@@ -373,7 +434,10 @@ const LIVE_PG_BOSS_STATES = new Set(["active", "created", "retry"]);
 export async function reconcileOrphanImportJobs(): Promise<void> {
   const prisma = getWorkerPrisma();
   const candidates = await prisma.importJob.findMany({
-    where: { status: { in: ["unpacking", "parsing", "upserting"] } },
+    where: {
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+      status: { in: ["unpacking", "parsing", "upserting"] },
+    },
     select: { id: true, pgBossJobId: true, updatedAt: true },
   });
   if (candidates.length === 0) return;
@@ -395,7 +459,7 @@ export async function reconcileOrphanImportJobs(): Promise<void> {
     let live = false;
     try {
       const job = await boss.getJobById(
-        APPLE_HEALTH_IMPORT_QUEUE,
+        APPLE_HEALTH_IMPORT_V2_QUEUE,
         row.pgBossJobId,
       );
       live = job !== null && LIVE_PG_BOSS_STATES.has(job.state);

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

@@ -38,9 +38,11 @@ import {
   type IntakeAutoSkipPayload,
 } from "@/lib/jobs/intake-auto-skip";
 import {
-  APPLE_HEALTH_IMPORT_QUEUE,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+  APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
   APPLE_HEALTH_IMPORT_CONCURRENCY,
   handleAppleHealthImport,
+  migrateLegacyAppleHealthImport,
   type AppleHealthImportPayload,
 } from "@/lib/jobs/apple-health-import-worker";
 import {
@@ -304,7 +306,8 @@ const allQueues = [
   // this entry the schedule silently no-ops and pending rows older than
   // 24 h pile up unflipped.
   INTAKE_AUTO_SKIP_QUEUE,
-  APPLE_HEALTH_IMPORT_QUEUE,
+  APPLE_HEALTH_IMPORT_V2_QUEUE,
+  APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
   // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues
   // one job per user holding two live intake rows that snap to the same
   // canonical slot (the pre-fix REMINDER-pending + API-taken pair). Also
@@ -472,9 +475,10 @@ const schedules: ScheduleEntry[] = [
  *     become one. That is exactly the class of silent work-dropping a policy is
  *     supposed to prevent, so the queue keeps `standard` until the enqueue side
  *     gives the explicit-range variant a key of its own.
- *   - APPLE_HEALTH_IMPORT_QUEUE, DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send
- *     keylessly by design (each import / backup / detection run is a distinct
- *     unit of work). A policy would coalesce independent runs.
+ *   - APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
+ *     DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send keylessly by design (each
+ *     import / backup / detection run is a distinct unit of work). A policy
+ *     would coalesce independent runs.
  */
 const queuePolicies: QueuePolicyTable = {
   // Per-user, boot- and cron-discovery driven, self-converging one-shots. The
@@ -672,7 +676,7 @@ export async function registerMaintenanceQueues(
   // caps at 1 because the parse loop is CPU-bound and a concurrent
   // second 1 GB import would race the first for RSS.
   await boss.work<AppleHealthImportPayload>(
-    APPLE_HEALTH_IMPORT_QUEUE,
+    APPLE_HEALTH_IMPORT_V2_QUEUE,
     { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY },
     async (jobs) => {
       // pg-boss v12 work callbacks always receive an array (batched
@@ -683,6 +687,18 @@ export async function registerMaintenanceQueues(
       }
     },
   );
+  // Drain pre-revision-2 backlog by moving each claimed legacy job onto the
+  // isolated v2 queue. This preserves the ImportJob status id without letting
+  // the current parser execute under a revision-1 mirror.
+  await boss.work<AppleHealthImportPayload>(
+    APPLE_HEALTH_IMPORT_LEGACY_QUEUE,
+    { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        await migrateLegacyAppleHealthImport(job);
+      }
+    },
+  );
 
   // v1.8.2 — duplicate dose-slot cleanup worker. The boot enqueue helper
   // below sends one job per user holding two live intake rows that snap

src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts

@@ -356,7 +356,13 @@ describe("drainPerSampleCumulative — cutoffHours", () => {
 // externalId first, then adopts it in place (`update`) or mints a fresh row
 // (`create`).
 describe("drainPerSampleCumulative — late-sync fold into existing total", () => {
-  function buildFoldMock(existingTotal: number | null) {
+  function buildFoldMock(
+    existingTotal: number | null,
+    aggregationProvenance:
+      | "HEALTHKIT_STATISTICS"
+      | "EXPORT_XML_SOURCE_MAX"
+      | "LEGACY_UNKNOWN" = "LEGACY_UNKNOWN",
+  ) {
     const findManyUser = vi
       .fn()
       .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]);
@@ -414,7 +420,11 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () =
         if ("externalId" in args.where) {
           return existingTotal === null
             ? null
-            : { id: "stats-row", value: existingTotal };
+            : {
+                id: "stats-row",
+                value: existingTotal,
+                aggregationProvenance,
+              };
         }
         if ("measuredAt" in args.where) return null; // no index-B collision
         return { unit: "count" }; // resolveCanonicalUnit
@@ -436,6 +446,7 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () =
       } as unknown as PrismaClient,
       update,
       create,
+      deleteMany,
     };
   }
 
@@ -451,6 +462,10 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () =
     expect(update).toHaveBeenCalledTimes(1);
     const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } };
     expect(updateArg.data.value).toBe(9500);
+    expect(updateArg.data).toMatchObject({
+      aggregationProvenance: "LEGACY_UNKNOWN",
+      syncVersion: { increment: 1 },
+    });
   });
 
   it("mints the partial sum when no collapsed total exists yet", async () => {
@@ -464,8 +479,29 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () =
     expect(create).toHaveBeenCalledTimes(1);
     const createArg = create.mock.calls[0]?.[0] as { data: { value: number } };
     expect(createArg.data.value).toBe(500);
+    expect(createArg.data).toMatchObject({
+      aggregationProvenance: "LEGACY_UNKNOWN",
+    });
   });
 
+  it.each(["HEALTHKIT_STATISTICS", "EXPORT_XML_SOURCE_MAX"] as const)(
+    "drains late source rows without changing %s authority",
+    async (aggregationProvenance) => {
+      const { prisma, update, create, deleteMany } = buildFoldMock(
+        9_000,
+        aggregationProvenance,
+      );
+
+      await drainPerSampleCumulative(prisma, { log: () => {} });
+
+      expect(update).not.toHaveBeenCalled();
+      expect(create).not.toHaveBeenCalled();
+      expect(deleteMany).toHaveBeenCalledWith({
+        where: { id: { in: ["late-1", "late-2"], not: "stats-row" } },
+      });
+    },
+  );
+
   it("recomputes the DAY rollup bucket for the drained day so the stale sum self-heals", async () => {
     vi.mocked(recomputeBucketsForMeasurement).mockClear();
     const { prisma } = buildFoldMock(null);
@@ -745,11 +781,10 @@ describe("drainPerSampleCumulative — concurrent fold does not double-count (F2
 
     // The advisory lock fired for the bucket.
     expect($queryRaw).toHaveBeenCalledTimes(1);
-    // The row is adopted in place and left at 9000 — NOT 9500. Pre-fix the
-    // merge added the stale `reducedValue` (500) onto the committed total.
+    // The row is left at 9000 — NOT 9500 — without a non-material update or
+    // syncVersion bump. Pre-fix the merge added the stale reducedValue (500)
+    // onto the committed total.
     expect(create).not.toHaveBeenCalled();
-    expect(update).toHaveBeenCalledTimes(1);
-    const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } };
-    expect(updateArg.data.value).toBe(9000);
+    expect(update).not.toHaveBeenCalled();
   });
 });

src/lib/measurements/__tests__/import-apple-health-export.test.ts

@@ -15,7 +15,7 @@ vi.mock("@/lib/arrivals/emit-shared", () => ({
   emitDataArrival: vi.fn().mockResolvedValue(undefined),
 }));
 
-import { Prisma } from "@/generated/prisma/client";
+import { Prisma, type PrismaClient } from "@/generated/prisma/client";
 import {
   hashSampleKey,
   parseRecordValue,
@@ -27,6 +27,10 @@ import {
 import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
 import { emitDataArrival } from "@/lib/arrivals/emit-shared";
+import {
+  APPLE_HEALTH_TYPE_MAP,
+  CUMULATIVE_HK_TYPES,
+} from "../apple-health-mapping";
 
 beforeEach(() => {
   vi.clearAllMocks();
@@ -173,10 +177,42 @@ function makeFakePrisma(
     workouts.push({ id: `w${nextId++}`, ...create });
   };
 
-  return {
+  const fake = {
     _measurements: measurements,
     _workouts: workouts,
+    $queryRaw: async () => [],
+    $executeRawUnsafe: async () => 0,
+    $transaction: async (
+      callback: (tx: unknown) => Promise<unknown>,
+    ): Promise<unknown> => callback(fake),
     measurement: {
+      create: async ({ data }: { data: Row }) => {
+        injectMeasurementRace(data);
+        if (
+          measurements.some(
+            (measurement) =>
+              (measurement.userId === data.userId &&
+                measurement.type === data.type &&
+                measurement.source === data.source &&
+                measurement.externalId === data.externalId) ||
+              naturalKeyMatch(measurement, data),
+          )
+        ) {
+          throw new Prisma.PrismaClientKnownRequestError(
+            "Unique constraint failed",
+            { code: "P2002", clientVersion: "test" },
+          );
+        }
+        const created: Row = {
+          id: `m${nextId++}`,
+          syncVersion: 1,
+          deletedAt: null,
+          sleepStage: null,
+          ...data,
+        };
+        measurements.push(created);
+        return created;
+      },
       createManyAndReturn: async ({ data }: { data: Row[] }) => {
         const returned: Row[] = [];
         for (const create of data) {
@@ -210,20 +246,61 @@ function makeFakePrisma(
       }: {
         where: {
           userId: string;
-          source: string;
-          OR: Array<Record<string, unknown>>;
+          source?: unknown;
+          OR?: Array<Record<string, unknown>>;
         };
       }) => {
-        return measurements
-          .filter((m) => {
-            if (m.userId !== where.userId || m.source !== where.source)
-              return false;
-            return where.OR.some(
-              (clause) =>
-                m.type === clause.type && m.externalId === clause.externalId,
-            );
-          })
-          .map((m) => ({ type: m.type, externalId: m.externalId }));
+        const externalClause = where.OR?.find(
+          (clause) => typeof clause.externalId === "string",
+        );
+        const externalId =
+          typeof externalClause?.externalId === "string"
+            ? externalClause.externalId
+            : null;
+        if (
+          externalId?.startsWith("stats:") &&
+          measurementRaceExternalIds.has(externalId) &&
+          !injectedMeasurementRaces.has(externalId)
+        ) {
+          const naturalClause = where.OR?.find(
+            (clause) => clause.measuredAt instanceof Date,
+          );
+          injectedMeasurementRaces.add(externalId);
+          measurements.push({
+            id: `m${nextId++}`,
+            userId: where.userId,
+            type: externalClause?.type,
+            source: externalClause?.source,
+            externalId,
+            measuredAt: naturalClause?.measuredAt,
+            sleepStage: naturalClause?.sleepStage ?? null,
+            deletedAt: null,
+            syncVersion: 1,
+            aggregationProvenance: "HEALTHKIT_STATISTICS",
+          });
+        }
+        return measurements.filter((measurement) => {
+          if (measurement.userId !== where.userId) return false;
+          if (
+            where.source !== undefined &&
+            measurement.source !== where.source
+          ) {
+            return false;
+          }
+          if (!where.OR) return true;
+          return where.OR.some(
+            (clause) =>
+              (clause.type === undefined || measurement.type === clause.type) &&
+              (clause.source === undefined ||
+                measurement.source === clause.source) &&
+              (clause.externalId === undefined ||
+                measurement.externalId === clause.externalId) &&
+              (clause.measuredAt === undefined ||
+                sameMeasuredAt(measurement.measuredAt, clause.measuredAt)) &&
+              (clause.sleepStage === undefined ||
+                sameStage(measurement.sleepStage, clause.sleepStage)),
+          );
+        });
       },
       findUnique: async ({
         where,
@@ -244,7 +321,15 @@ function makeFakePrisma(
       // Natural-key probe used by the rescue path. Tombstone-inclusive: no
       // `deletedAt` filter is applied here or by the caller.
       findFirst: async ({ where }: { where: Record<string, unknown> }) => {
-        const match = measurements.find((m) => naturalKeyMatch(m, where));
+        const excludedId =
+          typeof where.id === "object" && where.id !== null && "not" in where.id
+            ? where.id.not
+            : undefined;
+        const match = measurements.find(
+          (measurement) =>
+            measurement.id !== excludedId &&
+            naturalKeyMatch(measurement, where),
+        );
         return match ? { id: match.id } : null;
       },
       update: async ({
@@ -276,7 +361,19 @@ function makeFakePrisma(
             { code: "P2025", clientVersion: "test" },
           );
         }
-        measurements[idx] = { ...measurements[idx], ...data };
+        const syncVersion =
+          typeof data.syncVersion === "object" &&
+          data.syncVersion !== null &&
+          "increment" in data.syncVersion &&
+          typeof data.syncVersion.increment === "number"
+            ? Number(measurements[idx].syncVersion ?? 1) +
+              data.syncVersion.increment
+            : data.syncVersion;
+        measurements[idx] = {
+          ...measurements[idx],
+          ...data,
+          ...(syncVersion === undefined ? {} : { syncVersion }),
+        };
         return measurements[idx];
       },
       upsert: async ({
@@ -398,6 +495,7 @@ function makeFakePrisma(
       },
     },
   };
+  return fake;
 }
 
 describe("parseRecordValue", () => {
@@ -494,6 +592,143 @@ describe("hashSampleKey", () => {
     expect(a).not.toBe(b);
   });
 });
+const cumulativeMappings = Object.values(APPLE_HEALTH_TYPE_MAP).filter(
+  (mapping) => CUMULATIVE_HK_TYPES.has(mapping.measurementType),
+);
+
+function cumulativeExportXml(
+  hkIdentifier: string,
+  hkUnit: string,
+  rows: Array<{ value: number; sourceName: string; hour: number }>,
+): string {
+  return `<?xml version="1.0" encoding="UTF-8"?>
+<HealthData locale="en_US">
+${rows
+  .map(
+    ({ value, sourceName, hour }) => `  <Record type="${hkIdentifier}"
+          unit="${hkUnit}"
+          startDate="2026-05-14 ${String(hour).padStart(2, "0")}:00:00 +0200"
+          endDate="2026-05-14 ${String(hour).padStart(2, "0")}:30:00 +0200"
+          value="${value}"
+          sourceName="${sourceName}"
+          sourceVersion="1.0"/>`,
+  )
+  .join("\n")}
+</HealthData>`;
+}
+
+async function importCumulativeFixture(xml: string) {
+  const tmp = mkdtempSync(join(tmpdir(), "healthlog-parser-cumulative-test-"));
+  const xmlPath = join(tmp, "export.xml");
+  writeFileSync(xmlPath, xml);
+  const prisma = makeFakePrisma();
+  const result = await streamParseExportXml({
+    xmlPath,
+    userId: "user-cumulative",
+    userTimezone: "Europe/Berlin",
+    // The fake implements the importer-facing subset of PrismaClient.
+    prisma: prisma as unknown as PrismaClient,
+  });
+  return { prisma, result };
+}
+
+describe("streamParseExportXml — cumulative source-day estimates", () => {
+  it("covers exactly the six cumulative HealthKit measurement types", () => {
+    expect(CUMULATIVE_HK_TYPES.size).toBe(6);
+    expect(
+      new Set(cumulativeMappings.map((mapping) => mapping.measurementType)),
+    ).toEqual(CUMULATIVE_HK_TYPES);
+  });
+
+  it("stores 8,526 rather than source-blind 15,608 for overlapping iPhone and Zepp steps", async () => {
+    const { prisma, result } = await importCumulativeFixture(
+      cumulativeExportXml("HKQuantityTypeIdentifierStepCount", "count", [
+        { value: 8_526, sourceName: "iPhone", hour: 8 },
+        { value: 7_082, sourceName: "Zepp", hour: 9 },
+      ]),
+    );
+
+    expect(prisma._measurements).toHaveLength(1);
+    expect(prisma._measurements[0]).toMatchObject({
+      value: 8_526,
+      aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+      aggregationContributorCount: 2,
+    });
+    expect(prisma._measurements[0]?.aggregationSelectedSourceHash).toMatch(
+      /^[a-f0-9]{64}$/,
+    );
+    expect(result.cumulativeEstimates).toEqual({ days: 1, rows: 1 });
+  });
+
+  it.each(cumulativeMappings)(
+    "selects the largest source-day subtotal for $measurementType",
+    async (mapping) => {
+      const { prisma } = await importCumulativeFixture(
+        cumulativeExportXml(mapping.hkIdentifier, mapping.hkUnit, [
+          { value: 30, sourceName: "Source A", hour: 8 },
+          { value: 20, sourceName: "Source B", hour: 9 },
+        ]),
+      );
+
+      expect(prisma._measurements).toHaveLength(1);
+      expect(prisma._measurements[0]).toMatchObject({
+        type: mapping.measurementType,
+        value: 30,
+        aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+        aggregationContributorCount: 2,
+      });
+    },
+  );
+
+  it.each(cumulativeMappings)(
+    "sums non-overlapping $measurementType rows from the same source",
+    async (mapping) => {
+      const { prisma } = await importCumulativeFixture(
+        cumulativeExportXml(mapping.hkIdentifier, mapping.hkUnit, [
+          { value: 30, sourceName: "Source A", hour: 8 },
+          { value: 20, sourceName: "Source A", hour: 9 },
+        ]),
+      );
+
+      expect(prisma._measurements).toHaveLength(1);
+      expect(prisma._measurements[0]).toMatchObject({
+        type: mapping.measurementType,
+        value: 50,
+        aggregationContributorCount: 1,
+      });
+    },
+  );
+  it("folds unattributed cumulative records into one bounded source bucket", async () => {
+    const { prisma } = await importCumulativeFixture(`<?xml version="1.0"?>
+<HealthData locale="en_US">
+  <Record type="HKQuantityTypeIdentifierStepCount" unit="count"
+          startDate="2026-05-14 08:00:00 +0200"
+          endDate="2026-05-14 08:30:00 +0200" value="60"/>
+  <Record type="HKQuantityTypeIdentifierStepCount" unit="count"
+          startDate="2026-05-14 09:00:00 +0200"
+          endDate="2026-05-14 09:30:00 +0200" value="40"/>
+</HealthData>`);
+
+    expect(prisma._measurements).toHaveLength(1);
+    expect(prisma._measurements[0]).toMatchObject({
+      value: 100,
+      aggregationContributorCount: 1,
+    });
+  });
+
+  it("rejects a source-day subtotal outside the aggregate range", async () => {
+    const { prisma, result } = await importCumulativeFixture(
+      cumulativeExportXml("HKQuantityTypeIdentifierStepCount", "count", [
+        { value: 150_000, sourceName: "Source A", hour: 8 },
+        { value: 150_000, sourceName: "Source A", hour: 9 },
+      ]),
+    );
+
+    expect(prisma._measurements).toHaveLength(0);
+    expect(result.unknown["ACTIVITY_STEPS::aggregate_out_of_range"]).toBe(1);
+    expect(result.cumulativeEstimates).toEqual({ days: 0, rows: 0 });
+  });
+});
 
 describe("streamParseExportXml — end-to-end", () => {
   it("ingests the tiny fixture into per-type / workout / deferred / unknown buckets", async () => {
@@ -615,8 +850,8 @@ describe("streamParseExportXml — statement-level insertion identity", () => {
       xmlPath,
       userId: "user-1",
       userTimezone: "Europe/Berlin",
-      // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      prisma: prisma as any,
+      // The fake implements the importer-facing subset of PrismaClient.
+      prisma: prisma as unknown as PrismaClient,
       spotBatchSize: 10,
     });
 
@@ -625,9 +860,11 @@ describe("streamParseExportXml — statement-level insertion identity", () => {
       inserted: 1,
       updated: 0,
     });
+    // The concurrently visible native statistics row is authoritative, so the
+    // XML estimate is an intentional no-op.
     expect(result.perType.ACTIVITY_STEPS).toMatchObject({
       inserted: 0,
-      updated: 1,
+      updated: 0,
     });
 
     const arrivalRows = vi.mocked(emitInsertedMeasurementArrivals).mock
@@ -907,7 +1144,7 @@ describe("streamParseExportXml — memory ceiling", () => {
           ` unit="count"` +
           ` startDate="${ts}"` +
           ` endDate="${ts}"` +
-          ` value="${(i % 200) + 1}"/>\n`,
+          ` value="1"/>\n`,
       );
     }
     writeFileSync(xmlPath, header + rows.join("") + footer);
@@ -931,6 +1168,11 @@ describe("streamParseExportXml — memory ceiling", () => {
     // The cumulative fold collapses all 10 000 records into one daily
     // bucket (they all land on the same date in the fixture).
     expect(result.perType.ACTIVITY_STEPS?.inserted).toBe(1);
+    expect(prisma._measurements).toHaveLength(1);
+    expect(prisma._measurements[0]).toMatchObject({
+      value: 10_000,
+      aggregationContributorCount: 1,
+    });
 
     // Heap delta should be modest — we tolerate a generous 100 MB so a
     // GC-related spike doesn't flake the test, but a regression that

src/lib/measurements/__tests__/reconcile-external-measurement.test.ts

@@ -39,6 +39,9 @@ function row(overrides: Record<string, unknown> = {}): Record<string, unknown> {
     notesEncrypted: null,
     externalId: "sleep:new",
     externalSourceVersion: null,
+    aggregationProvenance: null,
+    aggregationContributorCount: null,
+    aggregationSelectedSourceHash: null,
     glucoseContext: null,
     sleepStage: "DEEP",
     rhythmClassification: null,
@@ -210,6 +213,167 @@ describe("reconcileExternalMeasurement", () => {
     expect(measurement.update).not.toHaveBeenCalled();
   });
 
+  it("keeps a higher-authority native aggregate unchanged when XML arrives later", async () => {
+    const native = row({
+      id: "native-stats",
+      type: "ACTIVITY_STEPS",
+      value: 8_600,
+      unit: "steps",
+      source: "APPLE_HEALTH",
+      measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+      sleepStage: null,
+      externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+      aggregationProvenance: "HEALTHKIT_STATISTICS",
+      aggregationContributorCount: null,
+      aggregationSelectedSourceHash: null,
+    });
+    const { tx, measurement } = transaction([native]);
+
+    const verdict = await reconcileExternalMeasurement(
+      tx,
+      desired({
+        type: "ACTIVITY_STEPS",
+        value: 8_526,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+        sleepStage: null,
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+        aggregationContributorCount: 2,
+        aggregationSelectedSourceHash: "source-hash",
+      }),
+      { exactExternalMatch: "update" },
+    );
+
+    expect(verdict).toMatchObject({
+      status: "duplicate",
+      row: { id: "native-stats" },
+    });
+    expect(measurement.update).not.toHaveBeenCalled();
+  });
+
+  it("retires a lower-authority external collision without changing the native natural-key row", async () => {
+    const xmlExternal = row({
+      id: "xml-external",
+      type: "ACTIVITY_STEPS",
+      value: 8_526,
+      unit: "steps",
+      source: "APPLE_HEALTH",
+      measuredAt: new Date("2026-07-20T09:00:00.000Z"),
+      sleepStage: null,
+      externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+      aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+    });
+    const nativeNatural = row({
+      id: "native-natural",
+      type: "ACTIVITY_STEPS",
+      value: 8_600,
+      unit: "steps",
+      source: "APPLE_HEALTH",
+      measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+      sleepStage: null,
+      externalId: "stats:legacy-native-id",
+      externalSourceVersion: "native-v1",
+      deviceType: "Apple Watch",
+      aggregationProvenance: "HEALTHKIT_STATISTICS",
+    });
+    const { tx, measurement } = transaction([xmlExternal, nativeNatural]);
+
+    const verdict = await reconcileExternalMeasurement(
+      tx,
+      desired({
+        type: "ACTIVITY_STEPS",
+        value: 8_526,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+        sleepStage: null,
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+      }),
+      { exactExternalMatch: "update" },
+    );
+
+    expect(verdict).toMatchObject({
+      status: "duplicate",
+      row: {
+        id: "native-natural",
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+      },
+      retiredCollisionId: "xml-external",
+    });
+    expect(measurement.update).toHaveBeenCalledTimes(1);
+    expect(measurement.update).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { id: "xml-external" },
+        data: expect.objectContaining({
+          measuredAt: new Date(0),
+          externalId:
+            "retired:xml-external:stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+          deletedAt: expect.any(Date),
+          syncVersion: { increment: 1 },
+        }),
+      }),
+    );
+  });
+
+  it("allows native statistics to repair a legacy aggregate", async () => {
+    const legacy = row({
+      id: "legacy-stats",
+      type: "ACTIVITY_STEPS",
+      value: 15_608,
+      unit: "steps",
+      source: "APPLE_HEALTH",
+      measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+      sleepStage: null,
+      externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+      aggregationProvenance: "LEGACY_UNKNOWN",
+    });
+    const { tx, measurement } = transaction([legacy]);
+    measurement.update.mockResolvedValue(
+      row({
+        ...legacy,
+        value: 8_600,
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+      }),
+    );
+
+    const verdict = await reconcileExternalMeasurement(
+      tx,
+      desired({
+        type: "ACTIVITY_STEPS",
+        value: 8_600,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+        sleepStage: null,
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+        aggregationContributorCount: null,
+        aggregationSelectedSourceHash: null,
+      }),
+      { exactExternalMatch: "update" },
+    );
+
+    expect(verdict).toMatchObject({
+      status: "updated",
+      row: { id: "legacy-stats" },
+    });
+    expect(measurement.update).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { id: "legacy-stats" },
+        data: expect.objectContaining({
+          value: 8_600,
+          aggregationProvenance: "HEALTHKIT_STATISTICS",
+          aggregationContributorCount: null,
+          aggregationSelectedSourceHash: null,
+          syncVersion: { increment: 1 },
+        }),
+      }),
+    );
+  });
+
   it("rolls back its savepoint and returns failed for a non-benign write error", async () => {
     const existing = row({
       id: "exact-row",

src/lib/measurements/drain-per-sample-cumulative.ts

@@ -6,20 +6,18 @@
  *
  * Scope per `CUMULATIVE_HK_TYPES`:
  *   ACTIVITY_STEPS, ACTIVE_ENERGY_BURNED, FLIGHTS_CLIMBED,
- *   WALKING_RUNNING_DISTANCE, TIME_IN_DAYLIGHT
+ *   WALKING_RUNNING_DISTANCE, TIME_IN_DAYLIGHT, FALL_COUNT
  *
  * Per user × type × calendar day (anchored to `User.timezone`):
- *   1. SELECT all `Measurement` rows with `source = 'APPLE_HEALTH'` and
- *      `type = <cumulative type>` and `measuredAt` within that user's
- *      calendar day boundary and `externalId NOT LIKE 'stats:%'`.
- *   2. If 0 rows → continue.
- *   3. If 1 row whose externalId already follows the `stats:...`
- *      shape → continue (already collapsed).
- *   4. SUM the values; pick canonical timestamp = midday UTC of the
- *      user's calendar day (matches the Withings activity-sync
- *      convention per R-A §5 / W17b).
- *   5. UPSERT a row with `externalId = dailyStatsExternalId(...)`,
- *      `value = sumValue`, `measuredAt = canonicalTimestamp`.
+ *   1. SELECT live APPLE_HEALTH rows for the cumulative type and day.
+ *   2. Bucket only non-`stats:` source rows and sum their values.
+ *   3. Resolve the canonical `stats:<HKType>:<day>` row under an advisory
+ *      transaction lock.
+ *   4. If it is HEALTHKIT_STATISTICS or EXPORT_XML_SOURCE_MAX, leave its
+ *      value/provenance untouched and drain only the source rows.
+ *   5. Otherwise preserve legacy late-increment semantics, explicitly stamp
+ *      LEGACY_UNKNOWN, and increment syncVersion; fresh/adopted rows are also
+ *      LEGACY_UNKNOWN.
  *   6. DELETE the original per-sample rows in the same transaction.
  *
  * Designed to be invoked by both:
@@ -32,7 +30,11 @@
  * committing.
  */
 import { Prisma } from "@/generated/prisma/client";
-import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import type {
+  MeasurementAggregationProvenance,
+  MeasurementType,
+  PrismaClient,
+} from "@/generated/prisma/client";
 import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors";
 import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups";
 
@@ -71,6 +73,11 @@ export type { PerSampleRow };
 
 /** Prefix marking an already-collapsed daily-stats row. */
 const DAILY_STATS_PREFIX = "stats:";
+const AUTHORITATIVE_AGGREGATION_PROVENANCE =
+  new Set<MeasurementAggregationProvenance>([
+    "HEALTHKIT_STATISTICS",
+    "EXPORT_XML_SOURCE_MAX",
+  ]);
 
 /**
  * v1.4.38 — canonical cutoff for the nightly scheduled drain. Rows
@@ -284,7 +291,7 @@ export async function drainPerSampleCumulative(
           // present, this is the existing collapsed daily total.
           const eidRow = await tx.measurement.findFirst({
             where: { userId, type, source: "APPLE_HEALTH", externalId },
-            select: { id: true, value: true },
+            select: { id: true, value: true, aggregationProvenance: true },
             orderBy: { id: "asc" },
           });
           // Index-B row: occupies the canonical local-noon instant
@@ -297,7 +304,7 @@ export async function drainPerSampleCumulative(
               measuredAt: canonicalTimestamp,
               sleepStage: null,
             },
-            select: { id: true },
+            select: { id: true, aggregationProvenance: true },
             orderBy: { id: "asc" },
           });
 
@@ -305,66 +312,63 @@ export async function drainPerSampleCumulative(
           // and stamping the target externalId onto it would collide with it.
           const adoptTarget = eidRow ?? slotRow;
 
-          // Fold semantics. When the index-A `stats:` total already exists,
-          // late per-sample rows are GENUINE additional readings, so the
-          // correct merge is `existing + late sum` (a blind overwrite with the
-          // partial late sum would shrink the day — the original samples are
-          // already hard-deleted). When adopting a NON-stats slot row (a
-          // per-sample row that fell on local noon, itself part of this
-          // bucket), its value is already inside `reducedValue`, so the merged
-          // value is just `reducedValue`. A fresh day mints `reducedValue`.
-          //
-          // Concurrency guard (paired with the advisory lock above): a sibling
-          // fold that committed just before this one may have already merged
-          // some or all of these per-sample rows into `eidRow` and hard-deleted
-          // them. Trusting the pre-scan `reducedValue` for the increment would
-          // then re-add a sibling's already-folded work → the permanent 2×.
-          // Re-read the still-present contributors inside the locked
-          // transaction: whatever survives is the genuine un-folded increment
-          // (all of `reducedValue` on a clean single run; zero for a
-          // serialised-second run whose rows the winner already drained).
+          // Only legacy aggregates may retain the historical late-increment
+          // behaviour. Native HealthKit statistics and export.xml source-max
+          // estimates are authoritative results from their own pipelines:
+          // this maintenance drain may remove their legacy source rows, but
+          // must never alter their value or provenance.
+          const adoptProvenance = adoptTarget?.aggregationProvenance;
+          const protectedAggregate =
+            adoptProvenance != null &&
+            AUTHORITATIVE_AGGREGATION_PROVENANCE.has(adoptProvenance);
+
+          // Re-read contributors under the advisory lock. A serialized loser
+          // sees zero live rows and therefore performs no canonical update.
           const liveIncrement =
-            eidRow !== null
+            eidRow !== null && !protectedAggregate
               ? (
                   await tx.measurement.findMany({
                     where: { id: { in: sourceRowIds } },
                     select: { value: true },
                   })
-                ).reduce((sum, r) => sum + r.value, 0)
+                ).reduce((sum, row) => sum + row.value, 0)
               : 0;
           const mergedValue =
             eidRow !== null ? eidRow.value + liveIncrement : reducedValue;
 
           let canonicalRowId: string;
           if (adoptTarget) {
-            // Pin the adopted row to local-noon only when the canonical slot
-            // is free or already this row — a DIFFERENT row occupying the slot
-            // would otherwise collide on index B. In that rare case (a tz/DST
-            // shift between mints) the row keeps its existing instant; the
-            // `stats:` externalId is the identity, measuredAt is secondary.
-            const slotIsFreeForTarget =
-              slotRow === null || slotRow.id === adoptTarget.id;
-            await tx.measurement.update({
-              where: { id: adoptTarget.id },
-              data: {
-                value: mergedValue,
-                externalId,
-                deletedAt: null,
-                ...(slotIsFreeForTarget
-                  ? { measuredAt: canonicalTimestamp }
-                  : {}),
-              },
-            });
             canonicalRowId = adoptTarget.id;
+            if (!protectedAggregate && (eidRow === null || liveIncrement > 0)) {
+              // Pin the adopted row to local-noon only when the canonical slot
+              // is free or already this row. A different row occupying the
+              // slot would collide on the natural identity.
+              const slotIsFreeForTarget =
+                slotRow === null || slotRow.id === adoptTarget.id;
+              await tx.measurement.update({
+                where: { id: adoptTarget.id },
+                data: {
+                  value: mergedValue,
+                  externalId,
+                  deletedAt: null,
+                  aggregationProvenance: "LEGACY_UNKNOWN",
+                  aggregationContributorCount: null,
+                  aggregationSelectedSourceHash: null,
+                  syncVersion: { increment: 1 },
+                  ...(slotIsFreeForTarget
+                    ? { measuredAt: canonicalTimestamp }
+                    : {}),
+                },
+              });
+            }
           } else {
             const created = await tx.measurement.create({
               data: {
                 userId,
                 type,
                 value: reducedValue,
-                // pick the canonical unit from an existing row; the per-sample
-                // rows all carry the same unit on a given type. dayRows always
-                // has ≥1 row (empty buckets are skipped).
+                // Pick the canonical unit from an existing row; the per-sample
+                // rows all carry the same unit on a given type.
                 unit:
                   dayRows[0]?.value !== undefined
                     ? await resolveCanonicalUnit(tx, userId, type)
@@ -372,6 +376,7 @@ export async function drainPerSampleCumulative(
                 source: "APPLE_HEALTH",
                 measuredAt: canonicalTimestamp,
                 externalId,
+                aggregationProvenance: "LEGACY_UNKNOWN",
               },
               select: { id: true },
             });

src/lib/measurements/import-apple-health-export.ts

@@ -4,18 +4,18 @@
  * Reads an `export.xml` byte-stream via `sax` (event-driven SAX, no
  * DOM) and folds every `<Record>`, `<Workout>`, and `<Correlation>`
  * element into the row shape the existing `Measurement` and
- * `Workout` models expect. Cumulative-quantity `<Record>` rows
- * (steps, energy, distance, flights, daylight) collapse into one
- * `stats:<HKType>:<YYYY-MM-DD>` `Measurement` per user-local day
- * — mirroring iOS's `HKStatisticsCollectionQuery` daily-aggregation
- * convention locked in v1.4.30. Spot rows (BP, weight, HRV, …)
- * survive verbatim, keyed by `HKSample.uuid` when present.
+ * `Workout` models expect. Cumulative-quantity `<Record>` rows are folded by
+ * `(type, local day, hashed source identity)`, summed within each source-day,
+ * then reduced to the largest source subtotal. The resulting
+ * `stats:<HKType>:<YYYY-MM-DD>` row is explicitly an export estimate; native
+ * HealthKit statistics remain authoritative.
  *
- * Memory profile: SAX callbacks fire as the byte cursor advances,
- * so peak RSS stays bounded regardless of the input file size. The
- * cumulative-bucket map holds at most one `(type, day)` entry per
- * observed day per cumulative type — a 10-year export with five
- * cumulative types lands at ~18 000 entries (~1 MB).
+ * Spot rows (BP, weight, HRV, …) survive verbatim, keyed by `HKSample.uuid`
+ * when present. SAX callbacks fire as the byte cursor advances, so peak RSS
+ * stays bounded regardless of input size. The cumulative map grows with the
+ * observed `(type, day, source hash)` combinations; every record without source
+ * metadata shares one bounded bucket, and raw source/device labels are never
+ * retained.
  *
  * Locks per `.planning/research/v1434-r-1-xml-import.md` §6.
  */
@@ -39,6 +39,7 @@ import {
   dayKeyForUserTz,
   canonicalDailyTimestamp,
 } from "@/lib/measurements/drain-per-sample-cumulative";
+import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement";
 import { resolveHkWorkoutSportType } from "@/lib/measurements/hk-workout-activity-type-map";
 import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
@@ -113,6 +114,12 @@ export interface ImportJobResult {
   cycle: CycleImportStats;
   deferred: Record<string, number>;
   unknown: Record<string, number>;
+  cumulativeEstimates: {
+    /** Distinct local calendar days containing at least one estimated total. */
+    days: number;
+    /** Estimated `(measurement type, local day)` aggregate rows considered. */
+    rows: number;
+  };
   totals: {
     recordsRead: number;
     rowsUpserted: number;
@@ -147,6 +154,8 @@ interface PreparedWorkout {
   metadata: Prisma.JsonValue | null;
 }
 
+type CumulativeSourceSubtotals = Map<string, number>;
+
 /** Per-type running stats accumulator. */
 interface MutableTypeStat {
   read: number;
@@ -182,6 +191,34 @@ export function hashSampleKey(
   );
 }
 
+const UNATTRIBUTED_CUMULATIVE_SOURCE_HASH = createHash("sha256")
+  .update(JSON.stringify(["unattributed"]))
+  .digest("hex");
+
+/**
+ * Hash the source tuple used only to keep overlapping export.xml contributors
+ * separate. Records without source metadata share one stable bucket so parser
+ * state remains bounded by actual source cardinality rather than record count.
+ * Raw source/device labels never leave the parser.
+ */
+export function hashCumulativeSourceIdentity(
+  sourceName: string | undefined,
+  sourceVersion: string | undefined,
+  device: string | undefined,
+): string {
+  const sourceTuple = [
+    sourceName?.trim() ?? "",
+    sourceVersion?.trim() ?? "",
+    device?.trim() ?? "",
+  ];
+  if (!sourceTuple.some(Boolean)) {
+    return UNATTRIBUTED_CUMULATIVE_SOURCE_HASH;
+  }
+  return createHash("sha256")
+    .update(JSON.stringify(["source", ...sourceTuple]))
+    .digest("hex");
+}
+
 /**
  * Parse the Apple `Record` `value` attribute. Quantity records carry
  * a numeric string; sleep-analysis records carry the symbolic name.
@@ -235,7 +272,7 @@ export interface StreamParseInput {
   /** IANA timezone, used to anchor cumulative-type day-keys. */
   userTimezone: string;
   /** Prisma client to flush rows through. */
-  prisma: Pick<PrismaClient, "measurement" | "workout">;
+  prisma: Pick<PrismaClient, "measurement" | "workout" | "$transaction">;
   /**
    * Live progress hook. Called every `PROGRESS_TICK_RECORDS` records
    * read, and once on terminal `done`. Best-effort; the parser
@@ -305,8 +342,11 @@ export async function streamParseExportXml(
     protectionUsed?: boolean;
   } | null = null;
 
-  // Cumulative-type fold: type -> dayKey -> running sum.
-  const cumulativeBucket = new Map<MeasurementType, Map<string, number>>();
+  // Cumulative fold: type -> local day -> hashed source identity -> subtotal.
+  const cumulativeBucket = new Map<
+    MeasurementType,
+    Map<string, CumulativeSourceSubtotals>
+  >();
   // Spot-row batch awaiting flush.
   const spotBatch: PreparedMeasurement[] = [];
   // Workout-row batch awaiting flush.
@@ -314,6 +354,8 @@ export async function streamParseExportXml(
 
   let recordsRead = 0;
   let rowsUpserted = 0;
+  const cumulativeEstimatedDays = new Set<string>();
+  let cumulativeEstimatedRows = 0;
   // Per R-1 §8 the percent stays best-effort and may remain null
   // until the parser sees the closing `</HealthData>` tag. We don't
   // currently mutate this in v1.4.34 — the iOS app keeps polling on
@@ -617,56 +659,74 @@ export async function streamParseExportXml(
 
   const flushCumulativeBuckets = async (): Promise<void> => {
     for (const [type, byDay] of cumulativeBucket.entries()) {
-      // Re-resolve the HK identifier from the mapping table — the
-      // bucket keys are MeasurementType, but the externalId carries
-      // the HK identifier so a future re-import collides on the
-      // same string.
+      // Re-resolve the HK identifier from the mapping table — the bucket keys
+      // are MeasurementType, but the shared externalId carries the HK
+      // identifier used by native HealthKit statistics.
       const mapping = Object.values(APPLE_HEALTH_TYPE_MAP).find(
-        (m) => m.measurementType === type,
+        (candidate) => candidate.measurementType === type,
       );
       if (!mapping) continue;
       const stat = bumpStat(type);
 
-      for (const [dayKey, sum] of byDay.entries()) {
+      for (const [dayKey, bySource] of byDay.entries()) {
+        let selected: [sourceHash: string, subtotal: number] | undefined;
+        for (const entry of bySource.entries()) {
+          if (
+            !selected ||
+            entry[1] > selected[1] ||
+            (entry[1] === selected[1] && entry[0] < selected[0])
+          ) {
+            selected = entry;
+          }
+        }
+        if (!selected) continue;
+        const [selectedSourceHash, selectedSubtotal] = selected;
+        if (validateMeasurementRange(type, selectedSubtotal) !== null) {
+          unknown[`${type}::aggregate_out_of_range`] =
+            (unknown[`${type}::aggregate_out_of_range`] ?? 0) + 1;
+          continue;
+        }
         const externalId = dailyStatsExternalId(mapping.hkIdentifier, dayKey);
         const measuredAt = canonicalDailyTimestamp(dayKey, userTimezone);
         const rowStart = Date.now();
-        const created = await prisma.measurement.createManyAndReturn({
-          data: [
+        const verdict = await prisma.$transaction((tx) =>
+          reconcileExternalMeasurement(
+            tx,
             {
               userId,
               type,
-              value: sum,
+              value: selectedSubtotal,
               unit: mapping.dbUnit,
               source: "APPLE_HEALTH",
               measuredAt,
               externalId,
+              externalSourceVersion: null,
+              sleepStage: null,
+              deviceType: null,
+              aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+              aggregationContributorCount: bySource.size,
+              aggregationSelectedSourceHash: selectedSourceHash,
             },
-          ],
-          skipDuplicates: true,
-          select: { id: true },
-        });
-        if (created.length > 0) {
+            { exactExternalMatch: "update" },
+          ),
+        );
+
+        cumulativeEstimatedDays.add(dayKey);
+        cumulativeEstimatedRows += 1;
+        if (verdict.status === "inserted") {
           stat.inserted += 1;
-        } else {
-          await prisma.measurement.update({
-            where: {
-              userId_type_source_externalId: {
-                userId,
-                type,
-                source: "APPLE_HEALTH",
-                externalId,
-              },
-            },
-            data: {
-              value: sum,
-              measuredAt,
-            },
-          });
+          rowsUpserted += 1;
+        } else if (
+          verdict.status === "updated" ||
+          verdict.status === "resurrected"
+        ) {
           stat.updated += 1;
+          rowsUpserted += 1;
+        } else if (verdict.status === "failed") {
+          unknown[`${type}::upsert_failed`] =
+            (unknown[`${type}::upsert_failed`] ?? 0) + 1;
         }
         stat.durationMs += Date.now() - rowStart;
-        rowsUpserted += 1;
       }
     }
   };
@@ -758,12 +818,25 @@ export async function streamParseExportXml(
 
       if (CUMULATIVE_HK_TYPES.has(mapped.type)) {
         const dayKey = dayKeyForUserTz(mapped.takenAt, userTimezone);
+        const sourceHash = hashCumulativeSourceIdentity(
+          attrs.sourceName,
+          attrs.sourceVersion,
+          attrs.device,
+        );
         let byDay = cumulativeBucket.get(mapped.type);
         if (!byDay) {
           byDay = new Map();
           cumulativeBucket.set(mapped.type, byDay);
         }
-        byDay.set(dayKey, (byDay.get(dayKey) ?? 0) + mapped.value);
+        let bySource = byDay.get(dayKey);
+        if (!bySource) {
+          bySource = new Map();
+          byDay.set(dayKey, bySource);
+        }
+        bySource.set(
+          sourceHash,
+          (bySource.get(sourceHash) ?? 0) + mapped.value,
+        );
       } else {
         // Spot row: derive a stable externalId, queue for flush.
         const externalId = hashSampleKey(
@@ -1008,6 +1081,10 @@ export async function streamParseExportXml(
     cycle,
     deferred,
     unknown,
+    cumulativeEstimates: {
+      days: cumulativeEstimatedDays.size,
+      rows: cumulativeEstimatedRows,
+    },
     totals: {
       recordsRead,
       rowsUpserted,

src/lib/measurements/reconcile-external-measurement.ts

@@ -1,6 +1,7 @@
 import {
   Prisma,
   type Measurement,
+  type MeasurementAggregationProvenance,
   type Prisma as PrismaTypes,
 } from "@/generated/prisma/client";
 
@@ -15,6 +16,7 @@ const rowSelect = {
   sleepStage: true,
   externalId: true,
   deletedAt: true,
+  aggregationProvenance: true,
 } satisfies PrismaTypes.MeasurementSelect;
 
 export type ExternalMeasurementWrite = Omit<
@@ -74,6 +76,22 @@ export interface MeasurementReconciliationOptions {
   exactExternalMatch?: "update" | "duplicate";
 }
 
+const AGGREGATION_AUTHORITY: Readonly<
+  Record<MeasurementAggregationProvenance, number>
+> = {
+  LEGACY_UNKNOWN: 1,
+  EXPORT_XML_SOURCE_MAX: 2,
+  HEALTHKIT_STATISTICS: 3,
+};
+
+function aggregationAuthority(
+  value: MeasurementAggregationProvenance | null | undefined,
+): number {
+  return value === null || value === undefined
+    ? 0
+    : AGGREGATION_AUTHORITY[value];
+}
+
 function errorVerdict(err: unknown): MeasurementReconciliationVerdict {
   const message = err instanceof Error ? err.message : String(err);
   if (err instanceof Prisma.PrismaClientKnownRequestError) {
@@ -162,6 +180,7 @@ async function lockAndReloadCandidates(
 async function retireCollision(
   tx: PrismaTypes.TransactionClient,
   row: PrismaTypes.MeasurementGetPayload<{ select: typeof rowSelect }>,
+  releaseExternalId = false,
 ): Promise<void> {
   let measuredAt = new Date(0);
   while (
@@ -183,6 +202,9 @@ async function retireCollision(
     where: { id: row.id },
     data: {
       measuredAt,
+      ...(releaseExternalId
+        ? { externalId: `retired:${row.id}:${row.externalId}` }
+        : {}),
       deletedAt: new Date(),
       syncVersion: { increment: 1 },
     },
@@ -240,12 +262,38 @@ export async function reconcileExternalMeasurement(
       return { status: "inserted", row: created };
     }
 
-    const canonical = externalHit ?? naturalHit!;
+    let canonical = externalHit ?? naturalHit!;
+    if (
+      desired.aggregationProvenance != null &&
+      externalHit &&
+      naturalHit &&
+      aggregationAuthority(naturalHit.aggregationProvenance) >
+        aggregationAuthority(externalHit.aggregationProvenance)
+    ) {
+      canonical = naturalHit;
+    }
     const redundant =
       externalHit && naturalHit && externalHit.id !== naturalHit.id
-        ? naturalHit
+        ? canonical.id === externalHit.id
+          ? naturalHit
+          : externalHit
         : undefined;
     const exactExternalMatch = externalHit?.id === naturalHit?.id;
+    if (
+      desired.aggregationProvenance != null &&
+      aggregationAuthority(desired.aggregationProvenance) <
+        aggregationAuthority(canonical.aggregationProvenance)
+    ) {
+      if (redundant) {
+        await retireCollision(tx, redundant, true);
+      }
+      await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`);
+      return {
+        status: "duplicate",
+        row: canonical,
+        ...(redundant ? { retiredCollisionId: redundant.id } : {}),
+      };
+    }
 
     if (
       options.exactExternalMatch === "duplicate" &&

src/lib/medications/scheduling/__tests__/next-due.test.ts

@@ -344,6 +344,49 @@ describe("computeDisplayDue — open overdue slot", () => {
   });
 });
 
+describe("computeDisplayDue — canonical availability start", () => {
+  const weekly = makeDailySchedule({
+    id: "sched-weekly",
+    windowStart: "08:00",
+    windowEnd: "08:00",
+    timesOfDay: ["08:00"],
+    daysOfWeek: "3",
+    rrule: "FREQ=WEEKLY;BYDAY=WE",
+  });
+
+  it("carries the weekly default band's early start for a future slot", () => {
+    const due = computeDisplayDue({
+      medication: makeMedication(),
+      schedules: [weekly],
+      now: d("2026-06-09T05:30:00Z"),
+      userTz: BERLIN,
+      lastIntakeAt: null,
+    });
+
+    expect(due?.at.toISOString()).toBe("2026-06-10T06:00:00.000Z");
+    expect(due?.overdue).toBe(false);
+    expect(due?.availableFrom?.toISOString()).toBe("2026-06-09T05:00:00.000Z");
+  });
+
+  it("uses an explicit weekly dose-window boundary instead of the default day-scale lead", () => {
+    const due = computeDisplayDue({
+      medication: makeMedication(),
+      schedules: [
+        {
+          ...weekly,
+          doseWindows: [{ timeOfDay: "08:00", start: "06:00", end: "10:00" }],
+        },
+      ],
+      now: d("2026-06-09T05:30:00Z"),
+      userTz: BERLIN,
+      lastIntakeAt: null,
+    });
+
+    expect(due?.at.toISOString()).toBe("2026-06-10T06:00:00.000Z");
+    expect(due?.availableFrom?.toISOString()).toBe("2026-06-10T03:00:00.000Z");
+  });
+});
+
 /**
  * v1.16.9 — an AD-HOC row (`scheduledFor === takenAt`) must never
  * ±6h-resolve a DIFFERENT slot. The proven failure: twice-daily 08:00 /

src/lib/medications/scheduling/next-due.ts

@@ -146,9 +146,14 @@ export function computeNextDueAt(input: {
   return earliest;
 }
 
-/** The instant a medication card should surface, plus its overdue state. */
+/** The instant a medication card should surface, plus its canonical band state. */
 export interface DisplayDue {
   at: Date;
+  /**
+   * Earliest instant at which the canonical attribution band accepts this
+   * slot. This is cadence- and per-dose-window-aware.
+   */
+  availableFrom?: Date;
   /**
    * True when `at` is an OPEN overdue slot: its anchor has passed but `now`
    * is still inside the slot's catch-up band (`anchor < now ≤ overdueEnd`)
@@ -190,12 +195,19 @@ export function computeDisplayDue(
   input: ComputeDisplayDueInput,
 ): DisplayDue | null {
   const open = findOpenOverdueSlot(input);
-  if (open) return { at: open, overdue: true };
+  if (open) return { ...open, overdue: true };
   const next = computeNextDueAt(input);
-  return next ? { at: next, overdue: false } : null;
+  if (!next) return null;
+  return {
+    at: next,
+    availableFrom: findAvailabilityStart(input, next) ?? next,
+    overdue: false,
+  };
 }
 
-function findOpenOverdueSlot(input: ComputeDisplayDueInput): Date | null {
+function findOpenOverdueSlot(
+  input: ComputeDisplayDueInput,
+): { at: Date; availableFrom: Date } | null {
   const { medication, schedules, now, userTz, lastIntakeAt } = input;
   if (schedules.length === 0) return null;
 
@@ -208,7 +220,7 @@ function findOpenOverdueSlot(input: ComputeDisplayDueInput): Date | null {
   if (floor.getTime() >= now.getTime()) return null;
 
   const ctx = buildRecurrenceContext({ medication, userTz, lastIntakeAt });
-  let latest: Date | null = null;
+  let latest: { at: Date; availableFrom: Date } | null = null;
   for (const schedule of schedules) {
     const { bands } = buildBandsForMedication({
       medication,
@@ -229,8 +241,57 @@ function findOpenOverdueSlot(input: ComputeDisplayDueInput): Date | null {
       if (anchor >= now.getTime()) continue;
       if (now.getTime() > band.overdueEnd.getTime()) continue;
       if (isResolved(band.at)) continue;
-      if (latest === null || anchor > latest.getTime()) latest = band.at;
+      if (
+        latest === null ||
+        anchor > latest.at.getTime() ||
+        (anchor === latest.at.getTime() &&
+          band.earlyStart.getTime() < latest.availableFrom.getTime())
+      ) {
+        latest = { at: band.at, availableFrom: band.earlyStart };
+      }
     }
   }
   return latest;
 }
+
+function findAvailabilityStart(
+  input: ComputeDisplayDueInput,
+  at: Date,
+): Date | null {
+  const ctx = buildRecurrenceContext({
+    medication: input.medication,
+    userTz: input.userTz,
+    lastIntakeAt: input.lastIntakeAt,
+  });
+  const from = new Date(at.getTime() - OVERDUE_LOOKBACK_MS);
+  let earliest: Date | null = null;
+
+  for (const scheduleRow of input.schedules) {
+    const schedule = buildCanonicalSchedule(scheduleRow);
+    const intakeInstants =
+      schedule.rollingIntervalDays === null
+        ? input.lastIntakeAt
+          ? [input.lastIntakeAt]
+          : []
+        : [input.lastIntakeAt, at].filter(
+            (instant): instant is Date => instant !== null,
+          );
+    const { bands } = buildBandsForMedication({
+      medication: input.medication,
+      schedule,
+      ctx,
+      userTz: input.userTz,
+      range: { from, to: at },
+      now: input.now,
+      intakeInstants,
+    });
+    for (const band of bands) {
+      if (band.at.getTime() !== at.getTime()) continue;
+      if (earliest === null || band.earlyStart.getTime() < earliest.getTime()) {
+        earliest = band.earlyStart;
+      }
+    }
+  }
+
+  return earliest;
+}

src/lib/openapi/routes/insights/schemas.ts

@@ -897,9 +897,20 @@ export const dashboardSnapshotResponse = z
         nextDueOverdue: z.boolean(),
         nextDueMedicationName: z.string().nullable(),
         nextDueMedicationId: z.string().nullable(),
+        dueCandidates: z
+          .array(
+            z.object({
+              medicationId: z.string(),
+              medicationName: z.string(),
+              dueAt: z.string(),
+              overdue: z.boolean(),
+              availableFrom: z.string(),
+            }),
+          )
+          .optional(),
       })
       .describe(
-        "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), and the earliest next-due slot across active medications. `nextDueOverdue: true` marks an OPEN overdue slot (anchor passed, still inside its catch-up band). `nextDueMedicationId` deep-links a consumer straight to the overdue medication's card. Staleness contract: the snapshot is cache-served, so a `nextDueAt` in the past with `nextDueOverdue: false` means the anchor passed after the snapshot was built — render the plain day summary, never an overdue state.",
+        "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), every active medication's current display-due candidate, and legacy scalar fields projected from candidate zero. `overdue: true` marks an OPEN overdue slot (anchor passed, still inside its catch-up band); `availableFrom` is the canonical cadence- and dose-window-derived attribution-band start. `dueCandidates` is optional so older cached snapshots remain valid. Medication ids let consumers deep-link straight to the relevant card.",
       ),
     // Dashboard hero — health score (warm phase, nullable on a
     // rollup-coverage miss). Score + band + delta only; the per-pillar

tests/integration/apple-health-import-reconcile.test.ts

@@ -22,6 +22,7 @@ import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { getPrismaClient, truncateAllTables } from "./setup";
 import {
   reconcileOrphanImportJobs,
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
   _setWorkerPrismaForTests,
 } from "@/lib/jobs/apple-health-import-worker";
 
@@ -75,6 +76,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
         pgBossJobId: "boss-active-fresh",
         status: "parsing",
         uploadBytes: 100,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       },
     });
     bossMock.getJobById.mockResolvedValue({
@@ -98,6 +100,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
         pgBossJobId: "boss-gone",
         status: "upserting",
         uploadBytes: 100,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       },
     });
     bossMock.getJobById.mockResolvedValue(null);
@@ -118,6 +121,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
         pgBossJobId: "boss-terminal",
         status: "parsing",
         uploadBytes: 100,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       },
     });
     bossMock.getJobById.mockResolvedValue({
@@ -140,6 +144,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
         pgBossJobId: "boss-stale-active",
         status: "parsing",
         uploadBytes: 100,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       },
     });
     await backdateHeartbeat(row.id, 45);
@@ -163,6 +168,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
         pgBossJobId: "boss-none",
         status: "parsing",
         uploadBytes: 100,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
       },
     });
     bossMock.handle = null; // getGlobalBoss() → null
@@ -173,6 +179,26 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => {
     expect(after!.status).toBe("failed");
   });
 
+  it("leaves a legacy-default revision-1 row to a revision-1 worker", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("recon-legacy-revision");
+    const row = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        pgBossJobId: "boss-legacy",
+        status: "parsing",
+        uploadBytes: 100,
+      },
+    });
+
+    expect(row.parserRevision).toBe(1);
+    await reconcileOrphanImportJobs();
+
+    const after = await prisma.importJob.findUnique({ where: { id: row.id } });
+    expect(after!.status).toBe("parsing");
+    expect(bossMock.getJobById).not.toHaveBeenCalled();
+  });
+
   it("leaves terminal rows untouched", async () => {
     const prisma = getPrismaClient();
     const user = await createUser("recon-already-terminal");
@@ -197,7 +223,12 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
   // Mirrors the exact `findFirst` the kick-off routes run.
   async function dedupLookup(userId: string, sha: string) {
     return getPrismaClient().importJob.findFirst({
-      where: { userId, uploadSha256: sha, status: { not: "failed" } },
+      where: {
+        userId,
+        uploadSha256: sha,
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+        status: { not: "failed" },
+      },
       orderBy: { startedAt: "desc" },
     });
   }
@@ -212,6 +243,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
         failureReason: "ENOENT: no such file or directory",
         uploadBytes: 100,
         uploadSha256: "sha-failed",
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
         completedAt: new Date(),
       },
     });
@@ -228,6 +260,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
         status: "done",
         uploadBytes: 100,
         uploadSha256: "sha-viable",
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
         completedAt: new Date(),
       },
     });
@@ -245,6 +278,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
         status: "failed",
         uploadBytes: 100,
         uploadSha256: "sha-mixed",
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
         completedAt: new Date(),
       },
     });
@@ -254,6 +288,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
         status: "done",
         uploadBytes: 100,
         uploadSha256: "sha-mixed",
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
         completedAt: new Date(),
       },
     });
@@ -261,4 +296,34 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => {
     const hit = await dedupLookup(user.id, "sha-mixed");
     expect(hit!.id).toBe(done.id);
   });
+  it("allows a revision-1 archive to be reprocessed at revision 2", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("dedup-parser-revision");
+    await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        status: "done",
+        uploadBytes: 100,
+        uploadSha256: "sha-parser-revision",
+        parserRevision: 1,
+        completedAt: new Date(),
+      },
+    });
+
+    expect(await dedupLookup(user.id, "sha-parser-revision")).toBeNull();
+
+    const current = await prisma.importJob.create({
+      data: {
+        userId: user.id,
+        status: "done",
+        uploadBytes: 100,
+        uploadSha256: "sha-parser-revision",
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+        completedAt: new Date(),
+      },
+    });
+    expect((await dedupLookup(user.id, "sha-parser-revision"))?.id).toBe(
+      current.id,
+    );
+  });
 });

tests/integration/drain-per-sample-cumulative.test.ts

@@ -92,6 +92,10 @@ describe("drainPerSampleCumulative (real Postgres)", () => {
     expect(remaining[0].measuredAt.toISOString()).toBe(
       "2026-05-16T10:00:00.000Z",
     );
+    expect(remaining[0]).toMatchObject({
+      aggregationProvenance: "LEGACY_UNKNOWN",
+      syncVersion: 1,
+    });
   });
 
   it("is idempotent — a second run after a successful drain is a no-op", async () => {
@@ -219,6 +223,93 @@ describe("drainPerSampleCumulative (real Postgres)", () => {
     expect(remaining).toHaveLength(2);
   });
 
+  it.each(["HEALTHKIT_STATISTICS", "EXPORT_XML_SOURCE_MAX"] as const)(
+    "drains source rows without mutating an authoritative %s total",
+    async (aggregationProvenance) => {
+      const prisma = getPrismaClient();
+      await prisma.measurement.createMany({
+        data: [
+          {
+            userId: TEST_USER_ID,
+            type: "ACTIVITY_STEPS",
+            value: 5_400,
+            unit: "steps",
+            source: "APPLE_HEALTH",
+            measuredAt: new Date("2026-05-16T10:00:00.000Z"),
+            externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
+            aggregationProvenance,
+          },
+          {
+            userId: TEST_USER_ID,
+            type: "ACTIVITY_STEPS",
+            value: 500,
+            unit: "steps",
+            source: "APPLE_HEALTH",
+            measuredAt: new Date("2026-05-16T20:00:00.000Z"),
+            externalId: `late-${aggregationProvenance}`,
+          },
+        ],
+      });
+
+      await drainPerSampleCumulative(prisma, {
+        userId: TEST_USER_ID,
+        log: () => {},
+      });
+
+      const remaining = await prisma.measurement.findMany({
+        where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" },
+      });
+      expect(remaining).toHaveLength(1);
+      expect(remaining[0]).toMatchObject({
+        value: 5_400,
+        aggregationProvenance,
+        syncVersion: 1,
+      });
+    },
+  );
+
+  it("keeps legacy late-increment semantics and bumps syncVersion", async () => {
+    const prisma = getPrismaClient();
+    await prisma.measurement.createMany({
+      data: [
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 5_400,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T10:00:00.000Z"),
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
+          aggregationProvenance: "LEGACY_UNKNOWN",
+        },
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 500,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T20:00:00.000Z"),
+          externalId: "late-legacy",
+        },
+      ],
+    });
+
+    await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      log: () => {},
+    });
+
+    const remaining = await prisma.measurement.findMany({
+      where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" },
+    });
+    expect(remaining).toHaveLength(1);
+    expect(remaining[0]).toMatchObject({
+      value: 5_900,
+      aggregationProvenance: "LEGACY_UNKNOWN",
+      syncVersion: 2,
+    });
+  });
+
   it("skips rows already in stats:... shape (re-running after an earlier drain)", async () => {
     const prisma = getPrismaClient();
     await prisma.measurement.create({

tests/integration/measurement-identity-reconcile.test.ts

@@ -97,6 +97,130 @@ describe("external measurement identity reconciliation (real Postgres)", () => {
     });
   });
 
+  it("retires a lower-authority XML collision while preserving the native aggregate", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("identity-native-authority");
+    const nativeMeasuredAt = new Date("2026-07-20T10:00:00.000Z");
+    const xmlExternal = await prisma.measurement.create({
+      data: {
+        ...desired(
+          user.id,
+          "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+          {
+            type: "ACTIVITY_STEPS",
+            value: 8_526,
+            unit: "steps",
+            source: "APPLE_HEALTH",
+            measuredAt: new Date("2026-07-20T09:00:00.000Z"),
+            sleepStage: null,
+            aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+            aggregationContributorCount: 2,
+            aggregationSelectedSourceHash: "xml-source-hash",
+          },
+        ),
+      },
+    });
+    const native = await prisma.measurement.create({
+      data: {
+        ...desired(user.id, "stats:legacy-native-id", {
+          type: "ACTIVITY_STEPS",
+          value: 8_600,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: nativeMeasuredAt,
+          sleepStage: null,
+          externalSourceVersion: "native-v1",
+          deviceType: "Apple Watch",
+          aggregationProvenance: "HEALTHKIT_STATISTICS",
+          aggregationContributorCount: null,
+          aggregationSelectedSourceHash: null,
+        }),
+      },
+    });
+
+    const verdict = await reconcile(
+      desired(user.id, "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", {
+        type: "ACTIVITY_STEPS",
+        value: 8_526,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: nativeMeasuredAt,
+        sleepStage: null,
+        aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+        aggregationContributorCount: 2,
+        aggregationSelectedSourceHash: "xml-source-hash",
+      }),
+      { exactExternalMatch: "update" },
+    );
+
+    expect(verdict).toMatchObject({
+      status: "duplicate",
+      row: { id: native.id, externalId: "stats:legacy-native-id" },
+      retiredCollisionId: xmlExternal.id,
+    });
+    const liveRows = await prisma.measurement.findMany({
+      where: { userId: user.id, deletedAt: null },
+    });
+    expect(liveRows).toHaveLength(1);
+    expect(liveRows[0]).toMatchObject({
+      id: native.id,
+      value: 8_600,
+      unit: "steps",
+      measuredAt: nativeMeasuredAt,
+      externalId: "stats:legacy-native-id",
+      externalSourceVersion: "native-v1",
+      deviceType: "Apple Watch",
+      aggregationProvenance: "HEALTHKIT_STATISTICS",
+      aggregationContributorCount: null,
+      aggregationSelectedSourceHash: null,
+    });
+    expect(
+      await prisma.measurement.findUniqueOrThrow({
+        where: { id: xmlExternal.id },
+      }),
+    ).toMatchObject({
+      externalId: `retired:${xmlExternal.id}:stats:HKQuantityTypeIdentifierStepCount:2026-07-20`,
+      deletedAt: expect.any(Date),
+    });
+
+    const nativeUpdate = await reconcile(
+      desired(user.id, "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", {
+        type: "ACTIVITY_STEPS",
+        value: 8_700,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: nativeMeasuredAt,
+        sleepStage: null,
+        externalSourceVersion: "native-v2",
+        deviceType: "Apple Watch",
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+        aggregationContributorCount: null,
+        aggregationSelectedSourceHash: null,
+      }),
+      { exactExternalMatch: "update" },
+    );
+
+    expect(nativeUpdate).toMatchObject({
+      status: "updated",
+      row: {
+        id: native.id,
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+      },
+    });
+    expect(
+      await prisma.measurement.findMany({
+        where: { userId: user.id, deletedAt: null },
+      }),
+    ).toEqual([
+      expect.objectContaining({
+        id: native.id,
+        value: 8_700,
+        externalSourceVersion: "native-v2",
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+      }),
+    ]);
+  });
+
   it("adopts a live natural-key row when a new external id collides", async () => {
     const prisma = getPrismaClient();
     const user = await createUser("identity-natural-live");

tests/integration/measurements-batch.test.ts

@@ -9,10 +9,14 @@
  *   - Idempotency-Key replay returns the cached envelope
  */
 import { NextRequest } from "next/server";
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 import { cookieJar } from "./mock-next-headers";
 import { getPrismaClient, truncateAllTables } from "./setup";
+import { streamParseExportXml } from "@/lib/measurements/import-apple-health-export";
 
 const TEST_USER_ID = "user-batch-ingest-test";
 
@@ -89,6 +93,41 @@ function makeRequest(
   });
 }
 
+function writeCumulativeXml(firstValue = 8_526, secondValue = 7_082): string {
+  const tmp = mkdtempSync(join(tmpdir(), "healthlog-aggregate-authority-"));
+  const xmlPath = join(tmp, "export.xml");
+  writeFileSync(
+    xmlPath,
+    `<?xml version="1.0" encoding="UTF-8"?>
+<HealthData locale="en_US">
+  <Record type="HKQuantityTypeIdentifierStepCount" unit="count"
+          startDate="2026-07-20 08:00:00 +0200"
+          endDate="2026-07-20 08:30:00 +0200"
+          value="${firstValue}" sourceName="iPhone" sourceVersion="18.5"/>
+  <Record type="HKQuantityTypeIdentifierStepCount" unit="count"
+          startDate="2026-07-20 09:00:00 +0200"
+          endDate="2026-07-20 09:30:00 +0200"
+          value="${secondValue}" sourceName="Zepp" sourceVersion="9.1"/>
+</HealthData>`,
+  );
+  return xmlPath;
+}
+
+function nativeStepsRequest(value: number): NextRequest {
+  return makeRequest({
+    entries: [
+      {
+        hkIdentifier: "HKQuantityTypeIdentifierStepCount",
+        value,
+        unit: "count",
+        startDate: "2026-07-20T00:00:00.000Z",
+        endDate: "2026-07-20T18:00:00.000Z",
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+      },
+    ],
+  });
+}
+
 describe("POST /api/measurements/batch (real Postgres)", () => {
   it("inserts a well-formed batch and returns per-entry status", async () => {
     const { POST } = await import("@/app/api/measurements/batch/route");
@@ -560,6 +599,12 @@ describe("POST /api/measurements/batch (real Postgres)", () => {
       });
       expect(stored).toHaveLength(1);
       expect(stored[0]!.value).toBeCloseTo(5200, 5);
+      expect(stored[0]).toMatchObject({
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+        aggregationContributorCount: null,
+        aggregationSelectedSourceHash: null,
+        syncVersion: 2,
+      });
     });
 
     it("sample-class externalIds (non-stats:* prefix) keep the strict duplicate contract", async () => {
@@ -1153,4 +1198,140 @@ describe("POST /api/measurements/batch (real Postgres)", () => {
       expect(stored).toHaveLength(0);
     });
   });
+  describe("Apple aggregate authority reconciliation", () => {
+    it("promotes an XML estimate to native HealthKit statistics", async () => {
+      const prisma = getPrismaClient();
+      await streamParseExportXml({
+        xmlPath: writeCumulativeXml(),
+        userId: TEST_USER_ID,
+        userTimezone: "Europe/Berlin",
+        prisma,
+      });
+      const estimated = await prisma.measurement.findMany({
+        where: {
+          userId: TEST_USER_ID,
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        },
+      });
+      expect(estimated).toHaveLength(1);
+      expect(estimated[0]).toMatchObject({
+        value: 8_526,
+        aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+        aggregationContributorCount: 2,
+        syncVersion: 1,
+      });
+
+      const { POST } = await import("@/app/api/measurements/batch/route");
+      const response = await POST(nativeStepsRequest(8_600));
+      expect(response.status).toBe(200);
+
+      const canonical = await prisma.measurement.findMany({
+        where: {
+          userId: TEST_USER_ID,
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        },
+      });
+      expect(canonical).toHaveLength(1);
+      expect(canonical[0]).toMatchObject({
+        value: 8_600,
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+        aggregationContributorCount: null,
+        aggregationSelectedSourceHash: null,
+        syncVersion: 2,
+      });
+    });
+
+    it("keeps native statistics authoritative when XML arrives later", async () => {
+      const { POST } = await import("@/app/api/measurements/batch/route");
+      expect((await POST(nativeStepsRequest(8_600))).status).toBe(200);
+
+      const result = await streamParseExportXml({
+        xmlPath: writeCumulativeXml(),
+        userId: TEST_USER_ID,
+        userTimezone: "Europe/Berlin",
+        prisma: getPrismaClient(),
+      });
+      expect(result.cumulativeEstimates).toEqual({ days: 1, rows: 1 });
+      expect(result.perType.ACTIVITY_STEPS).toMatchObject({
+        inserted: 0,
+        updated: 0,
+      });
+
+      const canonical = await getPrismaClient().measurement.findMany({
+        where: {
+          userId: TEST_USER_ID,
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        },
+      });
+      expect(canonical).toHaveLength(1);
+      expect(canonical[0]).toMatchObject({
+        value: 8_600,
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+        syncVersion: 1,
+      });
+    });
+
+    it("re-imports an XML estimate onto one row with monotonic syncVersion", async () => {
+      const prisma = getPrismaClient();
+      const xmlPath = writeCumulativeXml();
+      await streamParseExportXml({
+        xmlPath,
+        userId: TEST_USER_ID,
+        userTimezone: "Europe/Berlin",
+        prisma,
+      });
+      await streamParseExportXml({
+        xmlPath,
+        userId: TEST_USER_ID,
+        userTimezone: "Europe/Berlin",
+        prisma,
+      });
+
+      const rows = await prisma.measurement.findMany({
+        where: {
+          userId: TEST_USER_ID,
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        },
+      });
+      expect(rows).toHaveLength(1);
+      expect(rows[0]).toMatchObject({
+        value: 8_526,
+        aggregationProvenance: "EXPORT_XML_SOURCE_MAX",
+        aggregationContributorCount: 2,
+        syncVersion: 2,
+      });
+    });
+
+    it("repairs an explicitly legacy aggregate with native statistics", async () => {
+      const prisma = getPrismaClient();
+      await prisma.measurement.create({
+        data: {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 15_608,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-07-20T10:00:00.000Z"),
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+          aggregationProvenance: "LEGACY_UNKNOWN",
+        },
+      });
+
+      const { POST } = await import("@/app/api/measurements/batch/route");
+      expect((await POST(nativeStepsRequest(8_600))).status).toBe(200);
+
+      const repaired = await prisma.measurement.findMany({
+        where: {
+          userId: TEST_USER_ID,
+          externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20",
+        },
+      });
+      expect(repaired).toHaveLength(1);
+      expect(repaired[0]).toMatchObject({
+        value: 8_600,
+        aggregationProvenance: "HEALTHKIT_STATISTICS",
+        syncVersion: 2,
+      });
+    });
+  });
 });