test(arrivals): pin the spine's guardrails against heavy fixtures

Backfill immunity is proved by replay, not by a hand-picked case: a ten-year
Apple Health export (12772 writes) and a 30-day provider re-sync (58 writes)
must both emit zero events, while a fresh sleep night, workout and first
weight of the day emit exactly one each.

The provider-isolation guard walks the worker's static import graph and fails
if any provider client or resolver becomes reachable. It carries two positive
controls, so a broken entry path cannot make it pass against an empty graph.

The integration test covers what unit mocks cannot: the unique index, the
ON CONFLICT DO NOTHING claim, the losing side of ten concurrent claims, the
forward-only marker, ciphertext at rest, the delete cascade, and the
retention window.

src/lib/arrivals/__tests__/backfill-immunity.test.ts

@@ -0,0 +1,461 @@
+/**
+ * The acceptance gate for the arrival spine.
+ *
+ * The spine sits in front of every ingest path in the product, so its one
+ * existential failure mode is a backfill storm: a mass import turning into
+ * thousands of queued jobs and, downstream, provider calls. This file proves
+ * the opposite against REALISTIC replayed volumes rather than a hand-picked
+ * two-row case — a hand-picked case would pass even if the classifier only
+ * happened to be right at the boundary it was written against.
+ *
+ * Two heavy fixtures:
+ *
+ *   - a multi-year Apple Health export (daily weights, blood-pressure pairs,
+ *     sleep nights and workouts across ten years), replayed as the export
+ *     importer replays it;
+ *   - a 30-day provider re-sync (the shape a WHOOP / Fitbit / Google catch-up
+ *     takes after a token refresh or downtime).
+ *
+ * Both must emit ZERO salient events. Then the same seams, given genuinely
+ * fresh data, must emit EXACTLY ONE each.
+ *
+ * The assertions are on `boss.send` — the actual queue write — not on the
+ * classifier's return value. Testing the classifier alone would prove the
+ * predicate correct while saying nothing about whether the seam consults it.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const sendMock = vi.fn(async () => "job-id");
+const annotateMock = vi.fn();
+let moduleMap: Record<string, boolean> = {};
+let timezone = "Europe/Berlin";
+
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: () => ({ send: sendMock }),
+}));
+
+vi.mock("@/lib/logging/context", () => ({
+  annotate: (...args: unknown[]) => annotateMock(...args),
+}));
+
+vi.mock("@/lib/modules/gate", () => ({
+  resolveModuleMap: async () => moduleMap,
+}));
+
+vi.mock("@/lib/tz/resolver", () => ({
+  resolveUserTimezone: async () => timezone,
+}));
+
+const { emitDataArrival } = await import("../emit-shared");
+
+const USER = "user-fixture";
+
+/** Wall clock the fixtures are replayed against: a Tuesday afternoon. */
+const NOW = new Date("2026-07-14T14:30:00.000Z");
+
+const DAY_MS = 86_400_000;
+
+beforeEach(() => {
+  sendMock.mockClear();
+  annotateMock.mockClear();
+  moduleMap = { sleep: true, workouts: true, labs: true, insights: true };
+  timezone = "Europe/Berlin";
+});
+
+/** Every action name the seam annotated, in order. */
+function annotatedActions(): string[] {
+  return annotateMock.mock.calls
+    .map((call) => {
+      const arg = call[0] as { action?: { name?: string } } | undefined;
+      return arg?.action?.name;
+    })
+    .filter((n): n is string => typeof n === "string");
+}
+
+function skipReasons(): string[] {
+  return annotateMock.mock.calls
+    .map((call) => {
+      const arg = call[0] as { meta?: { reason?: string } } | undefined;
+      return arg?.meta?.reason;
+    })
+    .filter((r): r is string => typeof r === "string");
+}
+
+// ─────────────────────────────────────────────────────────────
+// Fixture builders
+// ─────────────────────────────────────────────────────────────
+
+interface ReplayedWrite {
+  kind: "sleep_night" | "workout" | "weight" | "blood_pressure" | "labs_panel";
+  newestSampleAt: Date;
+  insertedCount: number;
+  refId?: string;
+}
+
+/**
+ * A ten-year Apple Health export, in the shape the importer replays it: a
+ * weight most mornings, a blood-pressure pair most evenings, a sleep night
+ * every night, and a workout roughly every other day. Every row is genuinely
+ * INSERTED (a first import creates everything), which is exactly the case that
+ * would storm if the recency test were missing — `insertedCount` alone cannot
+ * save us here.
+ */
+function appleExportFixture(): ReplayedWrite[] {
+  const writes: ReplayedWrite[] = [];
+  const years = 10;
+  for (let dayOffset = years * 365; dayOffset >= 2; dayOffset--) {
+    const day = new Date(NOW.getTime() - dayOffset * DAY_MS);
+    const at = (hour: number) =>
+      new Date(
+        Date.UTC(
+          day.getUTCFullYear(),
+          day.getUTCMonth(),
+          day.getUTCDate(),
+          hour,
+          0,
+          0,
+        ),
+      );
+
+    writes.push({ kind: "weight", newestSampleAt: at(7), insertedCount: 1 });
+    writes.push({
+      kind: "blood_pressure",
+      newestSampleAt: at(20),
+      insertedCount: 2,
+    });
+    writes.push({
+      kind: "sleep_night",
+      newestSampleAt: at(6),
+      insertedCount: 4,
+    });
+    if (dayOffset % 2 === 0) {
+      writes.push({
+        kind: "workout",
+        newestSampleAt: at(18),
+        insertedCount: 1,
+        refId: `workout-${dayOffset}`,
+      });
+    }
+  }
+  return writes;
+}
+
+/**
+ * A 30-day provider re-sync. Distinct from the export in two ways that matter:
+ * it is recent enough that a naive "is it in the last month" test would let it
+ * through, and it re-posts nights and workouts the record may already hold.
+ */
+function providerResyncFixture(): ReplayedWrite[] {
+  const writes: ReplayedWrite[] = [];
+  for (let dayOffset = 30; dayOffset >= 2; dayOffset--) {
+    const day = new Date(NOW.getTime() - dayOffset * DAY_MS);
+    const at = (hour: number) =>
+      new Date(
+        Date.UTC(
+          day.getUTCFullYear(),
+          day.getUTCMonth(),
+          day.getUTCDate(),
+          hour,
+          0,
+          0,
+        ),
+      );
+    writes.push({
+      kind: "sleep_night",
+      newestSampleAt: at(6),
+      insertedCount: 5,
+    });
+    writes.push({
+      kind: "workout",
+      newestSampleAt: at(17),
+      insertedCount: 1,
+      refId: `whoop-${dayOffset}`,
+    });
+  }
+  return writes;
+}
+
+async function replay(writes: ReplayedWrite[], source: string): Promise<void> {
+  for (const w of writes) {
+    await emitDataArrival({
+      userId: USER,
+      kind: w.kind,
+      newestSampleAt: w.newestSampleAt,
+      insertedCount: w.insertedCount,
+      refId: w.refId,
+      source,
+      now: NOW,
+    });
+  }
+}
+
+// ─────────────────────────────────────────────────────────────
+
+describe("arrival spine — heavy backfill fixtures emit nothing", () => {
+  it("a ten-year Apple Health export produces ZERO salient events", async () => {
+    const writes = appleExportFixture();
+    // The fixture must actually be heavy, or the assertion below is vacuous.
+    expect(writes.length).toBeGreaterThan(10_000);
+
+    await replay(writes, "apple_export");
+
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(annotatedActions().every((n) => n.endsWith(".skipped"))).toBe(true);
+    expect(new Set(skipReasons())).toEqual(new Set(["backfill"]));
+  });
+
+  it("a 30-day provider re-sync produces ZERO salient events", async () => {
+    const writes = providerResyncFixture();
+    expect(writes.length).toBeGreaterThan(50);
+
+    await replay(writes, "whoop");
+
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(new Set(skipReasons())).toEqual(new Set(["backfill"]));
+  });
+
+  it("a re-sync that INSERTS nothing produces zero events, whatever its dates", async () => {
+    // The `stats:` overwrite path: rows are touched, values change, nothing is
+    // new. Dated TODAY, so recency cannot be what saves it — only the
+    // inserted-count test can.
+    for (let i = 0; i < 500; i++) {
+      await emitDataArrival({
+        userId: USER,
+        kind: "weight",
+        newestSampleAt: new Date(NOW.getTime() - 60_000),
+        insertedCount: 0,
+        source: "batch",
+        now: NOW,
+      });
+    }
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(new Set(skipReasons())).toEqual(new Set(["noop"]));
+  });
+
+  it("a far-future sample is refused, not enqueued", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date(NOW.getTime() + 3 * DAY_MS),
+      insertedCount: 1,
+      source: "manual",
+      now: NOW,
+    });
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(skipReasons()).toContain("backfill");
+  });
+
+  it("a future sample still inside TODAY is refused by the skew guard alone", async () => {
+    // Isolates the future-dated guard from the recency guard. A device clock
+    // running a few hours fast produces a sample whose LOCAL DAY is today — so
+    // the recency test happily passes it — but which has not happened yet.
+    // Without a separate skew guard this would enqueue, and the day's reaction
+    // would be claimed by a reading the user has not taken.
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date(NOW.getTime() + 5 * 60 * 60 * 1000),
+      insertedCount: 1,
+      source: "manual",
+      now: NOW,
+    });
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(skipReasons()).toContain("backfill");
+  });
+});
+
+describe("arrival spine — genuinely fresh data emits exactly one event", () => {
+  it("a fresh sleep night emits exactly one", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "sleep_night",
+      // Last night, ending this morning.
+      newestSampleAt: new Date("2026-07-14T05:40:00.000Z"),
+      insertedCount: 6,
+      source: "whoop",
+      now: NOW,
+    });
+    expect(sendMock).toHaveBeenCalledTimes(1);
+    expect(annotatedActions()).toEqual(["arrival.sleep_night.emitted"]);
+  });
+
+  it("a fresh workout emits exactly one", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "workout",
+      newestSampleAt: new Date("2026-07-14T11:00:00.000Z"),
+      insertedCount: 1,
+      refId: "workout-fresh",
+      source: "strava",
+      now: NOW,
+    });
+    expect(sendMock).toHaveBeenCalledTimes(1);
+    expect(annotatedActions()).toEqual(["arrival.workout.emitted"]);
+  });
+
+  it("the first weight of the day emits exactly one", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date("2026-07-14T06:15:00.000Z"),
+      insertedCount: 1,
+      source: "withings",
+      now: NOW,
+    });
+    expect(sendMock).toHaveBeenCalledTimes(1);
+    expect(annotatedActions()).toEqual(["arrival.weight.emitted"]);
+  });
+
+  it("one fresh row buried in a heavy backfill still emits exactly one", async () => {
+    // The case that matters most in production: a first-time import of an
+    // export taken today. Everything historical must stay silent, and the one
+    // genuinely current reading must still land.
+    await replay(appleExportFixture(), "apple_export");
+    expect(sendMock).not.toHaveBeenCalled();
+
+    await emitDataArrival({
+      userId: USER,
+      kind: "workout",
+      newestSampleAt: new Date("2026-07-14T09:00:00.000Z"),
+      insertedCount: 1,
+      refId: "workout-today",
+      source: "apple_export",
+      now: NOW,
+    });
+
+    expect(sendMock).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe("arrival spine — singleton keys", () => {
+  it("day-scoped kinds carry the local date and NO time-slot window", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date("2026-07-14T06:15:00.000Z"),
+      insertedCount: 1,
+      source: "withings",
+      now: NOW,
+    });
+    const opts = sendMock.mock.calls[0][2] as Record<string, unknown>;
+    expect(opts.singletonKey).toBe(`arrival:${USER}:weight:2026-07-14`);
+    // A wall-clock window cannot express a local date; asserting its ABSENCE
+    // is what stops someone reintroducing one.
+    expect(opts.singletonSeconds).toBeUndefined();
+  });
+
+  it("workout is keyed per workout, with the device double-post window", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "workout",
+      newestSampleAt: new Date("2026-07-14T11:00:00.000Z"),
+      insertedCount: 1,
+      refId: "workout-abc",
+      source: "strava",
+      now: NOW,
+    });
+    const opts = sendMock.mock.calls[0][2] as Record<string, unknown>;
+    expect(opts.singletonKey).toBe(`arrival:${USER}:workout:workout-abc`);
+    expect(opts.singletonSeconds).toBe(300);
+  });
+
+  it("two workouts on one day do NOT collapse into one event", async () => {
+    for (const id of ["workout-morning", "workout-evening"]) {
+      await emitDataArrival({
+        userId: USER,
+        kind: "workout",
+        newestSampleAt: new Date("2026-07-14T11:00:00.000Z"),
+        insertedCount: 1,
+        refId: id,
+        source: "strava",
+        now: NOW,
+      });
+    }
+    const keys = sendMock.mock.calls.map(
+      (c) => (c[2] as Record<string, unknown>).singletonKey,
+    );
+    expect(new Set(keys).size).toBe(2);
+  });
+
+  it("the retry policy is the cheap-worker one, not the LLM one", async () => {
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date("2026-07-14T06:15:00.000Z"),
+      insertedCount: 1,
+      source: "withings",
+      now: NOW,
+    });
+    const opts = sendMock.mock.calls[0][2] as Record<string, unknown>;
+    expect(opts.retryLimit).toBe(2);
+    expect(opts.retryBackoff).toBe(true);
+  });
+});
+
+describe("arrival spine — module gating and fault isolation", () => {
+  it("a disabled module emits nothing", async () => {
+    moduleMap = { ...moduleMap, workouts: false };
+    await emitDataArrival({
+      userId: USER,
+      kind: "workout",
+      newestSampleAt: new Date("2026-07-14T11:00:00.000Z"),
+      insertedCount: 1,
+      refId: "workout-x",
+      source: "strava",
+      now: NOW,
+    });
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(skipReasons()).toContain("module_off");
+  });
+
+  it("an enqueue failure never propagates to the caller", async () => {
+    sendMock.mockRejectedValueOnce(new Error("queue is down"));
+    await expect(
+      emitDataArrival({
+        userId: USER,
+        kind: "weight",
+        newestSampleAt: new Date("2026-07-14T06:15:00.000Z"),
+        insertedCount: 1,
+        source: "withings",
+        now: NOW,
+      }),
+    ).resolves.toBeUndefined();
+  });
+});
+
+describe("arrival spine — timezone honesty", () => {
+  it("judges recency in the user's zone, not UTC", async () => {
+    // 2026-07-14T22:30Z is still the 14th in UTC but already the 15th in
+    // Auckland. A sample from 2026-07-13T20:00Z is two local days back there
+    // and must be refused, while UTC arithmetic would call it "yesterday".
+    timezone = "Pacific/Auckland";
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date("2026-07-13T00:00:00.000Z"),
+      insertedCount: 1,
+      source: "withings",
+      now: new Date("2026-07-14T22:30:00.000Z"),
+    });
+    expect(sendMock).not.toHaveBeenCalled();
+    expect(skipReasons()).toContain("backfill");
+  });
+
+  it("files the arrival under the user's local day, not the UTC day", async () => {
+    // 23:30 in Berlin on the 14th is 21:30Z — same UTC day. Chosen the other
+    // way round: 00:30 Berlin on the 15th is 22:30Z on the 14th, so the local
+    // date must read 2026-07-15 while UTC still reads the 14th.
+    timezone = "Europe/Berlin";
+    const now = new Date("2026-07-14T22:30:00.000Z");
+    await emitDataArrival({
+      userId: USER,
+      kind: "weight",
+      newestSampleAt: new Date("2026-07-14T22:00:00.000Z"),
+      insertedCount: 1,
+      source: "withings",
+      now,
+    });
+    const payload = sendMock.mock.calls[0][1] as { localDate: string };
+    expect(payload.localDate).toBe("2026-07-15");
+  });
+});

src/lib/jobs/__tests__/data-arrival-provider-isolation.test.ts

@@ -0,0 +1,156 @@
+/**
+ * Structural guard: the data-arrival worker cannot reach a provider.
+ *
+ * The spine's central cost claim is that it spends nothing. That claim is only
+ * as good as the worker's module graph: the moment anything reachable from
+ * `data-arrival.ts` imports a provider client, a mass ingest becomes a mass of
+ * provider calls, and no amount of runtime gating in the handler would show up
+ * in review as the thing that broke it.
+ *
+ * So the guard is structural, not behavioural. It BFS-walks the static import
+ * graph from the worker and asserts no provider module is reachable — the shape
+ * of `documents/__tests__/fenced-chat-module-graph.test.ts`.
+ *
+ * Two deliberate design points:
+ *
+ *   - The forbidden set is GLOBBED, not hand-listed, for the provider clients.
+ *     A hand-written array silently stops covering a `gemini-client.ts` the day
+ *     someone adds one; a glob covers it on arrival.
+ *   - There is a positive control. A graph walk that silently resolves to
+ *     nothing (a moved file, a changed alias, a typo in the entry path) would
+ *     pass every negative assertion vacuously. The control asserts the walk
+ *     DOES reach the modules the worker genuinely uses, so an empty graph fails
+ *     loudly instead of reporting success.
+ *
+ * Known limit, stated because it bounds what this proves: `IMPORT_RE` sees only
+ * static import specifiers. A dynamic `await import(...)` built from a variable
+ * is invisible to it. That is acceptable here — the ban is on a static import,
+ * and a dynamic provider import in a worker would be a far louder thing to
+ * write — but it is not a proof, it is a tripwire.
+ */
+import { existsSync, readFileSync, readdirSync } from "node:fs";
+import { dirname, resolve } from "node:path";
+
+import { describe, expect, it } from "vitest";
+
+const SRC = resolve(__dirname, "../../..");
+
+const WORKER_ENTRY = resolve(SRC, "lib/jobs/data-arrival.ts");
+const EMIT_ENTRY = resolve(SRC, "lib/arrivals/emit-shared.ts");
+
+/** Non-client provider machinery. Hand-listed because it has no shared suffix. */
+const PROVIDER_MACHINERY = [
+  "lib/ai/provider.ts",
+  "lib/ai/provider-runner.ts",
+  "lib/ai/provider-chain.ts",
+  "lib/ai/provider-health-ledger.ts",
+  "lib/ai/openai-wire.ts",
+  "lib/ai/codex-oauth.ts",
+].map((p) => resolve(SRC, p));
+
+/** Every `*-client.ts` under `src/lib/ai`, discovered rather than enumerated. */
+function providerClients(): string[] {
+  const dir = resolve(SRC, "lib/ai");
+  return readdirSync(dir)
+    .filter((f) => f.endsWith("-client.ts"))
+    .map((f) => resolve(dir, f));
+}
+
+function resolveSpecifier(spec: string, fromFile: string): string | null {
+  let candidate: string;
+  if (spec.startsWith("@/")) {
+    candidate = resolve(SRC, spec.slice(2));
+  } else if (spec.startsWith(".")) {
+    candidate = resolve(dirname(fromFile), spec);
+  } else {
+    return null; // bare package import — outside our graph
+  }
+  for (const ext of [".ts", ".tsx", "/index.ts", "/index.tsx"]) {
+    const p = `${candidate}${ext}`;
+    if (existsSync(p)) return p;
+  }
+  return existsSync(candidate) ? candidate : null;
+}
+
+const IMPORT_RE = /(?:from|import)\s+["']([^"']+)["']/g;
+
+/** BFS the static import graph from `entries`, returning every reachable file. */
+function reachableFrom(entries: string[]): Set<string> {
+  const seen = new Set<string>();
+  const queue = [...entries];
+  while (queue.length > 0) {
+    const file = queue.pop()!;
+    if (seen.has(file)) continue;
+    seen.add(file);
+    let source: string;
+    try {
+      source = readFileSync(file, "utf8");
+    } catch {
+      continue;
+    }
+    for (const match of source.matchAll(IMPORT_RE)) {
+      const resolved = resolveSpecifier(match[1], file);
+      if (resolved && !seen.has(resolved)) queue.push(resolved);
+    }
+  }
+  return seen;
+}
+
+describe("data-arrival spine — provider isolation", () => {
+  it("the worker's module graph reaches no provider client", () => {
+    const reachable = reachableFrom([WORKER_ENTRY]);
+    const clients = providerClients();
+    expect(clients.length).toBeGreaterThan(0); // the glob itself must not be empty
+    for (const forbidden of clients) {
+      expect(
+        reachable.has(forbidden),
+        `The data-arrival worker must NOT be able to reach ${forbidden}. The spine's cost claim is that it spends nothing; a provider import here voids it.`,
+      ).toBe(false);
+    }
+  });
+
+  it("the worker's module graph reaches no provider resolver or wire module", () => {
+    const reachable = reachableFrom([WORKER_ENTRY]);
+    for (const forbidden of PROVIDER_MACHINERY) {
+      if (!existsSync(forbidden)) continue; // tolerate a renamed internal
+      expect(
+        reachable.has(forbidden),
+        `The data-arrival worker must NOT be able to reach ${forbidden}.`,
+      ).toBe(false);
+    }
+  });
+
+  it("the emit seam's module graph reaches no provider client either", () => {
+    // The seams run inside every ingest request. A provider import here would
+    // be worse than in the worker: it would land on the request hot path.
+    const reachable = reachableFrom([EMIT_ENTRY]);
+    for (const forbidden of providerClients()) {
+      expect(
+        reachable.has(forbidden),
+        `The arrival emit seam must NOT be able to reach ${forbidden}.`,
+      ).toBe(false);
+    }
+  });
+
+  it("sanity: the walk DOES reach the worker's own dependencies", () => {
+    // Positive control. Without this, a broken entry path or a changed alias
+    // would make every assertion above pass against an empty graph.
+    const reachable = reachableFrom([WORKER_ENTRY]);
+    expect(reachable.has(resolve(SRC, "lib/arrivals/types.ts"))).toBe(true);
+    expect(reachable.has(resolve(SRC, "lib/arrivals/emit-shared.ts"))).toBe(
+      true,
+    );
+    expect(reachable.has(resolve(SRC, "lib/cache/invalidate.ts"))).toBe(true);
+  });
+
+  it("sanity: the walk CAN see a provider client when one is genuinely imported", () => {
+    // Second positive control, on the detector rather than the graph: point the
+    // walk at a module that really does import a provider and assert it is
+    // found. If this ever goes red, the negative assertions above have stopped
+    // meaning anything.
+    const chainEntry = resolve(SRC, "lib/ai/provider.ts");
+    const reachable = reachableFrom([chainEntry]);
+    const clients = providerClients();
+    expect(clients.some((c) => reachable.has(c))).toBe(true);
+  });
+});

src/lib/jobs/__tests__/data-arrival-queue.test.ts

@@ -0,0 +1,86 @@
+/**
+ * v1.4.37 dead-queue guard for the `data-arrival` queue.
+ *
+ * Four facts have to hold together or the spine is silently inert: the queue
+ * name is provisioned, a handler drains it, the queue carries a policy that
+ * actually indexes `singleton_key`, and the retention job that prunes its rows
+ * is itself registered. Any one of them missing fails open in the worst way —
+ * the emits succeed, the jobs vanish, and nothing anywhere reports a problem.
+ *
+ * Source-text assertions over the registrar modules, matching the shape of the
+ * sibling `*-queue.test.ts` guards.
+ */
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+
+import { describe, expect, it } from "vitest";
+
+const statusRegistrar = readFileSync(
+  join(__dirname, "..", "reminder", "register-status.ts"),
+  "utf8",
+);
+
+const maintenanceSource =
+  readFileSync(
+    join(__dirname, "..", "reminder", "register-maintenance.ts"),
+    "utf8",
+  ) +
+  readFileSync(
+    join(__dirname, "..", "reminder", "cleanup-handlers.ts"),
+    "utf8",
+  );
+
+describe("data-arrival queue wiring", () => {
+  it("provisions the queue in the allQueues list", () => {
+    const allQueues = statusRegistrar.match(
+      /const allQueues\s*=\s*\[([\s\S]*?)\];/,
+    );
+    expect(allQueues).not.toBeNull();
+    expect(allQueues![1]).toMatch(/\bDATA_ARRIVAL_QUEUE\b/);
+  });
+
+  it("binds a boss.work handler that drains it", () => {
+    expect(statusRegistrar).toMatch(
+      /boss\.work[\s\S]{0,200}DATA_ARRIVAL_QUEUE[\s\S]{0,200}handleDataArrival/,
+    );
+  });
+
+  it("carries the exclusive policy its singleton keys depend on", () => {
+    // Under pg-boss's default `standard` policy NO index covers
+    // `singleton_key`, so the day-scoped keys would coalesce nothing at all and
+    // a chatty ingest would queue one job per batch. This assertion is anchored
+    // INSIDE the extracted policy table so an unrelated mention of the queue
+    // name elsewhere in the file cannot satisfy it.
+    const table = statusRegistrar.match(
+      /const queuePolicies: QueuePolicyTable\s*=\s*\{([\s\S]*?)\n\};/,
+    );
+    expect(table).not.toBeNull();
+    expect(table![1]).toMatch(
+      /\[DATA_ARRIVAL_QUEUE\][\s\S]{0,200}policy:\s*"exclusive"/,
+    );
+  });
+
+  it("has no cron schedule — it is send-only, driven by ingest", () => {
+    expect(statusRegistrar).not.toMatch(/\[DATA_ARRIVAL_QUEUE,\s*\w+_CRON\]/);
+  });
+
+  it("registers the retention job that prunes the markers", () => {
+    const allQueues = maintenanceSource.match(
+      /const allQueues\s*=\s*\[([\s\S]*?)\];/,
+    );
+    expect(allQueues).not.toBeNull();
+    expect(allQueues![1]).toMatch(/\bARRIVAL_REACTION_CLEANUP_QUEUE\b/);
+    expect(maintenanceSource).toMatch(
+      /\[ARRIVAL_REACTION_CLEANUP_QUEUE,\s*ARRIVAL_REACTION_CLEANUP_CRON\]/,
+    );
+    expect(maintenanceSource).toMatch(
+      /boss\.work[\s\S]{0,200}ARRIVAL_REACTION_CLEANUP_QUEUE[\s\S]{0,200}handleArrivalReactionCleanup/,
+    );
+  });
+
+  it("the retention handler actually deletes arrival reactions by age", () => {
+    expect(maintenanceSource).toMatch(
+      /handleArrivalReactionCleanup[\s\S]{0,600}arrivalReaction\.deleteMany/,
+    );
+  });
+});

src/lib/jobs/__tests__/data-arrival-worker.test.ts

@@ -0,0 +1,122 @@
+/**
+ * The `data-arrival` worker's contract.
+ *
+ * Two properties carry real risk and are pinned here:
+ *
+ *   - The marker is claimed BEFORE any fan-out. The S4 morning-refresh trigger
+ *     stamps its debounce marker only after a non-failed run, so a persistently
+ *     failing downstream let every subsequent sleep batch re-enqueue — an
+ *     unbounded chain. Claiming up front bounds a downstream failure to the
+ *     retries of one job.
+ *   - A refusal returns `skipped`; it does not throw. pg-boss retries a failed
+ *     job, and retrying against a ceiling that will not move until the local
+ *     day rolls over is a loop.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
+vi.mock("@/lib/logging/background", () => ({
+  withBackgroundEvent: async (
+    _name: string,
+    fn: (evt: { addMeta: () => void }) => Promise<void>,
+  ) => fn({ addMeta: () => {} }),
+}));
+vi.mock("@/lib/cache/invalidate", () => ({
+  invalidateUserDashboardSnapshot: vi.fn(),
+}));
+vi.mock("@/lib/jobs/reminder/shared", () => ({
+  getWorkerPrisma: () => fakePrisma,
+  workerLog: vi.fn(),
+}));
+
+const createMany = vi.fn(async () => ({ count: 1 }));
+const updateMany = vi.fn(async () => ({ count: 1 }));
+const fakePrisma = { arrivalReaction: { createMany, updateMany } };
+
+const { runDataArrival } = await import("../data-arrival");
+
+type Runnable = Parameters<typeof runDataArrival>[1];
+
+function arrival(overrides: Partial<Runnable> = {}): Runnable {
+  return {
+    userId: "user-1",
+    kind: "weight",
+    salience: "salient",
+    localDate: "2026-07-14",
+    occurredAt: "2026-07-14T06:15:00.000Z",
+    count: 1,
+    source: "withings",
+    ...overrides,
+  } as Runnable;
+}
+
+beforeEach(() => {
+  createMany.mockClear().mockResolvedValue({ count: 1 });
+  updateMany.mockClear().mockResolvedValue({ count: 1 });
+});
+
+describe("data-arrival worker", () => {
+  it("claims the day's marker before any fan-out", async () => {
+    const result = await runDataArrival(
+      fakePrisma as never,
+      arrival({ kind: "workout", refId: "w-1" }),
+    );
+    expect(createMany).toHaveBeenCalledTimes(1);
+    expect(createMany.mock.calls[0][0]).toMatchObject({ skipDuplicates: true });
+    expect(result).toMatchObject({ status: "processed", dedup: false });
+  });
+
+  it("a lost claim de-duplicates instead of writing a second line", async () => {
+    createMany.mockResolvedValue({ count: 0 });
+    const result = await runDataArrival(fakePrisma as never, arrival());
+    expect(result).toMatchObject({ status: "processed", dedup: true });
+    if (result.status !== "processed") throw new Error("unreachable");
+    expect(result.actions).not.toContain("line_pending");
+  });
+
+  it("only a fresh claim schedules the day's reaction line", async () => {
+    const first = await runDataArrival(fakePrisma as never, arrival());
+    if (first.status !== "processed") throw new Error("unreachable");
+    expect(first.actions).toContain("line_pending");
+  });
+
+  it("a second workout the same day still fans out, despite the day marker", async () => {
+    // The day-scoped marker can only be claimed once, but two sessions are two
+    // events. Gating the workout fan-out on the claim would silently drop the
+    // second one.
+    createMany.mockResolvedValue({ count: 0 });
+    const result = await runDataArrival(
+      fakePrisma as never,
+      arrival({ kind: "workout", refId: "w-2" }),
+    );
+    if (result.status !== "processed") throw new Error("unreachable");
+    expect(result.actions).toContain("workout_pending_insight");
+  });
+
+  it("moves the marker forward on a later arrival, never backwards", async () => {
+    createMany.mockResolvedValue({ count: 0 });
+    await runDataArrival(fakePrisma as never, arrival());
+    const where = updateMany.mock.calls[0][0] as {
+      where: { occurredAt?: { lt?: Date } };
+    };
+    expect(where.where.occurredAt?.lt).toBeInstanceOf(Date);
+  });
+
+  it("refuses an unknown kind as SKIPPED, not FAILED", async () => {
+    const result = await runDataArrival(
+      fakePrisma as never,
+      arrival({ kind: "not_a_kind" as never }),
+    );
+    // A throw here would earn two pointless retries against a payload that can
+    // never succeed.
+    expect(result).toEqual({ status: "skipped", reason: "unknown_kind" });
+    expect(createMany).not.toHaveBeenCalled();
+  });
+
+  it("a genuine transient fault DOES propagate, so the retry policy applies", async () => {
+    createMany.mockRejectedValue(new Error("connection terminated"));
+    await expect(
+      runDataArrival(fakePrisma as never, arrival()),
+    ).rejects.toThrow("connection terminated");
+  });
+});

tests/integration/arrival-reaction-claim.test.ts

@@ -0,0 +1,230 @@
+/**
+ * The `ArrivalReaction` claim, against real Postgres.
+ *
+ * Unit mocks cannot prove any of this: the once-per-kind-per-day contract is a
+ * unique INDEX, the claim is an ON CONFLICT DO NOTHING whose semantics live in
+ * the database, and the losing side of a concurrent claim is a race only a real
+ * connection pool can produce. A mocked `createMany` returning `{count: 1}`
+ * would happily report a successful claim for every caller.
+ */
+import { beforeEach, describe, expect, it } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+async function createUser(username: string) {
+  return getPrismaClient().user.create({
+    data: { username, email: `${username}@example.test`, role: "USER" },
+  });
+}
+
+const LOCAL_DATE = "2026-07-14";
+
+function marker(userId: string, kind: string, occurredAt: Date) {
+  return {
+    userId,
+    kind,
+    localDate: LOCAL_DATE,
+    occurredAt,
+    refId: null,
+  };
+}
+
+describe("ArrivalReaction — the day claim", () => {
+  it("admits the first claim and refuses the second for the same kind and day", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-claim");
+
+    const first = await prisma.arrivalReaction.createMany({
+      data: [marker(user.id, "weight", new Date("2026-07-14T06:00:00Z"))],
+      skipDuplicates: true,
+    });
+    const second = await prisma.arrivalReaction.createMany({
+      data: [marker(user.id, "weight", new Date("2026-07-14T18:00:00Z"))],
+      skipDuplicates: true,
+    });
+
+    expect(first.count).toBe(1);
+    // The second weigh-in of the day does not get its own line. This IS the
+    // throttle — there is no code path to a second generation.
+    expect(second.count).toBe(0);
+    expect(await prisma.arrivalReaction.count()).toBe(1);
+  });
+
+  it("keeps kinds independent within one day", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-kinds");
+    const at = new Date("2026-07-14T06:00:00Z");
+
+    for (const kind of ["weight", "sleep_night", "blood_pressure"]) {
+      const res = await prisma.arrivalReaction.createMany({
+        data: [marker(user.id, kind, at)],
+        skipDuplicates: true,
+      });
+      expect(res.count).toBe(1);
+    }
+    expect(await prisma.arrivalReaction.count()).toBe(3);
+  });
+
+  it("keeps users independent — one user's claim never blocks another's", async () => {
+    const prisma = getPrismaClient();
+    const a = await createUser("arrival-tenant-a");
+    const b = await createUser("arrival-tenant-b");
+    const at = new Date("2026-07-14T06:00:00Z");
+
+    const first = await prisma.arrivalReaction.createMany({
+      data: [marker(a.id, "weight", at)],
+      skipDuplicates: true,
+    });
+    const second = await prisma.arrivalReaction.createMany({
+      data: [marker(b.id, "weight", at)],
+      skipDuplicates: true,
+    });
+
+    expect(first.count).toBe(1);
+    expect(second.count).toBe(1);
+  });
+
+  it("exactly one of many concurrent claims wins", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-race");
+    const at = new Date("2026-07-14T06:00:00Z");
+
+    // Ten seams firing at once, the shape a device batch storm takes.
+    const results = await Promise.all(
+      Array.from({ length: 10 }, () =>
+        prisma.arrivalReaction.createMany({
+          data: [marker(user.id, "workout", at)],
+          skipDuplicates: true,
+        }),
+      ),
+    );
+
+    const winners = results.filter((r) => r.count === 1);
+    expect(winners).toHaveLength(1);
+    expect(await prisma.arrivalReaction.count()).toBe(1);
+  });
+
+  it("the marker moves forward on a later arrival and never backwards", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-forward");
+    const early = new Date("2026-07-14T06:00:00Z");
+    const late = new Date("2026-07-14T18:00:00Z");
+
+    await prisma.arrivalReaction.createMany({
+      data: [marker(user.id, "weight", early)],
+      skipDuplicates: true,
+    });
+
+    const forward = await prisma.arrivalReaction.updateMany({
+      where: {
+        userId: user.id,
+        kind: "weight",
+        localDate: LOCAL_DATE,
+        occurredAt: { lt: late },
+      },
+      data: { occurredAt: late },
+    });
+    expect(forward.count).toBe(1);
+
+    const backward = await prisma.arrivalReaction.updateMany({
+      where: {
+        userId: user.id,
+        kind: "weight",
+        localDate: LOCAL_DATE,
+        occurredAt: { lt: early },
+      },
+      data: { occurredAt: early },
+    });
+    expect(backward.count).toBe(0);
+
+    const row = await prisma.arrivalReaction.findFirst({
+      where: { userId: user.id },
+    });
+    expect(row?.occurredAt.toISOString()).toBe(late.toISOString());
+  });
+
+  it("stores the line as encrypted bytes, and tolerates its absence", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-crypto");
+    const { encryptToBytes, decryptFromBytes } =
+      await import("@/lib/ai/coach/bytes-codec");
+
+    await prisma.arrivalReaction.createMany({
+      data: [marker(user.id, "weight", new Date("2026-07-14T06:00:00Z"))],
+      skipDuplicates: true,
+    });
+
+    // A provider-less install leaves the column null — the marker is the
+    // feature, the sentence is garnish.
+    const before = await prisma.arrivalReaction.findFirst({
+      where: { userId: user.id },
+    });
+    expect(before?.lineEncrypted).toBeNull();
+    expect(before?.generatedAt).toBeNull();
+
+    const line = "Last night is in — a steady base for the day.";
+    await prisma.arrivalReaction.update({
+      where: { id: before!.id },
+      data: { lineEncrypted: encryptToBytes(line), generatedAt: new Date() },
+    });
+
+    const after = await prisma.arrivalReaction.findFirst({
+      where: { userId: user.id },
+    });
+    expect(after?.lineEncrypted).not.toBeNull();
+    // Ciphertext at rest — the plaintext must not be readable off the column.
+    expect(Buffer.from(after!.lineEncrypted!).toString("utf8")).not.toContain(
+      "steady base",
+    );
+    expect(decryptFromBytes(after!.lineEncrypted!)).toBe(line);
+  });
+
+  it("cascades on user delete, leaving no orphan markers", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-cascade");
+    await prisma.arrivalReaction.createMany({
+      data: [marker(user.id, "weight", new Date("2026-07-14T06:00:00Z"))],
+      skipDuplicates: true,
+    });
+
+    await prisma.user.delete({ where: { id: user.id } });
+    expect(await prisma.arrivalReaction.count()).toBe(0);
+  });
+
+  it("the retention window deletes only rows past it", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("arrival-retention");
+    const now = new Date();
+    const old = new Date(now.getTime() - 20 * 86_400_000);
+    const recent = new Date(now.getTime() - 3 * 86_400_000);
+
+    await prisma.arrivalReaction.create({
+      data: {
+        ...marker(user.id, "weight", now),
+        localDate: "2026-06-24",
+        createdAt: old,
+      },
+    });
+    await prisma.arrivalReaction.create({
+      data: {
+        ...marker(user.id, "sleep_night", now),
+        localDate: "2026-07-11",
+        createdAt: recent,
+      },
+    });
+
+    const cutoff = new Date();
+    cutoff.setUTCDate(cutoff.getUTCDate() - 14);
+    const deleted = await prisma.arrivalReaction.deleteMany({
+      where: { createdAt: { lt: cutoff } },
+    });
+
+    expect(deleted.count).toBe(1);
+    const survivor = await prisma.arrivalReaction.findFirst();
+    expect(survivor?.kind).toBe("sleep_night");
+  });
+});