fix(fitbit): single sync watermark, reject empty refresh token, correct interval time-anchor, mark provider experimental

Snapshot the incremental watermark once per sync cycle and thread the same
lower bound to every resource (metrics, activity, sleep, workout), stamping
lastSyncedAt once at the end. Previously each resource read lastSyncedAt then
stamped now() independently, so the first resource advanced the watermark and
later resources only fetched the last overlap window — silently dropping the
gap after an outage longer than the overlap. The full/backfill path is
unaffected (it passes no lower bound).

Reject an OAuth callback whose token response carries no refresh_token instead
of persisting an empty string. An empty refresh token bricks every future
token refresh; the callback now errors out cleanly so the user re-consents
rather than landing a dead connection.

Model the INTERVAL data types (steps/distance/calories/floors daily totals,
sleep and exercise sessions) on Google Health's interval.start_time /
interval.civil_start_time anchor instead of a sample_time / bare date. The
incremental filter for sleep/exercise previously targeted
sample_time.physical_time, which 400s/empties for interval types and stalls
incremental sync. Daily-summary metrics and spot readings keep their existing
date / sample anchors. Shape mismatches still fail safe (return []/null).

Mark the Google Health (Fitbit & Pixel) settings card as experimental with a
badge and a short note that data coverage is still being verified, in all six
locales.

(cherry picked from commit 3f116802149eebe9144729e2779f7f4405f248f2)

messages/de.json

@@ -3554,6 +3554,8 @@
     "fitbitTag": "Fitbit & Pixel",
     "fitbitDescription": "Verbinde Google Health, um deine Fitbit- und Pixel-Watch-Werte zu synchronisieren — Gewicht, Körperfett, Ruhepuls, Sauerstoffsättigung, Herzfrequenzvariabilität und mehr.",
     "fitbitOverlapNote": "Wenn Google Health und eine andere Quelle denselben Vitalwert liefern — Ruhepuls, Sauerstoffsättigung, Körpertemperatur oder Atemfrequenz —, siehst du möglicherweise beide Werte, bis sich ein künftiges Update auf eine bevorzugte Quelle festlegt.",
+    "fitbitExperimentalBadge": "Experimentell",
+    "fitbitExperimentalNote": "Diese Integration befindet sich in der Beta — die Datenabdeckung und die Feldzuordnung werden noch gegen echte Google-Health-Konten geprüft, daher können einzelne Werte fehlen oder sich in einem künftigen Update ändern.",
     "fitbitCredentials": "API-Zugangsdaten",
     "fitbitCredentialsHelp": "Registriere deinen eigenen Google-Cloud-OAuth-Client für die Google-Health-API und füge hier dessen Client-ID und Client-Secret ein.",
     "fitbitClientId": "Client-ID",

messages/en.json

@@ -3554,6 +3554,8 @@
     "fitbitTag": "Fitbit & Pixel",
     "fitbitDescription": "Connect Google Health to sync your Fitbit and Pixel Watch metrics — weight, body fat, resting heart rate, blood oxygen, heart-rate variability and more.",
     "fitbitOverlapNote": "If Google Health and another source both supply a vital — resting heart rate, blood oxygen, body temperature or respiratory rate — you may see both values until a future update settles on a single preferred source.",
+    "fitbitExperimentalBadge": "Experimental",
+    "fitbitExperimentalNote": "This integration is in beta — data coverage and field mapping are still being verified against live Google Health accounts, so some values may be missing or change in a future update.",
     "fitbitCredentials": "API Credentials",
     "fitbitCredentialsHelp": "Register your own Google Cloud OAuth client for the Google Health API and paste its Client ID and Client Secret here.",
     "fitbitClientId": "Client ID",

messages/es.json

@@ -3554,6 +3554,8 @@
     "fitbitTag": "Fitbit y Pixel",
     "fitbitDescription": "Conecta Google Health para sincronizar tus métricas de Fitbit y Pixel Watch: peso, grasa corporal, frecuencia cardíaca en reposo, saturación de oxígeno, variabilidad de la frecuencia cardíaca y más.",
     "fitbitOverlapNote": "Si Google Health y otra fuente proporcionan la misma constante vital —frecuencia cardíaca en reposo, saturación de oxígeno, temperatura corporal o frecuencia respiratoria—, es posible que veas ambos valores hasta que una actualización futura defina una única fuente preferida.",
+    "fitbitExperimentalBadge": "Experimental",
+    "fitbitExperimentalNote": "Esta integración está en fase beta: la cobertura de datos y la asignación de campos aún se están verificando con cuentas reales de Google Health, por lo que algunos valores pueden faltar o cambiar en una actualización futura.",
     "fitbitCredentials": "Credenciales de API",
     "fitbitCredentialsHelp": "Registra tu propio cliente OAuth de Google Cloud para la API de Google Health y pega aquí su Client ID y Client Secret.",
     "fitbitClientId": "Client ID",

messages/fr.json

@@ -3554,6 +3554,8 @@
     "fitbitTag": "Fitbit et Pixel",
     "fitbitDescription": "Connectez Google Health pour synchroniser vos données Fitbit et Pixel Watch : poids, masse grasse, fréquence cardiaque au repos, saturation en oxygène, variabilité de la fréquence cardiaque et plus encore.",
     "fitbitOverlapNote": "Si Google Health et une autre source fournissent la même constante vitale — fréquence cardiaque au repos, saturation en oxygène, température corporelle ou fréquence respiratoire —, vous pourriez voir les deux valeurs jusqu'à ce qu'une future mise à jour retienne une seule source préférée.",
+    "fitbitExperimentalBadge": "Expérimental",
+    "fitbitExperimentalNote": "Cette intégration est en bêta : la couverture des données et la correspondance des champs sont encore en cours de vérification avec de vrais comptes Google Health, donc certaines valeurs peuvent manquer ou changer lors d'une future mise à jour.",
     "fitbitCredentials": "Identifiants d'API",
     "fitbitCredentialsHelp": "Enregistrez votre propre client OAuth Google Cloud pour l'API Google Health et collez ici son Client ID et son Client Secret.",
     "fitbitClientId": "Client ID",

messages/it.json

@@ -3554,6 +3554,8 @@
     "fitbitTag": "Fitbit e Pixel",
     "fitbitDescription": "Collega Google Health per sincronizzare i dati di Fitbit e Pixel Watch: peso, massa grassa, frequenza cardiaca a riposo, saturazione di ossigeno, variabilità della frequenza cardiaca e altro.",
     "fitbitOverlapNote": "Se Google Health e un'altra fonte forniscono lo stesso parametro vitale — frequenza cardiaca a riposo, saturazione di ossigeno, temperatura corporea o frequenza respiratoria —, potresti vedere entrambi i valori finché un futuro aggiornamento non sceglierà un'unica fonte preferita.",
+    "fitbitExperimentalBadge": "Sperimentale",
+    "fitbitExperimentalNote": "Questa integrazione è in versione beta: la copertura dei dati e la mappatura dei campi sono ancora in fase di verifica con account Google Health reali, quindi alcuni valori potrebbero mancare o cambiare in un futuro aggiornamento.",
     "fitbitCredentials": "Credenziali API",
     "fitbitCredentialsHelp": "Registra il tuo client OAuth di Google Cloud per l'API Google Health e incolla qui il suo Client ID e Client Secret.",
     "fitbitClientId": "Client ID",

messages/pl.json

@@ -3554,6 +3554,8 @@
     "fitbitTag": "Fitbit i Pixel",
     "fitbitDescription": "Połącz Google Health, aby synchronizować dane z Fitbita i Pixel Watch — wagę, tkankę tłuszczową, tętno spoczynkowe, saturację, zmienność rytmu serca i więcej.",
     "fitbitOverlapNote": "Jeśli Google Health i inne źródło dostarczają ten sam parametr życiowy — tętno spoczynkowe, saturację, temperaturę ciała lub częstość oddechów — możesz widzieć obie wartości, dopóki przyszła aktualizacja nie wybierze jednego preferowanego źródła.",
+    "fitbitExperimentalBadge": "Eksperymentalne",
+    "fitbitExperimentalNote": "Ta integracja jest w wersji beta — zakres danych i mapowanie pól są wciąż weryfikowane na rzeczywistych kontach Google Health, więc niektóre wartości mogą być niedostępne lub zmienić się w przyszłej aktualizacji.",
     "fitbitCredentials": "Dane uwierzytelniające API",
     "fitbitCredentialsHelp": "Zarejestruj własny klient OAuth Google Cloud dla interfejsu Google Health API i wklej tutaj jego Client ID oraz Client Secret.",
     "fitbitClientId": "Client ID",

src/app/api/fitbit/callback/route.ts

@@ -109,30 +109,40 @@ export const GET = apiHandler(async (request: NextRequest) => {
     }
 
     const tokens = await exchangeCode(code, creds);
+
+    // The initial code exchange MUST return a refresh token (the connect route
+    // forces `prompt=consent` + `access_type=offline`). If Google ever omits it,
+    // refuse to persist — a stored empty refresh token silently bricks every
+    // future token refresh (the refresh POST then 400s and the connection dies
+    // the moment the access token expires). Error out cleanly so the user
+    // re-consents rather than landing a dead connection.
+    if (!tokens.refresh_token) {
+      getEvent()?.addWarning(
+        `fitbit callback: token response carried no refresh_token for user ${user.id}`,
+      );
+      annotate({ meta: { reason: "no_refresh_token" } });
+      return ERR("no_refresh_token");
+    }
+
     const profile = await fetchProfile(tokens.access_token);
     const fitbitUserId = resolveFitbitUserId(profile);
     const expiresAt = new Date(Date.now() + tokens.expires_in * 1000);
 
-    // The initial code exchange always returns a refresh token (the connect
-    // route forces `prompt=consent` + `access_type=offline`). Fall back to the
-    // existing stored token defensively if Google ever omits it on a re-consent.
     await prisma.fitbitConnection.upsert({
       where: { userId: user.id },
       update: {
         fitbitUserId,
         accessToken: encrypt(tokens.access_token),
+        refreshToken: encrypt(tokens.refresh_token),
         tokenExpiresAt: expiresAt,
         scope: tokens.scope ?? FITBIT_OAUTH_SCOPE,
         backfillCompletedAt: null,
-        ...(tokens.refresh_token
-          ? { refreshToken: encrypt(tokens.refresh_token) }
-          : {}),
       },
       create: {
         userId: user.id,
         fitbitUserId,
         accessToken: encrypt(tokens.access_token),
-        refreshToken: encrypt(tokens.refresh_token ?? ""),
+        refreshToken: encrypt(tokens.refresh_token),
         tokenExpiresAt: expiresAt,
         scope: tokens.scope ?? FITBIT_OAUTH_SCOPE,
       },

src/components/settings/integrations-section.tsx

@@ -26,6 +26,7 @@ import {
   AlertDialogTitle,
   AlertDialogTrigger,
 } from "@/components/ui/alert-dialog";
+import { Badge } from "@/components/ui/badge";
 import { Button } from "@/components/ui/button";
 import { Input } from "@/components/ui/input";
 import { Label } from "@/components/ui/label";
@@ -1298,18 +1299,31 @@ function FitbitCard({
       className="bg-card border-border rounded-xl border p-6"
     >
       <div className="flex flex-wrap items-start justify-between gap-2">
-        <div className="flex items-center gap-2">
+        <div className="flex flex-wrap items-center gap-2">
           <HeartPulse className="text-primary h-5 w-5" />
           <h2 className="text-lg font-semibold">{t("settings.fitbit")}</h2>
           <span className="bg-muted text-muted-foreground rounded-full px-2 py-0.5 text-[0.6875rem] font-medium">
             {t("settings.fitbitTag")}
           </span>
+          <Badge
+            variant="outline"
+            data-testid="fitbit-experimental-badge"
+            className="border-amber-500/50 text-amber-600 dark:text-amber-400"
+          >
+            {t("settings.fitbitExperimentalBadge")}
+          </Badge>
         </div>
         <IntegrationStatusPill state={pillState} lastSyncAt={pillLastSyncAt} />
       </div>
       <p className="text-muted-foreground mt-1 text-xs">
         {t("settings.fitbitDescription")}
       </p>
+      <p
+        data-testid="fitbit-experimental-note"
+        className="text-muted-foreground/80 mt-2 text-xs"
+      >
+        {t("settings.fitbitExperimentalNote")}
+      </p>
       <p className="text-muted-foreground/80 mt-2 text-xs">
         {t("settings.fitbitOverlapNote")}
       </p>

src/lib/fitbit/__tests__/client.test.ts

@@ -231,6 +231,41 @@ describe("fetchDataPoints", () => {
     );
   });
 
+  it("filters INTERVAL types on interval.start_time, not a sample_time or a bare date", async () => {
+    // steps / distance / calories / floors, sleep and exercise are INTERVAL data
+    // types — Google anchors them on `interval.start_time`. A sample_time filter
+    // 400s/empties for these and stalls incremental sync.
+    const stepsMock = installFetchMock([{ status: 200, body: { dataPoints: [] } }]);
+    await fetchDataPoints(FITBIT_DATA_TYPES.steps, "at", "fetchSteps", {
+      start: new Date("2026-03-04T10:00:00.000Z"),
+    });
+    expect(
+      new URL((stepsMock.mock.calls[0] as unknown as [string])[0]).searchParams.get(
+        "filter",
+      ),
+    ).toBe('steps.interval.start_time >= "2026-03-04T10:00:00.000Z"');
+
+    const sleepMock = installFetchMock([{ status: 200, body: { dataPoints: [] } }]);
+    await fetchDataPoints(FITBIT_DATA_TYPES.sleep, "at", "fetchSleep", {
+      start: new Date("2026-03-04T10:00:00.000Z"),
+    });
+    expect(
+      new URL((sleepMock.mock.calls[0] as unknown as [string])[0]).searchParams.get(
+        "filter",
+      ),
+    ).toBe('sleep.interval.start_time >= "2026-03-04T10:00:00.000Z"');
+
+    const exMock = installFetchMock([{ status: 200, body: { dataPoints: [] } }]);
+    await fetchDataPoints(FITBIT_DATA_TYPES.exercise, "at", "fetchExercise", {
+      start: new Date("2026-03-04T10:00:00.000Z"),
+    });
+    expect(
+      new URL((exMock.mock.calls[0] as unknown as [string])[0]).searchParams.get(
+        "filter",
+      ),
+    ).toBe('exercise.interval.start_time >= "2026-03-04T10:00:00.000Z"');
+  });
+
   it("throws a classified FitbitApiError on a non-2xx page", async () => {
     installFetchMock([{ status: 403, body: {} }]);
     await expect(
@@ -337,10 +372,17 @@ describe("FITBIT_FIELD_MAP ↔ FITBIT_DATA_TYPES casing parity", () => {
 });
 
 describe("activity mappers (cumulative daily)", () => {
-  const day = { year: 2026, month: 5, day: 10 };
+  // INTERVAL data types: Google buckets the daily total into an `interval` with
+  // a `start_time` (physical instant) — NOT a bare civil `date`. The mapper
+  // anchors measuredAt on `interval.start_time` and day-keys the externalId.
+  const interval = { start_time: "2026-05-10T00:00:00.000Z" };
+  // A civil-only fallback shape (no physical start_time) still day-keys cleanly.
+  const civilInterval = {
+    civil_start_time: { year: 2026, month: 5, day: 10 },
+  };
 
   it("maps steps with a stats:-shaped daily externalId and preserves a zero", () => {
-    const out = mapSteps({ steps: { count: 8421, date: day } });
+    const out = mapSteps({ steps: { count: 8421, interval } });
     expect(out).toHaveLength(1);
     expect(out[0]).toMatchObject({
       type: "ACTIVITY_STEPS",
@@ -350,30 +392,34 @@ describe("activity mappers (cumulative daily)", () => {
     });
     // fieldTag carries <tag>:<YYYY-MM-DD>; the sync layer prefixes `stats:`.
     expect(out[0]!.fieldTag).toBe("steps:2026-05-10");
-    // measuredAt anchored at the civil day (noon UTC) so it day-keys cleanly.
-    expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T12:00:00.000Z");
+    // measuredAt anchored on the interval start instant.
+    expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T00:00:00.000Z");
+
+    // The civil-only fallback day-keys the externalId at UTC midday.
+    const civil = mapSteps({ steps: { count: 100, interval: civilInterval } });
+    expect(civil[0]!.fieldTag).toBe("steps:2026-05-10");
 
     // A rest day records 0 steps — that is real data, not a gap.
-    const rest = mapSteps({ steps: { count: 0, date: day } });
+    const rest = mapSteps({ steps: { count: 0, interval } });
     expect(rest).toHaveLength(1);
     expect(rest[0]!.value).toBe(0);
   });
 
   it("maps distance in metres, converting km → m when reported in km", () => {
-    expect(mapDistance({ distance: { meters: 6200, date: day } })[0]).toMatchObject({
+    expect(mapDistance({ distance: { meters: 6200, interval } })[0]).toMatchObject({
       type: "WALKING_RUNNING_DISTANCE",
       value: 6200,
       unit: "m",
     });
     // km path multiplies by 1000.
     expect(
-      mapDistance({ distance: { kilometers: 6.2, date: day } })[0]!.value,
+      mapDistance({ distance: { kilometers: 6.2, interval } })[0]!.value,
     ).toBe(6200);
   });
 
   it("maps the ACTIVE-calories portion into ACTIVE_ENERGY_BURNED", () => {
     const out = mapActiveCalories({
-      active_calories: { active_kilocalories: 540, date: day },
+      active_calories: { active_kilocalories: 540, interval },
     });
     expect(out[0]).toMatchObject({
       type: "ACTIVE_ENERGY_BURNED",
@@ -383,15 +429,17 @@ describe("activity mappers (cumulative daily)", () => {
   });
 
   it("maps floors and preserves a zero", () => {
-    expect(mapFloors({ floors: { count: 12, date: day } })[0]).toMatchObject({
+    expect(mapFloors({ floors: { count: 12, interval } })[0]).toMatchObject({
       type: "FLIGHTS_CLIMBED",
       value: 12,
       unit: "flights",
     });
-    expect(mapFloors({ floors: { count: 0, date: day } })[0]!.value).toBe(0);
+    expect(mapFloors({ floors: { count: 0, interval } })[0]!.value).toBe(0);
   });
 
   it("maps VO2 max (strictly positive, daily latest-wins) and drops a zero", () => {
+    // VO2 max is a daily-summary metric (civil `date` anchor), not an interval.
+    const day = { year: 2026, month: 5, day: 10 };
     const out = mapVo2Max({
       vo2_max: { milliliters_per_kilogram_per_minute: 47.3, date: day },
     });
@@ -467,6 +515,26 @@ describe("sleep-stage mapping", () => {
     );
   });
 
+  it("anchors the session externalId on sleep.interval.end_time for an INTERVAL-shaped point", () => {
+    const out = mapSleepSession({
+      sleep: {
+        interval: {
+          start_time: "2026-05-10T22:00:00.000Z",
+          end_time: "2026-05-11T06:00:00.000Z",
+        },
+        segments: [
+          {
+            stage: "deep",
+            startTime: "2026-05-10T22:30:00.000Z",
+            endTime: "2026-05-10T23:30:00.000Z",
+          },
+        ],
+      },
+    });
+    const deep = out.find((m) => m.sleepStage === "DEEP");
+    expect(deep!.fieldTag).toBe("2026-05-11T06:00:00.000Z:sleep_deep");
+  });
+
   it("yields nothing for a session with no parseable segments", () => {
     expect(mapSleepSession({ sleep: {} })).toHaveLength(0);
     expect(mapSleepSession({})).toHaveLength(0);
@@ -513,6 +581,23 @@ describe("workout mapping", () => {
     expect(w!.avgHeartRate).toBeNull();
   });
 
+  it("reads the start/end from exercise.interval for an INTERVAL-shaped point", () => {
+    const w = mapWorkout({
+      exercise: {
+        session_id: "ex-iv",
+        activity_type: "run",
+        interval: {
+          start_time: "2026-05-10T07:00:00.000Z",
+          end_time: "2026-05-10T07:45:00.000Z",
+        },
+      },
+    });
+    expect(w).not.toBeNull();
+    expect(w!.startedAt.toISOString()).toBe("2026-05-10T07:00:00.000Z");
+    expect(w!.endedAt.toISOString()).toBe("2026-05-10T07:45:00.000Z");
+    expect(w!.durationSec).toBe(45 * 60);
+  });
+
   it("returns null for a session with no usable time span", () => {
     expect(mapWorkout({ exercise: { startTime: "2026-05-10T07:00:00.000Z" } })).toBeNull();
     expect(

src/lib/fitbit/__tests__/sync-user-fitbit.test.ts

@@ -19,6 +19,8 @@ const {
   syncUserActivity,
   syncUserSleep,
   syncUserWorkout,
+  findUnique,
+  update,
 } = vi.hoisted(() => ({
   recordSyncSuccess: vi.fn<(...a: unknown[]) => Promise<void>>(async () => {}),
   recordSyncFailure: vi.fn<(...a: unknown[]) => Promise<void>>(async () => {}),
@@ -29,9 +31,13 @@ const {
   syncUserActivity: vi.fn<(...a: unknown[]) => Promise<number>>(),
   syncUserSleep: vi.fn<(...a: unknown[]) => Promise<number>>(),
   syncUserWorkout: vi.fn<(...a: unknown[]) => Promise<number>>(),
+  findUnique: vi.fn<(...a: unknown[]) => Promise<unknown>>(),
+  update: vi.fn<(...a: unknown[]) => Promise<unknown>>(async () => ({})),
 }));
 
-vi.mock("@/lib/db", () => ({ prisma: {} }));
+vi.mock("@/lib/db", () => ({
+  prisma: { fitbitConnection: { findUnique, update } },
+}));
 vi.mock("@/lib/crypto", () => ({
   encrypt: (s: string) => s,
   decrypt: (s: string) => s,
@@ -83,11 +89,18 @@ async function softSkip403(...args: unknown[]): Promise<number> {
   );
 }
 
+/** A connection with a fixed last-synced watermark for the snapshot tests. */
+const LAST_SYNCED = new Date("2026-06-01T00:00:00.000Z");
+
 beforeEach(() => {
   vi.clearAllMocks();
   isReauthRequired.mockResolvedValue(false);
+  // Default: a connected account with a known watermark.
+  findUnique.mockResolvedValue({ lastSyncedAt: LAST_SYNCED });
+  update.mockResolvedValue({});
   // The W5 resources default to a quiet success (nothing new) so a test that
   // exercises only the metrics resource doesn't trip the all-soft-skipped guard.
+  syncUserMetrics.mockResolvedValue(0);
   syncUserActivity.mockResolvedValue(0);
   syncUserSleep.mockResolvedValue(0);
   syncUserWorkout.mockResolvedValue(0);
@@ -153,5 +166,85 @@ describe("syncUserFitbit — all-403 looks-healthy guard", () => {
     expect(total).toBe(0);
     expect(recordSyncSuccess).not.toHaveBeenCalled();
     expect(syncUserMetrics).not.toHaveBeenCalled();
+    // No connection read either — the reauth gate runs first.
+    expect(findUnique).not.toHaveBeenCalled();
+  });
+});
+
+describe("syncUserFitbit — single cycle-wide watermark", () => {
+  it("snapshots the incremental start ONCE and threads the SAME watermark to every resource", async () => {
+    syncUserMetrics.mockResolvedValue(1);
+
+    await syncUserFitbit("user1");
+
+    // The connection's `lastSyncedAt` is read exactly once for the whole cycle.
+    expect(findUnique).toHaveBeenCalledTimes(1);
+
+    // Every resource receives the identical resolved `start` (lastSyncedAt - 24h
+    // overlap). A per-resource read+stamp would have given later resources a
+    // start of ~now, dropping the gap after an outage longer than the overlap.
+    const expectedStart = new Date(
+      LAST_SYNCED.getTime() - 24 * 60 * 60 * 1000,
+    );
+    for (const fn of [
+      syncUserMetrics,
+      syncUserActivity,
+      syncUserSleep,
+      syncUserWorkout,
+    ]) {
+      expect(fn).toHaveBeenCalledTimes(1);
+      const opts = fn.mock.calls[0]![1] as { start?: Date; fullSync?: boolean };
+      expect(opts.start?.getTime()).toBe(expectedStart.getTime());
+    }
+  });
+
+  it("stamps markSynced exactly ONCE at the end of a non-degenerate cycle", async () => {
+    syncUserMetrics.mockResolvedValue(3);
+
+    await syncUserFitbit("user1");
+
+    // The single end-of-cycle stamp — never per resource.
+    expect(update).toHaveBeenCalledTimes(1);
+    expect(update.mock.calls[0]![0]).toMatchObject({
+      where: { userId: "user1" },
+    });
+  });
+
+  it("does NOT stamp markSynced when every resource soft-skipped (all-403)", async () => {
+    syncUserMetrics.mockImplementation(softSkip403);
+    syncUserActivity.mockImplementation(softSkip403);
+    syncUserSleep.mockImplementation(softSkip403);
+    syncUserWorkout.mockImplementation(softSkip403);
+
+    await syncUserFitbit("user1");
+
+    // Leaving the watermark untouched on a dead cycle keeps the next tick's
+    // window from silently advancing past the gap.
+    expect(update).not.toHaveBeenCalled();
+    expect(recordSyncSuccess).not.toHaveBeenCalled();
+  });
+
+  it("passes fullSync through with an undefined start on a backfill cycle", async () => {
+    syncUserMetrics.mockResolvedValue(2);
+
+    await syncUserFitbit("user1", { fullSync: true });
+
+    const opts = syncUserMetrics.mock.calls[0]![1] as {
+      start?: Date;
+      fullSync?: boolean;
+    };
+    expect(opts.fullSync).toBe(true);
+    // Full sync walks deep history with no lower bound.
+    expect(opts.start).toBeUndefined();
+  });
+
+  it("returns 0 without running resources when the connection is gone", async () => {
+    findUnique.mockResolvedValue(null);
+
+    const total = await syncUserFitbit("user1");
+
+    expect(total).toBe(0);
+    expect(syncUserMetrics).not.toHaveBeenCalled();
+    expect(update).not.toHaveBeenCalled();
   });
 });

src/lib/fitbit/client.ts

@@ -274,10 +274,15 @@ export interface FitbitDataType {
   filter: string;
   /**
    * Which time anchor the filter / measuredAt resolution targets:
-   *   - `sample` → spot reading (`{type}.sample_time.physical_time`).
-   *   - `date`   → daily summary (`{type}.date`).
+   *   - `sample`   → spot reading (`{type}.sample_time.physical_time`).
+   *   - `date`     → daily summary keyed on a civil date (`{type}.date`).
+   *   - `interval` → an INTERVAL data type (steps / distance / calories / floors,
+   *     sleep, exercise). Google Health anchors these on `{type}.interval.start_time`
+   *     (the physical instant) with `{type}.interval.civil_start_time` as the civil
+   *     fallback — NOT on a `sample_time` (which 400s/empties for interval types and
+   *     stalls the incremental filter) and NOT on a bare `date`.
    */
-  timeField: "sample" | "date";
+  timeField: "sample" | "date" | "interval";
 }
 
 /**
@@ -316,27 +321,34 @@ export const FITBIT_DATA_TYPES = {
     timeField: "date",
   },
   // ── Activity bundle (W5) — daily cumulative totals ─────────────
-  // Scope: `googlehealth.activity_and_fitness.readonly`. Each is a per-day
-  // summary (`timeField: "date"`); the externalId carries the `stats:` prefix
-  // so a re-fetched day overwrites in place (mirrors the Apple-Health
-  // `stats:<HK>:<YYYY-MM-DD>` daily-total overwrite contract).
-  steps: { path: "steps", filter: "steps", timeField: "date" },
-  distance: { path: "distance", filter: "distance", timeField: "date" },
+  // Scope: `googlehealth.activity_and_fitness.readonly`. These are INTERVAL data
+  // types: Google buckets a daily total into an `interval` (a `start_time` +
+  // `end_time`, with `civil_start_time` for the calendar-day grain). The
+  // incremental filter targets `interval.start_time` and the externalId carries
+  // the `stats:` prefix so a re-fetched day overwrites in place (mirrors the
+  // Apple-Health `stats:<HK>:<YYYY-MM-DD>` daily-total overwrite contract).
+  steps: { path: "steps", filter: "steps", timeField: "interval" },
+  distance: { path: "distance", filter: "distance", timeField: "interval" },
   activeCalories: {
     path: "active-calories",
     filter: "active_calories",
-    timeField: "date",
+    timeField: "interval",
   },
-  floors: { path: "floors", filter: "floors", timeField: "date" },
+  floors: { path: "floors", filter: "floors", timeField: "interval" },
+  // VO2 max is a daily-summary metric (one civil-date reading), not an interval
+  // bucket — keep the `date` anchor.
   vo2Max: { path: "vo2-max", filter: "vo2_max", timeField: "date" },
   // ── Sleep bundle (W5) ──────────────────────────────────────────
-  // Scope: `googlehealth.sleep.readonly`. A sleep session carries a start +
-  // end + a per-stage breakdown; mapped to per-stage SLEEP_DURATION rows.
-  sleep: { path: "sleep", filter: "sleep", timeField: "sample" },
+  // Scope: `googlehealth.sleep.readonly`. A sleep session is an INTERVAL data
+  // type (a start + end span carrying a per-stage breakdown); the incremental
+  // filter must target `interval.start_time`, not a `sample_time` (which 400s
+  // for interval types). Mapped to per-stage SLEEP_DURATION rows.
+  sleep: { path: "sleep", filter: "sleep", timeField: "interval" },
   // ── Exercise bundle (W5) ───────────────────────────────────────
-  // Scope: `googlehealth.activity_and_fitness.readonly`. An exercise session
-  // → a `Workout` row (NOT a Measurement).
-  exercise: { path: "exercise", filter: "exercise", timeField: "sample" },
+  // Scope: `googlehealth.activity_and_fitness.readonly`. An exercise session is
+  // an INTERVAL data type (a start + end span) → a `Workout` row (NOT a
+  // Measurement). The incremental filter targets `interval.start_time`.
+  exercise: { path: "exercise", filter: "exercise", timeField: "interval" },
 } as const satisfies Record<string, FitbitDataType>;
 
 export type FitbitDataTypeKey = keyof typeof FITBIT_DATA_TYPES;
@@ -361,6 +373,39 @@ interface DataPointQuery {
   maxPages?: number;
 }
 
+/**
+ * Build the incremental `filter` field + bound for one data type. Centralised so
+ * the predicate and the read-time anchor resolution can never drift on which
+ * field a given `timeField` targets.
+ *   - `sample`   → spot reading, filter on `{filter}.sample_time.physical_time`.
+ *   - `interval` → INTERVAL type, filter on `{filter}.interval.start_time`
+ *     (Google's anchor for steps/distance/calories/floors, sleep, exercise — a
+ *     `sample_time` filter 400s/empties for these and stalls incremental sync).
+ *   - `date`     → daily civil-date summary, filter on `{filter}.date`.
+ */
+export function incrementalFilter(
+  dataType: FitbitDataType,
+  start: Date,
+): { field: string; bound: string } {
+  switch (dataType.timeField) {
+    case "sample":
+      return {
+        field: `${dataType.filter}.sample_time.physical_time`,
+        bound: start.toISOString(),
+      };
+    case "interval":
+      return {
+        field: `${dataType.filter}.interval.start_time`,
+        bound: start.toISOString(),
+      };
+    case "date":
+      return {
+        field: `${dataType.filter}.date`,
+        bound: start.toISOString().slice(0, 10),
+      };
+  }
+}
+
 /**
  * Walk every `DataPoint` for one data type since the incremental cursor.
  * Mirrors the WHOOP `fetchCollection` page-loop: `nextPageToken` pagination, a
@@ -383,16 +428,12 @@ export async function fetchDataPoints(
   do {
     const params = new URLSearchParams({ pageSize: String(pageSize) });
     if (query.start) {
-      // snake_case filter prefix; sample readings filter on the sample time,
-      // daily summaries on the civil date.
-      const field =
-        dataType.timeField === "sample"
-          ? `${dataType.filter}.sample_time.physical_time`
-          : `${dataType.filter}.date`;
-      const bound =
-        dataType.timeField === "sample"
-          ? query.start.toISOString()
-          : query.start.toISOString().slice(0, 10);
+      // snake_case filter prefix. The filter field + bound shape follow the
+      // data type's time anchor:
+      //   - sample   → `{filter}.sample_time.physical_time` >= an ISO instant
+      //   - interval → `{filter}.interval.start_time`        >= an ISO instant
+      //   - date     → `{filter}.date`                       >= a civil date
+      const { field, bound } = incrementalFilter(dataType, query.start);
       params.set("filter", `${field} >= "${bound}"`);
     }
     if (pageToken) params.set("pageToken", pageToken);
@@ -512,11 +553,32 @@ function readPath(obj: unknown, path: string): unknown {
   return cur;
 }
 
+/** Parse a `{year,month,day}` civil-date object into a UTC-midday Date, or null. */
+function parseCivilDateObject(val: unknown): Date | null {
+  if (!val || typeof val !== "object") return null;
+  const o = val as Record<string, unknown>;
+  if (
+    typeof o.year === "number" &&
+    typeof o.month === "number" &&
+    typeof o.day === "number"
+  ) {
+    // Google civil dates are 1-based months; anchor at UTC midday so a
+    // timezone shift can't roll the civil day across a boundary.
+    return new Date(Date.UTC(o.year, o.month - 1, o.day, 12));
+  }
+  return null;
+}
+
 /**
- * Resolve a `DataPoint`'s measurement timestamp. Sample readings carry a
- * `sample_time.physical_time`; daily summaries carry a `date` (a civil date or a
- * `{year,month,day}` object). Falls back to the fetch time only when nothing
- * parses, so a row is never dropped for a missing anchor.
+ * Resolve a `DataPoint`'s measurement timestamp.
+ *   - `sample`   → `sample_time.physical_time` (a spot ISO instant).
+ *   - `interval` → `interval.start_time` (the bucket's start ISO instant), with
+ *     `interval.civil_start_time` (a civil string / `{year,month,day}`) as the
+ *     fallback. This is Google's anchor for the INTERVAL types; the prior `date`
+ *     read returned undefined for them and forced the fetch-time fallback.
+ *   - `date`     → `date` (a civil string or `{year,month,day}` object).
+ * Falls back to `fallback` only when nothing parses, so a row is never dropped
+ * for a missing anchor.
  */
 function resolveMeasuredAt(
   point: FitbitDataPoint,
@@ -529,24 +591,27 @@ function resolveMeasuredAt(
       const d = new Date(t);
       if (!Number.isNaN(d.getTime())) return d;
     }
+  } else if (dataType.timeField === "interval") {
+    const t = readPath(point, `${dataType.filter}.interval.start_time`);
+    if (typeof t === "string") {
+      const d = new Date(t);
+      if (!Number.isNaN(d.getTime())) return d;
+    }
+    const civil = readPath(point, `${dataType.filter}.interval.civil_start_time`);
+    if (typeof civil === "string") {
+      const d = new Date(civil);
+      if (!Number.isNaN(d.getTime())) return d;
+    }
+    const civilObj = parseCivilDateObject(civil);
+    if (civilObj) return civilObj;
   } else {
     const dateVal = readPath(point, `${dataType.filter}.date`);
     if (typeof dateVal === "string") {
       const d = new Date(dateVal);
       if (!Number.isNaN(d.getTime())) return d;
     }
-    if (dateVal && typeof dateVal === "object") {
-      const o = dateVal as Record<string, unknown>;
-      if (
-        typeof o.year === "number" &&
-        typeof o.month === "number" &&
-        typeof o.day === "number"
-      ) {
-        // Google civil dates are 1-based months; anchor at UTC midday so a
-        // timezone shift can't roll the civil day across a boundary.
-        return new Date(Date.UTC(o.year, o.month - 1, o.day, 12));
-      }
-    }
+    const civilObj = parseCivilDateObject(dateVal);
+    if (civilObj) return civilObj;
   }
   return fallback;
 }
@@ -559,7 +624,13 @@ function resolveMeasuredAt(
  */
 function externalAnchor(point: FitbitDataPoint, dataType: FitbitDataType): string {
   const at = resolveMeasuredAt(point, dataType, new Date(0));
-  if (dataType.timeField === "date") return at.toISOString().slice(0, 10);
+  // The mapper-driven interval types (steps/distance/calories/floors) are daily
+  // totals, so they share the civil-day externalId grain with `date` summaries —
+  // a re-fetched day overwrites in place. (Sleep/exercise are interval too but
+  // mint their own per-session anchors and never reach this helper.)
+  if (dataType.timeField === "date" || dataType.timeField === "interval") {
+    return at.toISOString().slice(0, 10);
+  }
   return at.toISOString();
 }
 
@@ -1086,6 +1157,8 @@ function readSleepSegments(point: FitbitDataPoint): FitbitSleepSegment[] {
  */
 function sleepSessionAnchor(point: FitbitDataPoint): string {
   const end =
+    readPath(point, "sleep.interval.end_time") ??
+    readPath(point, "interval.end_time") ??
     readPath(point, "sleep.endTime") ??
     readPath(point, "sleep.end_time") ??
     readPath(point, "endTime") ??
@@ -1095,6 +1168,8 @@ function sleepSessionAnchor(point: FitbitDataPoint): string {
     if (!Number.isNaN(d.getTime())) return d.toISOString();
   }
   const start =
+    readPath(point, "sleep.interval.start_time") ??
+    readPath(point, "interval.start_time") ??
     readPath(point, "sleep.startTime") ??
     readPath(point, "sleep.start_time") ??
     readPath(point, "startTime");
@@ -1252,15 +1327,19 @@ function readInstant(point: FitbitDataPoint, paths: string[]): Date | null {
 export function mapWorkout(point: FitbitDataPoint): FitbitMappedWorkout | null {
   const f = FITBIT_DATA_TYPES.exercise.filter;
   const startedAt = readInstant(point, [
+    `${f}.interval.start_time`,
     `${f}.startTime`,
     `${f}.start_time`,
     `${f}.sample_time.physical_time`,
+    "interval.start_time",
     "startTime",
     "start_time",
   ]);
   const endedAt = readInstant(point, [
+    `${f}.interval.end_time`,
     `${f}.endTime`,
     `${f}.end_time`,
+    "interval.end_time",
     "endTime",
     "end_time",
   ]);

src/lib/fitbit/mapping.md

@@ -4,8 +4,22 @@ Single source of truth for the Fitbit (Google Health API v4) data-type →
 HealthLog `Measurement` mapping. Keep this in sync with `FITBIT_FIELD_MAP` and
 the per-type mappers in `client.ts`. Every Fitbit row ingests server-side with
 `source = FITBIT` and `externalId = <anchor>:<fieldTag>` (the anchor is the data
-point's sample time for spot readings or its civil date for daily summaries; the
-field-tag disambiguates the metric).
+point's sample time for spot readings, its civil date for daily summaries, or its
+`interval.start_time` for INTERVAL types — steps/distance/calories/floors, sleep,
+exercise; the field-tag disambiguates the metric).
+
+## Time anchor by grain (design §A.1)
+
+Google Health exposes three time-anchor shapes. The incremental `filter` and the
+read-time `measuredAt`/externalId resolution must target the right one or the
+incremental sync stalls (a `sample_time` filter 400s/empties for an INTERVAL type):
+
+- **sample** — spot reading; anchor `{filter}.sample_time.physical_time`.
+- **date** — daily-summary metric (SpO2, HRV, RHR, respiratory rate, wrist temp,
+  VO2 max); anchor `{filter}.date` (a civil date / `{year,month,day}`).
+- **interval** — INTERVAL type (steps/distance/calories/floors daily totals, plus
+  sleep + exercise sessions); anchor `{filter}.interval.start_time` (a physical
+  instant) with `{filter}.interval.civil_start_time` as the civil fallback.
 
 Transport docs: `developers.google.com/health` (v4, post-Fitbit-Web-API). The
 per-type value-field JSON is **NOT fully published** — the mappers read a small

src/lib/fitbit/sync-activity.ts

@@ -36,12 +36,10 @@ import {
 import {
   getValidToken,
   handleCollectionFetchError,
-  incrementalStart,
-  markSynced,
   upsertFitbitMeasurements,
   type FitbitMeasurementUpsert,
+  type FitbitResourceSyncOptions,
 } from "./sync";
-import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 
 /** One mappable activity metric: its data-type encoding + the per-point mapper + a verb. */
@@ -79,20 +77,14 @@ function externalIdFor(m: FitbitMappedMeasurement): string {
 
 export async function syncUserActivity(
   userId: string,
-  opts: { fullSync?: boolean } = {},
+  opts: FitbitResourceSyncOptions = {},
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
 
-  const connection = await prisma.fitbitConnection.findUnique({
-    where: { userId },
-    select: { lastSyncedAt: true },
-  });
-  if (!connection) return 0;
-
-  const start = incrementalStart(connection.lastSyncedAt, {
-    fullSync: opts.fullSync,
-  });
+  // Cycle-wide watermark snapshotted once by `syncUserFitbit`; undefined on a
+  // full/backfill run.
+  const start = opts.start;
 
   let imported = 0;
   for (const resource of ACTIVITY_RESOURCES) {
@@ -124,7 +116,7 @@ export async function syncUserActivity(
     imported += await upsertFitbitMeasurements(userId, readings);
   }
 
-  await markSynced(userId);
+  // `markSynced` is owned by the orchestrator (`syncUserFitbit`).
   annotate({ action: { name: "fitbit.activity.sync", details: { imported } } });
   return imported;
 }

src/lib/fitbit/sync-metrics.ts

@@ -43,13 +43,12 @@ import {
 import {
   getValidToken,
   handleCollectionFetchError,
-  incrementalStart,
-  markSynced,
   upsertFitbitMeasurements,
   type FitbitMeasurementUpsert,
+  type FitbitResourceSyncOptions,
 } from "./sync";
-import { prisma } from "@/lib/db";
 import { getEvent } from "@/lib/logging/context";
+import { prisma } from "@/lib/db";
 
 /** One mappable metric: its data-type encoding + the per-point mapper + a verb. */
 interface MetricResource {
@@ -96,20 +95,15 @@ const METRIC_RESOURCES: MetricResource[] = [
 
 export async function syncUserMetrics(
   userId: string,
-  opts: { fullSync?: boolean } = {},
+  opts: FitbitResourceSyncOptions = {},
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
 
-  const connection = await prisma.fitbitConnection.findUnique({
-    where: { userId },
-    select: { lastSyncedAt: true },
-  });
-  if (!connection) return 0;
-
-  const start = incrementalStart(connection.lastSyncedAt, {
-    fullSync: opts.fullSync,
-  });
+  // The incremental lower bound is the cycle-wide watermark snapshotted once by
+  // `syncUserFitbit` — never re-read here, so a sibling resource's stamp can't
+  // shrink this one's window. Undefined on a full/backfill run.
+  const start = opts.start;
 
   let imported = 0;
 
@@ -184,6 +178,7 @@ export async function syncUserMetrics(
     );
   }
 
-  await markSynced(userId);
+  // `markSynced` is owned by the orchestrator (`syncUserFitbit`), stamped once
+  // at the end of the cycle — never here, so the watermark can't move mid-cycle.
   return imported;
 }

src/lib/fitbit/sync-sleep.ts

@@ -20,30 +20,22 @@ import { FITBIT_DATA_TYPES, fetchDataPoints, mapSleepSession } from "./client";
 import {
   getValidToken,
   handleCollectionFetchError,
-  incrementalStart,
-  markSynced,
   upsertFitbitMeasurements,
   type FitbitMeasurementUpsert,
+  type FitbitResourceSyncOptions,
 } from "./sync";
-import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 
 export async function syncUserSleep(
   userId: string,
-  opts: { fullSync?: boolean } = {},
+  opts: FitbitResourceSyncOptions = {},
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
 
-  const connection = await prisma.fitbitConnection.findUnique({
-    where: { userId },
-    select: { lastSyncedAt: true },
-  });
-  if (!connection) return 0;
-
-  const start = incrementalStart(connection.lastSyncedAt, {
-    fullSync: opts.fullSync,
-  });
+  // Cycle-wide watermark snapshotted once by `syncUserFitbit`; undefined on a
+  // full/backfill run.
+  const start = opts.start;
 
   let points: Record<string, unknown>[];
   try {
@@ -72,7 +64,7 @@ export async function syncUserSleep(
   }
 
   const imported = await upsertFitbitMeasurements(userId, readings);
-  await markSynced(userId);
+  // `markSynced` is owned by the orchestrator (`syncUserFitbit`).
   annotate({ action: { name: "fitbit.sleep.sync", details: { imported } } });
   return imported;
 }

src/lib/fitbit/sync-workout.ts

@@ -23,28 +23,21 @@ import {
 import {
   getValidToken,
   handleCollectionFetchError,
-  incrementalStart,
-  markSynced,
+  type FitbitResourceSyncOptions,
 } from "./sync";
 import { prisma } from "@/lib/db";
 import { annotate, getEvent } from "@/lib/logging/context";
 
 export async function syncUserWorkout(
   userId: string,
-  opts: { fullSync?: boolean } = {},
+  opts: FitbitResourceSyncOptions = {},
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
 
-  const connection = await prisma.fitbitConnection.findUnique({
-    where: { userId },
-    select: { lastSyncedAt: true },
-  });
-  if (!connection) return 0;
-
-  const start = incrementalStart(connection.lastSyncedAt, {
-    fullSync: opts.fullSync,
-  });
+  // Cycle-wide watermark snapshotted once by `syncUserFitbit`; undefined on a
+  // full/backfill run.
+  const start = opts.start;
 
   let points: Record<string, unknown>[];
   try {
@@ -104,7 +97,7 @@ export async function syncUserWorkout(
     }
   }
 
-  await markSynced(userId);
+  // `markSynced` is owned by the orchestrator (`syncUserFitbit`).
   annotate({ action: { name: "fitbit.workout.sync", details: { imported } } });
   return imported;
 }

src/lib/fitbit/sync.ts

@@ -186,6 +186,20 @@ export async function handleCollectionFetchError(
   throw err;
 }
 
+/**
+ * Options threaded from `syncUserFitbit` to each per-resource sync. The `start`
+ * watermark is snapshotted ONCE by the orchestrator (a single `lastSyncedAt`
+ * read) so every resource fetches from the same lower bound; no resource re-reads
+ * `lastSyncedAt` or stamps `markSynced` itself. On a full/backfill run `start`
+ * is undefined (no lower bound). The orchestrator owns the single end-of-cycle
+ * `markSynced`.
+ */
+export interface FitbitResourceSyncOptions {
+  fullSync?: boolean;
+  /** The incremental lower bound, snapshotted once by the orchestrator. */
+  start?: Date;
+}
+
 /**
  * Compute the incremental `start` for a resource sync. `fullSync` returns
  * undefined (the backfill walks deep history without a lower bound). Otherwise
@@ -301,7 +315,11 @@ export async function upsertFitbitMeasurements(
   return imported;
 }
 
-/** Stamp `lastSyncedAt = now` after a successful resource sync. */
+/**
+ * Stamp `lastSyncedAt = now`. Called ONCE per cycle by `syncUserFitbit` after a
+ * non-degenerate run — never per resource, so the watermark can't move mid-cycle
+ * and shrink a later resource's fetch window.
+ */
 export async function markSynced(userId: string): Promise<void> {
   await prisma.fitbitConnection.update({
     where: { userId },
@@ -321,6 +339,15 @@ export async function markSynced(userId: string): Promise<void> {
  * Resources: health-metrics (W3), daily-cumulative activity, per-stage sleep,
  * and exercise workouts (W5). Each runs independently so a per-resource 403
  * (a Restricted bundle the user did not grant) soft-skips only that resource.
+ *
+ * WATERMARK: the incremental `start` is snapshotted ONCE here (from the single
+ * `lastSyncedAt` read at the top of the cycle) and threaded to every resource,
+ * and `markSynced` is stamped ONCE at the end. A per-resource read+stamp would
+ * let the first resource move `lastSyncedAt` to now() so later resources only
+ * see the last overlap window — silently dropping the gap after an outage longer
+ * than the overlap. The full/backfill path passes `fullSync: true`, which makes
+ * `incrementalStart` return undefined regardless of the snapshot, so the deep
+ * history walk is unaffected.
  */
 export async function syncUserFitbit(
   userId: string,
@@ -333,6 +360,23 @@ export async function syncUserFitbit(
     return 0;
   }
 
+  // Snapshot the incremental watermark ONCE for the whole cycle. Every resource
+  // sees the same `start`; no resource re-reads `lastSyncedAt` mid-cycle. On a
+  // full/backfill run `incrementalStart` ignores the snapshot and returns
+  // undefined (no lower bound).
+  const connection = await prisma.fitbitConnection.findUnique({
+    where: { userId },
+    select: { lastSyncedAt: true },
+  });
+  if (!connection) return 0;
+  const start = incrementalStart(connection.lastSyncedAt, {
+    fullSync: opts.fullSync,
+  });
+  const resourceOpts: FitbitResourceSyncOptions = {
+    fullSync: opts.fullSync,
+    start,
+  };
+
   const [{ syncUserMetrics }, { syncUserActivity }, { syncUserSleep }, { syncUserWorkout }] =
     await Promise.all([
       import("./sync-metrics"),
@@ -354,7 +398,7 @@ export async function syncUserFitbit(
   await softSkipStorage.run(tracker, async () => {
     for (const fn of resources) {
       try {
-        total += await fn(userId, opts);
+        total += await fn(userId, resourceOpts);
       } catch (err) {
         anyFailed = true;
         getEvent()?.addWarning(`fitbit ${fn.name} failed for ${userId}: ${err}`);
@@ -372,6 +416,10 @@ export async function syncUserFitbit(
   const allSoftSkipped = tracker.count >= resources.length && total === 0;
 
   if (!anyFailed && !allSoftSkipped) {
+    // Stamp the watermark ONCE for the whole cycle, only on a non-degenerate
+    // run. Every resource already saw the snapshot `start`, so stamping now()
+    // here can't shrink a later resource's window.
+    await markSynced(userId);
     await recordSyncSuccess(userId, "fitbit");
   }
 

tests/integration/fitbit-activity-sleep-workout-sync.test.ts

@@ -70,11 +70,15 @@ function stubGoogleHealth(byPathSegment: Record<string, unknown[]>) {
   );
 }
 
-const DAY = { year: 2026, month: 5, day: 10 };
+// INTERVAL data type: the daily total is bucketed into an `interval` carrying a
+// physical `start_time` (the day's start instant), NOT a bare civil `date`.
+const DAY_INTERVAL = { start_time: "2026-05-10T00:00:00.000Z" };
 
 describe("syncUserActivity — cumulative overwrite", () => {
   it("mints a stats: externalId and overwrites the same day on re-sync (no duplicate)", async () => {
-    stubGoogleHealth({ steps: [{ steps: { count: 8000, date: DAY } }] });
+    stubGoogleHealth({
+      steps: [{ steps: { count: 8000, interval: DAY_INTERVAL } }],
+    });
     const { syncUserActivity } = await import("@/lib/fitbit/sync-activity");
     const first = await syncUserActivity(TEST_USER_ID);
     expect(first).toBe(1);
@@ -90,7 +94,9 @@ describe("syncUserActivity — cumulative overwrite", () => {
     expect(rows[0]!.externalId).toBe("stats:steps:2026-05-10");
 
     // Re-fetch the same day with a corrected total → overwrite, not duplicate.
-    stubGoogleHealth({ steps: [{ steps: { count: 9123, date: DAY } }] });
+    stubGoogleHealth({
+      steps: [{ steps: { count: 9123, interval: DAY_INTERVAL } }],
+    });
     await syncUserActivity(TEST_USER_ID);
 
     rows = await prisma.measurement.findMany({
@@ -102,7 +108,9 @@ describe("syncUserActivity — cumulative overwrite", () => {
   });
 
   it("preserves a 0-step rest day rather than dropping it as a gap", async () => {
-    stubGoogleHealth({ steps: [{ steps: { count: 0, date: DAY } }] });
+    stubGoogleHealth({
+      steps: [{ steps: { count: 0, interval: DAY_INTERVAL } }],
+    });
     const { syncUserActivity } = await import("@/lib/fitbit/sync-activity");
     const imported = await syncUserActivity(TEST_USER_ID);
     expect(imported).toBe(1);