Merge branch 'fix/v1439-jobs' into release/v1.43.9

# Conflicts:
#	src/lib/integrations/oauth-refresh.ts

src/lib/ai/__tests__/codex-oauth.test.ts

@@ -225,6 +225,23 @@ describe("codex-oauth", () => {
       expect(err.oauthError).toBeNull();
     });
 
+    it("does not read a bare 403 as a dead sign-in", async () => {
+      // An edge challenge in front of the issuer answers a valid grant so.
+      failWith(403, "<html>challenge</html>");
+      const err = await refreshError();
+      expect(err.credentialExpired).toBe(false);
+      expect(err.httpStatus).toBe(403);
+    });
+
+    it("does not read refresh_token_reused as a dead sign-in", async () => {
+      // The loser of two refreshes of one token hears this; the winner's
+      // rotated token is alive.
+      failWith(400, JSON.stringify({ error: "refresh_token_reused" }));
+      const err = await refreshError();
+      expect(err.credentialExpired).toBe(false);
+      expect(err.oauthError).toBe("refresh_token_reused");
+    });
+
     it("never quotes the response body in the message", async () => {
       failWith(
         401,

src/lib/ai/codex-oauth.ts

@@ -296,14 +296,18 @@ export async function refreshDeviceTokens(
 
 /**
  * OAuth error codes on the refresh grant that mean the stored refresh token
- * is dead for good: revoked, expired, already rotated, or issued to another
- * client. Retrying cannot help; only signing in again can.
+ * is dead for good: revoked, expired, or issued to another client. Retrying
+ * cannot help; only signing in again can.
+ *
+ * `refresh_token_reused` is deliberately not among them. It is what the loser
+ * of two refreshes of one token hears while the winner's rotated token is
+ * perfectly alive, so it says nothing about the sign-in: the refresh re-reads
+ * the row and carries on with the newer token (`refreshCodexCredential`).
  */
 const DEAD_REFRESH_TOKEN_CODES: ReadonlySet<string> = new Set([
   "invalid_grant",
   "invalid_refresh_token",
   "refresh_token_expired",
-  "refresh_token_reused",
   "refresh_token_invalidated",
   "invalid_token",
 ]);
@@ -333,9 +337,12 @@ export class CodexRefreshError extends Error {
   readonly oauthError: string | null;
 
   constructor(status: number, oauthError: string | null) {
+    // A bare 403 is not proof: an edge challenge in front of the issuer
+    // answers 403 to a perfectly valid grant. It keeps its own status and
+    // with it the ledger's timed auth cooldown; only a 401 or the issuer's
+    // own dead-grant code parks the sign-in until the person reconnects.
     const credentialExpired =
       status === 401 ||
-      status === 403 ||
       (oauthError !== null && DEAD_REFRESH_TOKEN_CODES.has(oauthError));
     super(
       credentialExpired
@@ -345,7 +352,7 @@ export class CodexRefreshError extends Error {
     this.name = "CodexRefreshError";
     this.credentialExpired = credentialExpired;
     this.oauthError = oauthError;
-    this.httpStatus = credentialExpired && status !== 403 ? 401 : status;
+    this.httpStatus = credentialExpired ? 401 : status;
   }
 }
 

src/lib/ai/provider.ts

@@ -14,6 +14,7 @@ import {
   decryptCodexCreds,
   encryptAdminCodexCreds,
   decryptAdminCodexCreds,
+  type CodexCreds,
 } from "./codex-oauth";
 import { isPublicUrl } from "@/lib/validations/notifications";
 import { isLocalAiHostAllowed } from "./local-host-allowlist";
@@ -31,6 +32,11 @@ import {
 } from "@/lib/sharing/provider-work-authority";
 import { memoizePerRequest } from "@/lib/request-cache";
 import { emitSignal } from "@/lib/logging/signal";
+import {
+  acquireProviderTokenRefreshLock,
+  PROVIDER_REFRESH_TRANSACTION_OPTIONS,
+} from "@/lib/integrations/oauth-refresh";
+import type { Prisma } from "@/generated/prisma/client";
 import type { AiModality } from "@/lib/ai/capabilities/types";
 import type {
   ProviderEntryPresence,
@@ -62,72 +68,253 @@ function oauthErrorOf(error: unknown): string | null {
   return typeof code === "string" ? code : null;
 }
 
+type CodexCredentialOwner = "codex" | "admin-codex";
+
 /**
- * Park a Codex connection whose sign-in the issuer has refused for good.
- *
- * Retrying a dead refresh token cannot succeed; until this existed every
- * generation of every card of every run asked the issuer again, failed with
- * the same 401, and logged it. Parking flips the connection to the existing
- * `expired` state — the one Settings already renders as "reconnect ChatGPT" —
- * so the chain, the presence probe and with it every AI capability stop
- * counting it, and nothing is sent until the person signs in again (the
- * device-code flow writes `connected`).
- *
- * Compare-and-set on the refresh token that was refused: a reconnect that
- * landed in between wrote a different one and is left alone, and of several
- * workers failing at once exactly one flips the row, so the transition is
- * signalled once and never again per run.
+ * One Codex credential row as a refresh sees it: the user's own connection or
+ * the operator's central one. `writeRotated` and `park` are compare-and-set on
+ * the refresh token this refresh spent, so a sign-in that landed meanwhile (the
+ * device-code flow writes without the refresh lock) is never overwritten.
  */
-async function parkUserCodex(
-  userId: string,
-  refusedRefreshEncrypted: string,
-  error: unknown,
-): Promise<void> {
-  try {
-    const parked = await prisma.user.updateMany({
-      where: {
-        id: userId,
-        codexConnectionStatus: "connected",
-        codexRefreshTokenEncrypted: refusedRefreshEncrypted,
-      },
-      data: { codexConnectionStatus: "expired" },
-    });
-    if (parked.count === 1) {
-      emitSignal({
-        action: "ai.credential.expired",
-        level: "warn",
-        meta: { provider: "codex", oauth_error: oauthErrorOf(error) },
-      });
-    }
-  } catch {
-    // Best effort: an unparked connection is retried exactly as before.
-  }
+interface CodexCredentialStore {
+  owner: CodexCredentialOwner;
+  lockId: string;
+  read(tx: Prisma.TransactionClient): Promise<{
+    status: string | null;
+    refreshEncrypted: string | null;
+    creds: CodexCreds | null;
+  } | null>;
+  writeRotated(
+    tx: Prisma.TransactionClient,
+    spentRefreshEncrypted: string,
+    fresh: CodexCreds,
+  ): Promise<number>;
+  park(
+    tx: Prisma.TransactionClient,
+    spentRefreshEncrypted: string,
+  ): Promise<number>;
 }
 
-/** The operator's central Codex, parked the same way. */
-async function parkCentralCodex(
-  refusedRefreshEncrypted: string,
-  error: unknown,
-): Promise<void> {
-  try {
-    const parked = await prisma.appSettings.updateMany({
+function userCodexStore(userId: string): CodexCredentialStore {
+  return {
+    owner: "codex",
+    lockId: userId,
+    async read(tx) {
+      const row = await tx.user.findUnique({
+        where: { id: userId },
+        select: {
+          codexConnectionStatus: true,
+          codexAccessTokenEncrypted: true,
+          codexRefreshTokenEncrypted: true,
+        },
+      });
+      if (!row) return null;
+      return {
+        status: row.codexConnectionStatus,
+        refreshEncrypted: row.codexRefreshTokenEncrypted,
+        creds:
+          row.codexAccessTokenEncrypted && row.codexRefreshTokenEncrypted
+            ? decryptCodexCreds({
+                accessEncrypted: row.codexAccessTokenEncrypted,
+                refreshEncrypted: row.codexRefreshTokenEncrypted,
+              })
+            : null,
+      };
+    },
+    async writeRotated(tx, spent, fresh) {
+      const enc = encryptCodexCreds(fresh);
+      const written = await tx.user.updateMany({
+        where: { id: userId, codexRefreshTokenEncrypted: spent },
+        data: {
+          codexAccessTokenEncrypted: enc.accessEncrypted,
+          codexRefreshTokenEncrypted: enc.refreshEncrypted,
+          codexTokenExpiresAt: fresh.expiresAt,
+          codexConnectionStatus: "connected",
+        },
+      });
+      return written.count;
+    },
+    async park(tx, spent) {
+      const parked = await tx.user.updateMany({
+        where: {
+          id: userId,
+          codexConnectionStatus: "connected",
+          codexRefreshTokenEncrypted: spent,
+        },
+        data: { codexConnectionStatus: "expired" },
+      });
+      return parked.count;
+    },
+  };
+}
+
+const centralCodexStore: CodexCredentialStore = {
+  owner: "admin-codex",
+  lockId: "singleton",
+  async read(tx) {
+    const row = await tx.appSettings.findUnique({
+      where: { id: "singleton" },
+      select: {
+        adminCodexConnectionStatus: true,
+        adminCodexAccessTokenEncrypted: true,
+        adminCodexRefreshTokenEncrypted: true,
+        adminCodexAccountIdEncrypted: true,
+        adminCodexTokenExpiresAt: true,
+      },
+    });
+    if (!row) return null;
+    return {
+      status: row.adminCodexConnectionStatus,
+      refreshEncrypted: row.adminCodexRefreshTokenEncrypted,
+      creds:
+        row.adminCodexAccessTokenEncrypted &&
+        row.adminCodexRefreshTokenEncrypted &&
+        row.adminCodexAccountIdEncrypted
+          ? decryptAdminCodexCreds({
+              accessEncrypted: row.adminCodexAccessTokenEncrypted,
+              refreshEncrypted: row.adminCodexRefreshTokenEncrypted,
+              accountIdEncrypted: row.adminCodexAccountIdEncrypted,
+              expiresAt: row.adminCodexTokenExpiresAt,
+            })
+          : null,
+    };
+  },
+  async writeRotated(tx, spent, fresh) {
+    const enc = encryptAdminCodexCreds(fresh);
+    const written = await tx.appSettings.updateMany({
+      where: { id: "singleton", adminCodexRefreshTokenEncrypted: spent },
+      data: {
+        adminCodexAccessTokenEncrypted: enc.accessEncrypted,
+        adminCodexRefreshTokenEncrypted: enc.refreshEncrypted,
+        adminCodexAccountIdEncrypted: enc.accountIdEncrypted,
+        adminCodexTokenExpiresAt: enc.expiresAt,
+        adminCodexConnectionStatus: "connected",
+      },
+    });
+    return written.count;
+  },
+  async park(tx, spent) {
+    const parked = await tx.appSettings.updateMany({
       where: {
         id: "singleton",
         adminCodexConnectionStatus: "connected",
-        adminCodexRefreshTokenEncrypted: refusedRefreshEncrypted,
+        adminCodexRefreshTokenEncrypted: spent,
       },
       data: { adminCodexConnectionStatus: "expired" },
     });
-    if (parked.count === 1) {
-      emitSignal({
-        action: "ai.credential.expired",
-        level: "warn",
-        meta: { provider: "admin-codex", oauth_error: oauthErrorOf(error) },
-      });
-    }
-  } catch {
-    // Best effort, as above.
+    return parked.count;
+  },
+};
+
+/** Refreshes in flight in this process, keyed on credential and spent token. */
+const codexRefreshesInFlight = new Map<string, Promise<CodexCreds>>();
+
+/**
+ * Refresh one Codex credential, at most once per spent token.
+ *
+ * The issuer rotates the refresh token on every refresh and honours each one
+ * once. Two callers that refreshed the same token used to race: the loser got
+ * `refresh_token_reused`, read it as a dead sign-in and parked the connection
+ * while the winner's fresh tokens landed beside an `expired` status. So:
+ *
+ *   - callers in one process share one refresh (single flight), and callers
+ *     in different processes queue on the per-credential advisory lock the
+ *     provider syncs use (`acquireProviderTokenRefreshLock`);
+ *   - inside the lock the row is read again, and a token a peer already
+ *     rotated is used as it is, without asking the issuer;
+ *   - the rotated tokens are written compare-and-set on the spent token, and
+ *     the write sets `connected`;
+ *   - after a refusal the row is read again too, and only a refusal of the
+ *     token still stored, with the issuer's dead-grant answer, parks it
+ *     (`expired`, the state Settings renders as "reconnect ChatGPT"). The
+ *     transition is signalled once: only the lock holder can make it.
+ *
+ * Parking stops the chain, the presence probe and every AI capability from
+ * counting the connection until the person signs in again, which is what
+ * stopped every run from asking the issuer about the same dead token.
+ */
+function refreshCodexCredential(
+  store: CodexCredentialStore,
+  spentRefreshEncrypted: string,
+): Promise<CodexCreds> {
+  const key = `${store.owner}:${store.lockId}:${spentRefreshEncrypted}`;
+  const inFlight = codexRefreshesInFlight.get(key);
+  if (inFlight) return inFlight;
+  const run = runCodexRefresh(store, spentRefreshEncrypted).finally(() => {
+    codexRefreshesInFlight.delete(key);
+  });
+  codexRefreshesInFlight.set(key, run);
+  return run;
+}
+
+async function runCodexRefresh(
+  store: CodexCredentialStore,
+  spent: string,
+): Promise<CodexCreds> {
+  const outcome = await prisma.$transaction(
+    async (
+      tx,
+    ): Promise<
+      | { fresh: CodexCreds }
+      | { refused: unknown; parked: boolean }
+      | { failed: unknown }
+    > => {
+      await acquireProviderTokenRefreshLock(tx, store.owner, store.lockId);
+      const row = await store.read(tx);
+      if (!row || row.status !== "connected" || !row.refreshEncrypted) {
+        return {
+          refused: codexReconnectRequired("no refresh token"),
+          parked: false,
+        };
+      }
+      if (row.refreshEncrypted !== spent && row.creds) {
+        // A peer rotated while this caller waited: its token is the live one.
+        return { fresh: row.creds };
+      }
+      if (!row.creds) {
+        const refused = codexReconnectRequired("token storage corrupt");
+        return {
+          refused,
+          parked: (await store.park(tx, row.refreshEncrypted)) === 1,
+        };
+      }
+      let fresh: CodexCreds;
+      try {
+        fresh = await refreshDeviceTokens(row.creds.refreshToken);
+      } catch (error) {
+        const now = await store.read(tx);
+        if (now?.refreshEncrypted !== row.refreshEncrypted && now?.creds) {
+          // A sign-in replaced the token while the issuer refused the old one.
+          return { fresh: now.creds };
+        }
+        if (!isDeadCodexSignIn(error)) return { failed: error };
+        return {
+          refused: error,
+          parked: (await store.park(tx, row.refreshEncrypted)) === 1,
+        };
+      }
+      if ((await store.writeRotated(tx, row.refreshEncrypted, fresh)) === 0) {
+        // A sign-in replaced the row meanwhile; it wins over this refresh.
+        const now = await store.read(tx);
+        if (now?.creds) return { fresh: now.creds };
+      }
+      return { fresh };
+    },
+    PROVIDER_REFRESH_TRANSACTION_OPTIONS,
+  );
+  if ("fresh" in outcome) return outcome.fresh;
+  if ("failed" in outcome) throw outcome.failed;
+  if (outcome.parked) {
+    emitSignal({
+      action: "ai.credential.expired",
+      level: "warn",
+      meta: {
+        provider: store.owner,
+        oauth_error: oauthErrorOf(outcome.refused),
+      },
+    });
   }
+  throw outcome.refused;
 }
 
 class NoProvider implements AIProvider {
@@ -384,27 +571,16 @@ async function resolveCodexProvider(
   // Proactive refresh if the access token is within 5 min of expiry.
   if (stored.expiresAt.getTime() < Date.now() + TOKEN_REFRESH_BUFFER_MS) {
     try {
-      const fresh = await refreshDeviceTokens(stored.refreshToken);
-      active = fresh;
-
-      const enc = encryptCodexCreds(fresh);
-      await prisma.user.update({
-        where: { id: userId },
-        data: {
-          codexAccessTokenEncrypted: enc.accessEncrypted,
-          codexRefreshTokenEncrypted: enc.refreshEncrypted,
-          codexTokenExpiresAt: fresh.expiresAt,
-        },
-      });
+      active = await refreshCodexCredential(
+        userCodexStore(userId),
+        user.codexRefreshTokenEncrypted,
+      );
     } catch (error) {
-      // A refused grant parks the connection and drops it from this chain:
-      // calling Codex with the expired access token could only 401 into a
-      // second refusal. Anything else falls through — CodexClient will
-      // trigger an on-401 refresh and persist there.
-      if (isDeadCodexSignIn(error)) {
-        await parkUserCodex(userId, user.codexRefreshTokenEncrypted, error);
-        return null;
-      }
+      // A refused grant drops the connection from this chain (the refresh
+      // parked it): calling Codex with the expired access token could only
+      // 401 into a second refusal. Anything else falls through — CodexClient
+      // will trigger an on-401 refresh.
+      if (isDeadCodexSignIn(error)) return null;
     }
   }
 
@@ -412,54 +588,17 @@ async function resolveCodexProvider(
     accessToken: active.accessToken,
     accountId: active.accountId,
     onTokenRefresh: async () => {
-      const freshUser = await prisma.user.findUnique({
+      const current = await prisma.user.findUnique({
         where: { id: userId },
-        select: {
-          codexAccessTokenEncrypted: true,
-          codexRefreshTokenEncrypted: true,
-        },
+        select: { codexRefreshTokenEncrypted: true },
       });
-      if (
-        !freshUser?.codexAccessTokenEncrypted ||
-        !freshUser.codexRefreshTokenEncrypted
-      ) {
+      if (!current?.codexRefreshTokenEncrypted) {
         throw codexReconnectRequired("no refresh token");
       }
-      const decoded = decryptCodexCreds({
-        accessEncrypted: freshUser.codexAccessTokenEncrypted,
-        refreshEncrypted: freshUser.codexRefreshTokenEncrypted,
-      });
-      if (!decoded) {
-        const error = codexReconnectRequired("token storage corrupt");
-        await parkUserCodex(
-          userId,
-          freshUser.codexRefreshTokenEncrypted,
-          error,
-        );
-        throw error;
-      }
-      let fresh;
-      try {
-        fresh = await refreshDeviceTokens(decoded.refreshToken);
-      } catch (error) {
-        if (isDeadCodexSignIn(error)) {
-          await parkUserCodex(
-            userId,
-            freshUser.codexRefreshTokenEncrypted,
-            error,
-          );
-        }
-        throw error;
-      }
-      const enc = encryptCodexCreds(fresh);
-      await prisma.user.update({
-        where: { id: userId },
-        data: {
-          codexAccessTokenEncrypted: enc.accessEncrypted,
-          codexRefreshTokenEncrypted: enc.refreshEncrypted,
-          codexTokenExpiresAt: fresh.expiresAt,
-        },
-      });
+      const fresh = await refreshCodexCredential(
+        userCodexStore(userId),
+        current.codexRefreshTokenEncrypted,
+      );
       return { accessToken: fresh.accessToken, accountId: fresh.accountId };
     },
   });
@@ -521,26 +660,15 @@ async function resolveAdminCodexProvider(): Promise<AIProvider | null> {
 
   if (stored.expiresAt.getTime() < Date.now() + TOKEN_REFRESH_BUFFER_MS) {
     try {
-      const fresh = await refreshDeviceTokens(stored.refreshToken);
-      active = fresh;
-      const enc = encryptAdminCodexCreds(fresh);
-      await prisma.appSettings.update({
-        where: { id: "singleton" },
-        data: {
-          adminCodexAccessTokenEncrypted: enc.accessEncrypted,
-          adminCodexRefreshTokenEncrypted: enc.refreshEncrypted,
-          adminCodexAccountIdEncrypted: enc.accountIdEncrypted,
-          adminCodexTokenExpiresAt: enc.expiresAt,
-        },
-      });
+      active = await refreshCodexCredential(
+        centralCodexStore,
+        settings.adminCodexRefreshTokenEncrypted,
+      );
     } catch (error) {
-      // A refused grant parks the central connection (see `parkUserCodex`);
-      // anything else falls through — CodexClient triggers an on-401 refresh
-      // and persists via the callback below.
-      if (isDeadCodexSignIn(error)) {
-        await parkCentralCodex(settings.adminCodexRefreshTokenEncrypted, error);
-        return null;
-      }
+      // A refused grant parked the central connection (see
+      // `refreshCodexCredential`); anything else falls through — CodexClient
+      // triggers an on-401 refresh via the callback below.
+      if (isDeadCodexSignIn(error)) return null;
     }
   }
 
@@ -548,52 +676,17 @@ async function resolveAdminCodexProvider(): Promise<AIProvider | null> {
     accessToken: active.accessToken,
     accountId: active.accountId,
     onTokenRefresh: async () => {
-      const fresh = await prisma.appSettings.findUnique({
+      const current = await prisma.appSettings.findUnique({
         where: { id: "singleton" },
-        select: {
-          adminCodexAccessTokenEncrypted: true,
-          adminCodexRefreshTokenEncrypted: true,
-          adminCodexAccountIdEncrypted: true,
-          adminCodexTokenExpiresAt: true,
-        },
+        select: { adminCodexRefreshTokenEncrypted: true },
       });
-      if (
-        !fresh?.adminCodexAccessTokenEncrypted ||
-        !fresh.adminCodexRefreshTokenEncrypted ||
-        !fresh.adminCodexAccountIdEncrypted
-      ) {
+      if (!current?.adminCodexRefreshTokenEncrypted) {
         throw codexReconnectRequired("no central refresh token");
       }
-      const decoded = decryptAdminCodexCreds({
-        accessEncrypted: fresh.adminCodexAccessTokenEncrypted,
-        refreshEncrypted: fresh.adminCodexRefreshTokenEncrypted,
-        accountIdEncrypted: fresh.adminCodexAccountIdEncrypted,
-        expiresAt: fresh.adminCodexTokenExpiresAt,
-      });
-      if (!decoded) {
-        const error = codexReconnectRequired("central token storage corrupt");
-        await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error);
-        throw error;
-      }
-      let refreshed;
-      try {
-        refreshed = await refreshDeviceTokens(decoded.refreshToken);
-      } catch (error) {
-        if (isDeadCodexSignIn(error)) {
-          await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error);
-        }
-        throw error;
-      }
-      const enc = encryptAdminCodexCreds(refreshed);
-      await prisma.appSettings.update({
-        where: { id: "singleton" },
-        data: {
-          adminCodexAccessTokenEncrypted: enc.accessEncrypted,
-          adminCodexRefreshTokenEncrypted: enc.refreshEncrypted,
-          adminCodexAccountIdEncrypted: enc.accountIdEncrypted,
-          adminCodexTokenExpiresAt: enc.expiresAt,
-        },
-      });
+      const refreshed = await refreshCodexCredential(
+        centralCodexStore,
+        current.adminCodexRefreshTokenEncrypted,
+      );
       return {
         accessToken: refreshed.accessToken,
         accountId: refreshed.accountId,

src/lib/integrations/oauth-refresh.ts

@@ -61,7 +61,8 @@ export async function persistRotatedToken(
   return opts.readPeerAccessToken();
 }
 
-export type SerializedRefreshProvider = "whoop" | "withings" | "oura";
+export type SerializedRefreshProvider =
+  "whoop" | "withings" | "oura" | "codex" | "admin-codex";
 
 /**
  * Provider refreshes hold a database transaction open around one bounded

src/lib/jobs/queue-runtime.ts

@@ -190,7 +190,9 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
   "daily-briefing": short("A light cohort scan every fifteen minutes."),
 
   // ── Rollups ────────────────────────────────────────────────────────
-  "rollup-recompute": short("One bucket."),
+  "rollup-recompute": short(
+    "One type at one granularity over one bucket or a run of adjacent buckets, inside the five-year fold window.",
+  ),
   "rollup-full-backfill": {
     runtime: "long",
     expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS,

src/lib/rollups/measurement-rollups.ts

@@ -362,8 +362,19 @@ export async function recomputeBucketsForMeasurement(
   );
 }
 
+/**
+ * UTC wall-clock literals for a `timestamp[]` parameter. `measured_at` and
+ * `bucket_start` are naive UTC timestamps, so the instants they are compared
+ * with must be too: a `timestamptz[]` parameter is converted through the
+ * session time zone, and in any zone but UTC neither the day filter nor the
+ * keyed delete matched a single day, so a refold left its buckets stale.
+ */
+function utcStamps(dates: readonly Date[]): string[] {
+  return dates.map((d) => d.toISOString().replace("Z", ""));
+}
+
 /** Keys one set-based DAY refold carries; bounds the statement's arrays. */
-const KEYED_REFOLD_CHUNK = 2_000;
+const KEYED_REFOLD_CHUNK = 500;
 
 /**
  * The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write
@@ -381,7 +392,17 @@ export async function recomputeBucketsForKeys(
   userId: string,
   rows: Array<{ type: MeasurementType; measuredAt: Date }>,
 ): Promise<void> {
-  const keys = collapseToTypeDayKeys(rows);
+  // Sorted on (type, day), the order every upsert of the tier takes its row
+  // locks in (`persistRollupRows` sorts the same way), and chunked small, so
+  // a keyed refold racing the full fold or another refold holds a short,
+  // deterministic slice and waits rather than interleaving lock orders.
+  const keys = collapseToTypeDayKeys(rows).sort((a, b) =>
+    a.type !== b.type
+      ? a.type < b.type
+        ? -1
+        : 1
+      : a.measuredAt.getTime() - b.measuredAt.getTime(),
+  );
   for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) {
     await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK));
   }
@@ -441,11 +462,11 @@ async function refoldDayBuckets(
             AND granularity = 'DAY'::"measurement_rollup_granularity"
             AND (type, bucket_start) IN (
               SELECT k.t::"measurement_type", k.b
-                FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b)
+                FROM unnest($2::text[], $3::timestamp[]) AS k(t, b)
             )`,
         userId,
         keys.map((k) => k.type),
-        keys.map((k) => k.measuredAt.toISOString()),
+        utcStamps(keys.map((k) => k.measuredAt)),
       );
       for (let i = 0; i < ordered.length; i += 500) {
         const { sql, params } = buildRollupUpsert(
@@ -642,9 +663,9 @@ async function runRollupAggregate(
   if (input.types && input.types.length === 0) return [];
   // Bound as the fourth parameter, never spliced.
   const dayFilter = input.days
-    ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])`
+    ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])`
     : "";
-  const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : [];
+  const dayParams = input.days ? [utcStamps(input.days)] : [];
 
   // Postgres rejects `date_trunc($1::text, m."measured_at")` in a
   // GROUP BY because the planner sees the parameterised expression
@@ -857,7 +878,7 @@ async function applyHourlyMeanDay(
           AND m."deleted_at" IS NULL
           ${
             input.days
-              ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])`
+              ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])`
               : ""
           }
         GROUP BY m."type", m."source", date_trunc('hour', m."measured_at")
@@ -869,7 +890,7 @@ async function applyHourlyMeanDay(
     input.userId,
     input.from,
     input.to,
-    ...(input.days ? [input.days.map((d) => d.toISOString())] : []),
+    ...(input.days ? [utcStamps(input.days)] : []),
   );
   const byKey = new Map(
     means.map((m) => [

tests/integration/codex-credential-park.test.ts

@@ -261,3 +261,61 @@ describe("a refused Codex sign-in", () => {
     expect(vi.mocked(refreshDeviceTokens)).toHaveBeenCalledTimes(1);
   });
 });
+
+describe("two refreshes of one Codex token", () => {
+  async function codexStatus(userId: string) {
+    return (
+      await prisma.user.findUniqueOrThrow({
+        where: { id: userId },
+        select: { codexConnectionStatus: true },
+      })
+    ).codexConnectionStatus;
+  }
+
+  it("share one refresh: the slower caller never parks the winner's sign-in", async () => {
+    const user = await seedCodexUser();
+    // The issuer honours each refresh token once. The first caller's refresh
+    // takes a moment; a second refresh of the same token is refused as reused.
+    let asked = 0;
+    vi.mocked(refreshDeviceTokens).mockImplementation(async () => {
+      asked += 1;
+      if (asked > 1) throw new CodexRefreshError(400, "refresh_token_reused");
+      await new Promise((resolve) => setTimeout(resolve, 200));
+      return creds("rotated", 10 * HOUR_MS);
+    });
+
+    const [first, second] = await Promise.all([
+      resolveProviderChain(user.id),
+      resolveProviderChain(user.id),
+    ]);
+
+    expect(first.map((p) => p.providerType)).toContain("codex");
+    expect(second.map((p) => p.providerType)).toContain("codex");
+    expect(asked).toBe(1);
+    expect(await codexStatus(user.id)).toBe("connected");
+    expect(parkSignals("codex")).toHaveLength(0);
+  });
+
+  it("carries on with the token a peer rotated while the issuer refused the old one", async () => {
+    const user = await seedCodexUser();
+    // Another process won the race: by the time the issuer answers this
+    // caller "reused", the peer's rotated tokens are in the row.
+    vi.mocked(refreshDeviceTokens).mockImplementation(async () => {
+      const peer = encryptCodexCreds(creds("peer", 10 * HOUR_MS));
+      await prisma.user.update({
+        where: { id: user.id },
+        data: {
+          codexAccessTokenEncrypted: peer.accessEncrypted,
+          codexRefreshTokenEncrypted: peer.refreshEncrypted,
+        },
+      });
+      throw new CodexRefreshError(400, "refresh_token_reused");
+    });
+
+    const chain = await resolveProviderChain(user.id);
+
+    expect(chain.map((p) => p.providerType)).toContain("codex");
+    expect(await codexStatus(user.id)).toBe("connected");
+    expect(parkSignals("codex")).toHaveLength(0);
+  });
+});

tests/integration/rollup-keyed-refold-session-tz.test.ts

@@ -0,0 +1,76 @@
+/**
+ * The keyed DAY refold does not depend on the session time zone.
+ *
+ * `measured_at` and `bucket_start` are naive UTC timestamps. The refold's day
+ * filter and its delete compared them with a `timestamptz[]` parameter, which
+ * Postgres converts through the session time zone: outside UTC neither matched
+ * a single day, so the refold silently left every touched bucket stale. The
+ * app pins its sessions to UTC, so this runs the pool in another zone on
+ * purpose.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/db-session", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/lib/db-session")>();
+  return { ...actual, TIME_ZONE_OPTION: "-c TimeZone=Pacific/Auckland" };
+});
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import { prisma } from "@/lib/db";
+import {
+  recomputeBucketsForKeys,
+  recomputeBucketsForMeasurement,
+} from "@/lib/rollups/measurement-rollups";
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+describe("keyed DAY refold in a non-UTC session", () => {
+  it("folds a new reading into its day", async () => {
+    const [{ zone }] = await prisma.$queryRaw<Array<{ zone: string }>>`
+      SELECT current_setting('TimeZone') AS zone
+    `;
+    expect(zone).toBe("Pacific/Auckland");
+
+    const user = await getPrismaClient().user.create({
+      data: { username: "tz-refold", email: "tz-refold@example.test" },
+    });
+    const day = new Date("2026-05-10T00:00:00.000Z");
+    await getPrismaClient().measurement.createMany({
+      data: [7, 19].map((hour, i) => ({
+        userId: user.id,
+        type: "WEIGHT" as const,
+        value: 80 + i,
+        unit: "kg",
+        measuredAt: new Date(day.getTime() + hour * 3_600_000),
+      })),
+    });
+    await recomputeBucketsForMeasurement(user.id, "WEIGHT", day);
+    // A third reading lands; the refold has to fold it in.
+    await getPrismaClient().measurement.create({
+      data: {
+        userId: user.id,
+        type: "WEIGHT",
+        value: 82,
+        unit: "kg",
+        measuredAt: new Date(day.getTime() + 21 * 3_600_000),
+      },
+    });
+
+    await recomputeBucketsForKeys(user.id, [
+      { type: "WEIGHT", measuredAt: new Date(day.getTime() + 7 * 3_600_000) },
+    ]);
+
+    const rows = await getPrismaClient().measurementRollup.findMany({
+      where: { userId: user.id, granularity: "DAY" },
+    });
+    expect(rows).toHaveLength(1);
+    expect(rows[0]!.bucketStart.toISOString()).toBe(day.toISOString());
+    expect(rows[0]!.count).toBe(3);
+  });
+});