observability(intraday-pulse): surface per-shape row counts and HR-bucket write outcomes

Related to #585. Investigated the intraday pulse chart's read path
end to end: the ZIP-import raw-sample fold, the live-sync uploaded
10-min bucket fold, their merge, and the chart's render logic. All of
it treats the two ingestion shapes consistently — the chart has no
shape awareness at all, it paints whatever bucket series it receives.
No read-side or display-side defect explains a systematic density
difference between the two ingestion paths.

What was missing is visibility into the write and read sides that
would let a future report actually be diagnosed instead of
re-investigated from scratch. The batch endpoint's wide-event trace
never broke a skip down by reason, and never distinguished an
HR-bucket write from any other entry in the same request. The
intraday route's own trace surfaced only the final bucket count, with
no visibility into which shape (raw sample rows vs uploaded bucket
rows vs folded hourly rows) a day's data actually came from. Both
traces now carry that breakdown, so the next report of "the chart
looks sparser after a given date" is answerable by grep instead of by
re-deriving the whole read path again.

src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts

@@ -0,0 +1,269 @@
+/**
+ * v1.32.1 (issue #585) — wide-event observability for the go-forward
+ * aggregated 10-min heart-rate bucket wire contract.
+ *
+ * The intraday pulse chart reads exactly the `stats:HKQuantityTypeIdentifier
+ * HeartRate:<10-min-Z>` rows this batch route accepts (or rejects). A report
+ * of "the chart looks sparser after the ZIP-import cutoff" was previously
+ * undiagnosable from server logs: the baseline `measurement.batch.ingest`
+ * annotation carried only a total `skipped` count with no reason breakdown
+ * and no visibility into how many of a batch's entries specifically targeted
+ * an HR bucket. These tests pin the new `measurement.batch.hr-bucket`
+ * annotation (fires only when the batch actually carried an HR-bucket entry)
+ * and the `skipped_by_reason` breakdown on the baseline annotation.
+ */
+import { describe, expect, it, vi, beforeEach } from "vitest";
+import { NextRequest } from "next/server";
+
+const mocks = vi.hoisted(() => ({ annotate: vi.fn() }));
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    user: {
+      update: vi.fn(),
+    },
+    measurement: {
+      findMany: vi.fn(),
+      createManyAndReturn: vi.fn(),
+      updateMany: vi.fn(),
+    },
+    $transaction: vi.fn(async (fn: unknown) => {
+      if (typeof fn === "function") {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        return (fn as any)(prisma as unknown as { measurement: unknown });
+      }
+    }),
+  },
+}));
+
+vi.mock("@/lib/logging/context", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/lib/logging/context")>();
+  return { ...actual, annotate: mocks.annotate };
+});
+
+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: async (
+    tx: {
+      measurement: {
+        findMany: (args: unknown) => Promise<
+          Array<{
+            id?: string;
+            type?: string;
+            source?: string;
+            externalId?: string | null;
+          }>
+        >;
+        createManyAndReturn: (args: {
+          data: Record<string, unknown>;
+        }) => Promise<Array<Record<string, unknown>>>;
+        updateMany: (args: {
+          where: Record<string, unknown>;
+          data: Record<string, unknown>;
+        }) => Promise<unknown>;
+      };
+    },
+    input: Record<string, unknown> & {
+      userId: string;
+      type: string;
+      source: string;
+      externalId: string;
+    },
+    options: { exactExternalMatch?: "update" | "duplicate" } = {},
+  ) => {
+    const existing = await tx.measurement.findMany({});
+    const exact = existing.find(
+      (row: { type?: string; source?: string; externalId?: string | null }) =>
+        row.type === input.type &&
+        row.source === input.source &&
+        row.externalId === input.externalId,
+    );
+    if (exact && options.exactExternalMatch !== "update") {
+      return { status: "duplicate", row: exact };
+    }
+    if (exact) {
+      await tx.measurement.updateMany({
+        where: { id: exact.id },
+        data: { ...input, deletedAt: null },
+      });
+      return { status: "updated", row: { ...exact, ...input } };
+    }
+    const [inserted] = await tx.measurement.createManyAndReturn({
+      data: input,
+    });
+    if (inserted) return { status: "inserted", row: inserted };
+    if (options.exactExternalMatch === "update") {
+      await tx.measurement.updateMany({ where: {}, data: input });
+      return { status: "updated", row: input };
+    }
+    return { status: "duplicate" };
+  },
+}));
+
+vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
+vi.mock("@/lib/auth/audit", () => ({
+  auditLog: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/rate-limit", () => ({ checkRateLimit: vi.fn() }));
+vi.mock("@/lib/jobs/pr-detection", () => ({
+  enqueuePrDetection: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/jobs/reminder-satisfy", () => ({
+  enqueueReminderSatisfy: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/cache/invalidate", () => ({
+  invalidateUserMeasurements: vi.fn(),
+}));
+vi.mock("@/lib/insights/comprehensive-generate", () => ({
+  invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/rollups/measurement-rollups", () => ({
+  recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+  collapseToTypeDayKeys: vi.fn(() => []),
+}));
+
+vi.mock("next/headers", () => ({
+  headers: vi.fn(async () => ({ get: () => null })),
+  cookies: vi.fn(async () => ({
+    get: () => undefined,
+    set: () => {},
+    delete: () => {},
+  })),
+}));
+
+import { POST } from "../route";
+import { prisma } from "@/lib/db";
+import { getSession } from "@/lib/auth/session";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+const SESSION_OK = {
+  session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
+  user: { id: "user-1", username: "testuser", role: "USER" as const },
+};
+
+const FRESH_BUCKET_ID =
+  "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:00:00.000Z";
+const REPOST_BUCKET_ID =
+  "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T15:00:00.000Z";
+const MALFORMED_BUCKET_ID =
+  "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:35:00.000Z";
+const SAMPLE_ID = "C0FFEE00-0000-4000-8000-000000000000";
+
+function makeRequest(body: unknown): NextRequest {
+  return new NextRequest("http://localhost/api/measurements/batch", {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body),
+  });
+}
+
+function hrEntry(externalId: string, value: number) {
+  return {
+    hkIdentifier: "HKQuantityTypeIdentifierHeartRate",
+    value,
+    unit: "count/min",
+    startDate: "2026-06-21T14:00:00.000Z",
+    endDate: "2026-06-21T14:59:59.000Z",
+    externalId,
+  };
+}
+
+function annotateCallsFor(actionName: string) {
+  return mocks.annotate.mock.calls.filter(
+    ([arg]) =>
+      (arg as { action?: { name?: string } })?.action?.name === actionName,
+  );
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+  vi.mocked(checkRateLimit).mockResolvedValue({
+    allowed: true,
+    remaining: 60,
+    resetAt: Date.now() + 60_000,
+  });
+  vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation((async (
+    args: unknown,
+  ) => {
+    const { data } = args as {
+      data:
+        | { type: unknown; source: string; externalId: string | null }
+        | Array<{ type: unknown; source: string; externalId: string | null }>;
+    };
+    const rows = Array.isArray(data) ? data : [data];
+    return rows.map(({ type, source, externalId }) => ({
+      type,
+      source,
+      externalId,
+    }));
+  }) as never);
+  vi.mocked(prisma.measurement.updateMany).mockResolvedValue({ count: 1 });
+});
+
+describe("POST /api/measurements/batch — HR-bucket observability (issue #585)", () => {
+  it("reports a per-outcome breakdown covering only HR-bucket entries", async () => {
+    // REPOST_BUCKET_ID already exists — the reconcile mock overwrites it.
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([
+      {
+        type: "PULSE",
+        source: "APPLE_HEALTH",
+        externalId: REPOST_BUCKET_ID,
+      },
+    ] as never);
+
+    const res = await POST(
+      makeRequest({
+        entries: [
+          hrEntry(FRESH_BUCKET_ID, 72), // inserted
+          hrEntry(REPOST_BUCKET_ID, 78), // updated (overwrite)
+          hrEntry(MALFORMED_BUCKET_ID, 70), // skipped (malformed)
+          hrEntry(SAMPLE_ID, 64), // NOT an HR bucket — must be excluded
+        ],
+      }),
+    );
+    expect(res.status).toBe(200);
+
+    const calls = annotateCallsFor("measurement.batch.hr-bucket");
+    expect(calls).toHaveLength(1);
+    const meta = (calls[0][0] as { meta: Record<string, number> }).meta;
+    expect(meta).toEqual({
+      attempted: 3,
+      inserted: 1,
+      updated: 1,
+      duplicate: 0,
+      skipped: 1,
+      failed: 0,
+      processed: 4,
+    });
+  });
+
+  it("never fires the HR-bucket annotation for a batch with no bucket entries", async () => {
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([]);
+    const res = await POST(makeRequest({ entries: [hrEntry(SAMPLE_ID, 64)] }));
+    expect(res.status).toBe(200);
+    expect(annotateCallsFor("measurement.batch.hr-bucket")).toHaveLength(0);
+  });
+
+  it("breaks the baseline ingest annotation's skip count down by reason", async () => {
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([]);
+    const res = await POST(
+      makeRequest({
+        entries: [
+          hrEntry(MALFORMED_BUCKET_ID, 70), // skipped: malformed_hr_bucket_id
+        ],
+      }),
+    );
+    expect(res.status).toBe(200);
+
+    const calls = annotateCallsFor("measurement.batch.ingest");
+    expect(calls).toHaveLength(1);
+    const meta = calls[0][0] as { meta: { skipped_by_reason: unknown } };
+    expect(meta.meta.skipped_by_reason).toEqual({
+      malformed_hr_bucket_id: 1,
+    });
+  });
+});

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

@@ -603,6 +603,18 @@ async function postBatch(request: NextRequest): Promise<Response> {
     .filter((r) => r.status === "skipped")
     .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" }));
 
+  // v1.32.1 (issue #585) — per-reason skip breakdown. The baseline
+  // annotation below only ever carried the total `skipped` count, so an
+  // operator investigating "why does this account's uploaded data look
+  // sparser than expected" had no way to see WHICH validation gate an
+  // entry tripped (malformed bucket id, out-of-range value, unmappable
+  // identifier, …) without pulling raw rows. Grouping by reason turns
+  // that into a one-glance wide-event field.
+  const skippedByReason: Record<string, number> = {};
+  for (const s of skipped) {
+    skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1;
+  }
+
   await auditLog("measurement.batch.ingest", {
     userId: user.id,
     ipAddress: getClientIp(request),
@@ -648,6 +660,64 @@ async function postBatch(request: NextRequest): Promise<Response> {
     });
   }
 
+  // v1.32.1 (issue #585) — dedicated outcome breakdown for entries
+  // targeting the go-forward aggregated 10-min heart-rate bucket prefix
+  // (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday
+  // pulse chart reads exactly these rows for a live-synced day; a report
+  // of "the chart looks sparser after the ZIP import cutoff" is
+  // unanswerable from the baseline ingest trace alone, since it never
+  // distinguished an HR-bucket write from any other entry in the batch.
+  // Grepping `measurement.batch.hr-bucket` now shows, per sync, how many
+  // bucket entries the client sent and what happened to each — confirming
+  // (or ruling out) server-side rejection/dedup as the density-gap cause.
+  // Only fires when the batch actually carried at least one HR-bucket
+  // entry so the baseline trace is unchanged for every other sync.
+  let hrBucketInserted = 0;
+  let hrBucketUpdated = 0;
+  let hrBucketDuplicate = 0;
+  let hrBucketSkipped = 0;
+  let hrBucketFailed = 0;
+  for (let index = 0; index < entries.length; index++) {
+    if (!targetsAggregatedBucket(entries[index].externalId)) continue;
+    switch (results[index]?.status) {
+      case "inserted":
+        hrBucketInserted++;
+        break;
+      case "updated":
+        hrBucketUpdated++;
+        break;
+      case "duplicate":
+        hrBucketDuplicate++;
+        break;
+      case "skipped":
+        hrBucketSkipped++;
+        break;
+      case "failed":
+        hrBucketFailed++;
+        break;
+    }
+  }
+  const hrBucketAttempted =
+    hrBucketInserted +
+    hrBucketUpdated +
+    hrBucketDuplicate +
+    hrBucketSkipped +
+    hrBucketFailed;
+  if (hrBucketAttempted > 0) {
+    annotate({
+      action: { name: "measurement.batch.hr-bucket" },
+      meta: {
+        attempted: hrBucketAttempted,
+        inserted: hrBucketInserted,
+        updated: hrBucketUpdated,
+        duplicate: hrBucketDuplicate,
+        skipped: hrBucketSkipped,
+        failed: hrBucketFailed,
+        processed: entries.length,
+      },
+    });
+  }
+
   // v1.4.25 W16c — kick off PR detection for this user. We always
   // enqueue when at least one row was written (or the batch had any
   // measurements to consider) so a single off-day reading still gets
@@ -693,6 +763,9 @@ async function postBatch(request: NextRequest): Promise<Response> {
       updated: updatedCount,
       duplicates: duplicateCount,
       skipped: skipped.length,
+      // v1.32.1 (issue #585) — per-reason skip breakdown; see the
+      // `skippedByReason` build above.
+      skipped_by_reason: skippedByReason,
       failed: failedCount,
     },
   });

src/lib/analytics/__tests__/intraday-pulse-io.test.ts

@@ -13,6 +13,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
 
 const findManyMock = vi.fn();
 const workoutFindManyMock = vi.fn();
+const { annotateMock } = vi.hoisted(() => ({ annotateMock: vi.fn() }));
 
 vi.mock("@/lib/db", () => ({
   prisma: {
@@ -21,6 +22,8 @@ vi.mock("@/lib/db", () => ({
   },
 }));
 
+vi.mock("@/lib/logging/context", () => ({ annotate: annotateMock }));
+
 import { loadIntradayPulse } from "@/lib/analytics/intraday-pulse-io";
 import {
   resolveRestingPulseSeries,
@@ -42,6 +45,7 @@ describe("loadIntradayPulse", () => {
     findManyMock.mockReset();
     workoutFindManyMock.mockReset();
     workoutFindManyMock.mockResolvedValue([]);
+    annotateMock.mockReset();
   });
 
   it("M2 — queries PULSE/step rows against the exact local-day window, not the padded lead/trail superset", async () => {
@@ -209,4 +213,59 @@ describe("loadIntradayPulse", () => {
       { startMinute: 490, mean: 75, count: 2, min: 70, max: 81 },
     ]);
   });
+
+  it("v1.32.1 (issue #585) — annotates the per-shape row/bucket breakdown for observability", async () => {
+    const tz = "UTC";
+    const dateKey = "2026-06-15";
+
+    const restingRows = Array.from({ length: 20 }, (_, i) => ({
+      value: 55,
+      measuredAt: new Date(
+        `2026-05-${String((i % 28) + 1).padStart(2, "0")}T06:00:00.000Z`,
+      ),
+    }));
+    // One uploaded 10-min bucket row and one raw per-sample row sharing the
+    // day — the DTO merges them (bucket overlays raw on a slot collision),
+    // but the annotation must still report each SHAPE's own row count, not
+    // just the merged bucket total.
+    const uploadedBucket = {
+      value: 72,
+      valueMin: 64,
+      valueMax: 88,
+      measuredAt: new Date("2026-06-15T08:09:00.000Z"),
+      externalId:
+        "stats:HKQuantityTypeIdentifierHeartRate:2026-06-15T08:00:00.000Z",
+    };
+    const rawSample = {
+      value: 65,
+      valueMin: null,
+      valueMax: null,
+      measuredAt: new Date("2026-06-15T09:05:00.000Z"),
+      externalId: "sample:abc123",
+    };
+
+    findManyMock.mockImplementation(
+      async (args: {
+        where: { type: string; measuredAt?: unknown };
+        take?: number;
+      }) => {
+        if (args.where.type === "RESTING_HEART_RATE") return restingRows;
+        if (args.where.type === "PULSE" && args.where.measuredAt) {
+          return [uploadedBucket, rawSample];
+        }
+        return [];
+      },
+    );
+
+    await loadIntradayPulse("user-1", tz, dateKey);
+
+    expect(annotateMock).toHaveBeenCalledWith({
+      meta: {
+        intraday_raw_sample_rows: 1,
+        intraday_uploaded_bucket_rows: 1,
+        intraday_folded_hourly_rows: 0,
+        intraday_ten_min_buckets: 2,
+      },
+    });
+  });
 });

src/lib/analytics/intraday-pulse-io.ts

@@ -16,6 +16,7 @@
  * digest's `tension_window` builder, so the signal is computed one way.
  */
 import { prisma } from "@/lib/db";
+import { annotate } from "@/lib/logging/context";
 import { percentile } from "@/lib/insights/strain-score";
 import { resolveRestingPulseSeries } from "@/lib/analytics/resting-pulse";
 import { localDayWindow } from "@/lib/measurements/consolidation-tz";
@@ -349,6 +350,26 @@ export async function loadIntradayPulse(
         })
       : null;
 
+  // v1.32.1 (issue #585) — per-shape row/bucket breakdown. A report of
+  // "the chart looks sparser after the ZIP-import cutoff" is otherwise
+  // undiagnosable from server logs: the route's own annotation only ever
+  // surfaced the FINAL bucket count, with no visibility into which
+  // ingestion shape (raw ZIP-imported samples vs uploaded 10-min
+  // aggregates) the day's data actually came from, nor how many DB rows
+  // fed each shape before bucketing. Grepping `insights.pulse.intraday`
+  // now shows both — e.g. `uploaded_bucket_rows: 6` next to
+  // `ten_min_buckets: 6` confirms the uploaded shape IS being read and
+  // placed 1:1, narrowing a future report to "the client uploaded too few
+  // buckets" rather than leaving the read path itself as a live suspect.
+  annotate({
+    meta: {
+      intraday_raw_sample_rows: rawPulseRows.length,
+      intraday_uploaded_bucket_rows: uploadedBucketRows.length,
+      intraday_folded_hourly_rows: foldedPulseRows.length,
+      intraday_ten_min_buckets: tenMinSeries.length,
+    },
+  });
+
   return {
     dateKey,
     timezone,