diff --git a/prisma/migrations/0257_arrival_reactions/migration.sql b/prisma/migrations/0257_arrival_reactions/migration.sql
new file mode 100644
index 000000000..a363319bd
--- /dev/null
+++ b/prisma/migrations/0257_arrival_reactions/migration.sql
@@ -0,0 +1,53 @@
+-- The data-arrival spine's durable reaction marker.
+--
+-- One row per (user, arrival kind, local day), written by the `data-arrival`
+-- worker after a SALIENT ingest. A backfill never reaches here: the seam
+-- classifier drops historical and future-dated samples before anything is
+-- enqueued, so a ten-year import writes nothing to this table.
+--
+-- The unique index is the point of the table, not decoration. It carries three
+-- separate contracts at once:
+--   1. the "just in" marker every reaction surface reads,
+--   2. the at-most-one-generated-line per kind per day claim — the worker
+--      inserts with ON CONFLICT DO NOTHING, so a lost race simply means
+--      another worker already owns today's line,
+--   3. the audit trail of what the spine actually reacted to.
+--
+-- `line_encrypted` is AES-256-GCM at rest, following the
+-- `insight_narratives.encrypted_content` precedent. It is NULLABLE on purpose:
+-- the marker is the feature and the sentence is garnish, so a provider-less
+-- install fills the table normally and simply never sets this column.
+--
+-- Rows are pruned after 14 days by the daily `arrival-reaction-cleanup` job.
+
+CREATE TABLE "arrival_reactions" (
+  "id" TEXT NOT NULL,
+  "user_id" TEXT NOT NULL,
+  "kind" TEXT NOT NULL,
+  "local_date" TEXT NOT NULL,
+  "occurred_at" TIMESTAMP(3) NOT NULL,
+  "ref_id" TEXT,
+  "line_encrypted" BYTEA,
+  "generated_at" TIMESTAMP(3),
+  "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+  CONSTRAINT "arrival_reactions_pkey" PRIMARY KEY ("id")
+);
+
+-- The once-per-kind-per-local-day claim. Do NOT widen this to include
+-- `ref_id`: the day scope IS the throttle.
+CREATE UNIQUE INDEX "arrival_reactions_user_id_kind_local_date_key"
+  ON "arrival_reactions" ("user_id", "kind", "local_date");
+
+-- The reaction surfaces read "today's rows for this user".
+CREATE INDEX "arrival_reactions_user_id_local_date_idx"
+  ON "arrival_reactions" ("user_id", "local_date");
+
+-- The nightly retention sweep deletes by age.
+CREATE INDEX "arrival_reactions_created_at_idx"
+  ON "arrival_reactions" ("created_at");
+
+ALTER TABLE "arrival_reactions"
+  ADD CONSTRAINT "arrival_reactions_user_id_fkey"
+  FOREIGN KEY ("user_id") REFERENCES "users"("id")
+  ON DELETE CASCADE ON UPDATE CASCADE;
diff --git a/prisma/schema.prisma b/prisma/schema.prisma
index ffaaff10a..4b9712d20 100644
--- a/prisma/schema.prisma
+++ b/prisma/schema.prisma
@@ -792,6 +792,10 @@ model User {
   // v1.11.0 W3 — one durable period-narrative row per (period, locale).
   // Delete/regenerate-clean; the generated prose is AES-256-GCM at rest.
   insightNarratives           InsightNarrative[]
+  // v1.31.0 — the data-arrival spine's durable per-day reaction marker. One
+  // row per (kind, local day); the unique index IS the once-per-kind-per-day
+  // claim every reaction surface reads. Retained 14 days.
+  arrivalReactions            ArrivalReaction[]
   // v1.17.1 — preventive-care / measurement (Vorsorge) reminders. One row
   // per user-created reminder ("BP every 7 days", "annual blood panel").
   // Server computes the canonical `nextDueAt`; the cron fires through the
@@ -3242,9 +3246,9 @@ model RefreshToken {
 // the token pair the first exchange issued (refresh-token reuse-detection
 // posture). 90-second TTL enforced at read; a daily sweep prunes stale rows.
 model OidcNativeHandoff {
-  id        String @id @default(cuid())
-  userId    String @map("user_id")
-  codeHash  String @unique @map("code_hash") // hashToken(hlh_<…>)
+  id                     String    @id @default(cuid())
+  userId                 String    @map("user_id")
+  codeHash               String    @unique @map("code_hash") // hashToken(hlh_<…>)
   /// The app's S256 PKCE challenge, bound at mint; the exchange verifies
   /// s256(code_verifier) against it (constant-time) before consuming.
   codeChallenge          String    @map("code_challenge")
@@ -3254,8 +3258,8 @@ model OidcNativeHandoff {
   /// revokes exactly this family member (+ its paired access token).
   issuedRefreshTokenHash String?   @map("issued_refresh_token_hash")
   createdAt              DateTime  @default(now()) @map("created_at")
-  ipAddress             String?   @map("ip_address")
-  userAgent             String?   @map("user_agent")
+  ipAddress              String?   @map("ip_address")
+  userAgent              String?   @map("user_agent")
 
   user User @relation(fields: [userId], references: [id], onDelete: Cascade)
 
@@ -5258,6 +5262,49 @@ model InsightNarrative {
   @@map("insight_narratives")
 }
 
+// ─── Data-arrival spine (v1.31.0) ─────────────────────────────
+//
+// One row per (user, arrival kind, local day). The row is written by the
+// `data-arrival` worker after a SALIENT ingest — never by a backfill, which
+// the seam classifier drops before anything is enqueued.
+//
+// The unique index is load-bearing three times over: it is the "just in"
+// marker the reaction surfaces read, it is the at-most-one-generated-line
+// per kind per day claim (inserted with ON CONFLICT DO NOTHING, so a lost
+// race means another worker already owns today's line), and it is the audit
+// trail for what the spine actually reacted to.
+model ArrivalReaction {
+  id            String    @id @default(cuid())
+  userId        String    @map("user_id")
+  /// One of `ARRIVAL_KINDS` (`@/lib/arrivals/types`).
+  kind          String
+  /// The user's PROFILE-timezone YYYY-MM-DD this reaction is filed under.
+  localDate     String    @map("local_date")
+  /// Timestamp of the newest sample that triggered it.
+  occurredAt    DateTime  @map("occurred_at")
+  /// Kind-scoped referent: the workout id for `workout`, the panel day for labs.
+  refId         String?   @map("ref_id")
+  /// AES-256-GCM ciphertext of the generated reaction line. Null until — and
+  /// unless — a line is generated: the marker is the feature, the sentence is
+  /// garnish, and a provider-less install never fills this in. Follows the
+  /// `InsightNarrative.encryptedContent` precedent, NOT the plaintext pattern
+  /// the status cards use. Registered in `scripts/rotate-encryption-key.ts`.
+  lineEncrypted Bytes?    @map("line_encrypted")
+  generatedAt   DateTime? @map("generated_at")
+  createdAt     DateTime  @default(now()) @map("created_at")
+
+  user User @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  // The once-per-kind-per-local-day claim. Do not relax this to include
+  // `refId` — the day scope is the throttle.
+  @@unique([userId, kind, localDate])
+  // The reaction surfaces read "today's rows for this user".
+  @@index([userId, localDate])
+  // The nightly retention sweep deletes by age.
+  @@index([createdAt])
+  @@map("arrival_reactions")
+}
+
 // ─── Cycle Tracking (v1.15.0) ─────────────────────────────────
 //
 // Self-hosted reproductive-health tracking: period + flow, symptom
@@ -6334,14 +6381,14 @@ model DocumentConditionLink {
 /// versioned key (no schema column), and key rotation re-tokenises from the
 /// decrypted `textEncrypted`.
 model DocumentContentIndex {
-  id               String   @id @default(cuid())
-  documentId       String   @unique @map("document_id")
-  userId           String   @map("user_id")
+  id                    String   @id @default(cuid())
+  documentId            String   @unique @map("document_id")
+  userId                String   @map("user_id")
   /// AES-256-GCM ciphertext of the normalised extracted text (capped, first
   /// ~64 KiB), stored as the `encrypt()`-string-as-UTF-8 Bytes shape shared
   /// with the Coach columns. The plaintext is recoverable server-side so
   /// rotation can re-tokenise; the list query never selects it.
-  textEncrypted    Bytes    @map("text_encrypted")
+  textEncrypted         Bytes    @map("text_encrypted")
   /// v1.27.33 (Document vault P4 — chat about a document) — AES-256-GCM
   /// ciphertext of the VERBATIM extracted text (byte-capped, NOT lowercased or
   /// de-accented), stored additionally alongside `textEncrypted` so a document
@@ -6350,22 +6397,22 @@ model DocumentContentIndex {
   /// UTF-8 Bytes shape as every other PHI column. Nullable: rows indexed before
   /// P4 carry NULL until re-indexed, and the chat route falls back to the
   /// normalised `textEncrypted` when it is absent. The list query never selects it.
-  verbatimTextEncrypted Bytes? @map("verbatim_text_encrypted")
+  verbatimTextEncrypted Bytes?   @map("verbatim_text_encrypted")
   /// Deduped HMAC-SHA256 (truncated hex) of the normalised tokens — the blind
   /// index. One-way: opaque without the server-held index subkey. GIN-indexed
   /// (raw SQL in the migration; Prisma cannot express a GIN index).
-  searchTokens     String[] @default([]) @map("search_tokens")
+  searchTokens          String[] @default([]) @map("search_tokens")
   /// Provenance of the indexed text: "vision" (provider transcription of the
   /// stored original), "text-ocr" (browser-OCR text posted by the client),
   /// "local-pdf" (server-side text-layer extraction, no provider) or "local-ocr"
   /// (server-side OCR of a scanned document — deferred follow-up). Free String;
   /// adding a source value needs no migration.
-  source           String
+  source                String
   /// Provider type that produced the text (diagnostic only), or null on the
   /// text-ocr path.
-  providerType     String?  @map("provider_type")
+  providerType          String?  @map("provider_type")
   /// Tokeniser algorithm version — bump to force a re-index on a scheme change.
-  tokenizerVersion String   @map("tokenizer_version")
+  tokenizerVersion      String   @map("tokenizer_version")
 
   createdAt DateTime @default(now()) @map("created_at")
   updatedAt DateTime @updatedAt @map("updated_at")
@@ -6386,23 +6433,23 @@ model DocumentContentIndex {
 /// source, so the preview is metadata-free by construction. Owner-scoped
 /// serve route only; cascades on delete.
 model DocumentThumbnail {
-  id                 String   @id @default(cuid())
-  documentId         String   @unique @map("document_id")
-  userId             String   @map("user_id")
+  id                 String @id @default(cuid())
+  documentId         String @unique @map("document_id")
+  userId             String @map("user_id")
   /// AES-256-GCM ciphertext of the JPEG preview, stored as the
   /// `encrypt()`-string-as-UTF-8 Bytes shape (base64 of the JPEG), the same
   /// codec `DocumentContentIndex.textEncrypted` uses so the standard
   /// key-rotation walk covers it. A scanned medical preview is PHI — never
   /// stored in the clear or logged. The list query never selects it.
-  thumbnailEncrypted Bytes    @map("thumbnail_encrypted")
+  thumbnailEncrypted Bytes  @map("thumbnail_encrypted")
   /// Decoded thumbnail dimensions (px) — diagnostic + lets a future consumer
   /// size the tile without decoding the blob.
   width              Int
   height             Int
   /// Plaintext byte length of the JPEG (pre-encryption), for `Content-Length`.
-  byteSize           Int      @map("byte_size")
+  byteSize           Int    @map("byte_size")
   /// Source MIME the preview was rendered from (diagnostic only).
-  sourceMime         String   @map("source_mime")
+  sourceMime         String @map("source_mime")
 
   createdAt DateTime @default(now()) @map("created_at")
   updatedAt DateTime @updatedAt @map("updated_at")
diff --git a/scripts/rotate-encryption-key.ts b/scripts/rotate-encryption-key.ts
index 7ebf9f0dd..a07276aa1 100644
--- a/scripts/rotate-encryption-key.ts
+++ b/scripts/rotate-encryption-key.ts
@@ -392,6 +392,19 @@ async function main() {
     ),
   );
 
+  // ───── v1.31.0 ArrivalReaction."lineEncrypted" (Bytes, nullable) ─────
+  // The data-arrival spine's generated reaction line. Nullable by design — a
+  // provider-less install writes markers with no line at all — and
+  // `rotateBytesColumn` already skips a NULL / zero-length payload, so those
+  // rows are scanned and cleanly left alone.
+  results.push(
+    await rotateBytesColumn(
+      "ArrivalReaction",
+      "lineEncrypted",
+      prisma.arrivalReaction,
+    ),
+  );
+
   // ───── v1.18.1 clinical-spine notes (Bytes columns) ─────
   // "noteEncrypted" (LabResult / IllnessEpisode / IllnessDayLog) +
   // "contextEncrypted" (Biomarker). Mirror the CoachFact.factEncrypted block.
diff --git a/src/app/api/labs/ocr/commit/route.ts b/src/app/api/labs/ocr/commit/route.ts
index 4f7decad4..5d3f37b82 100644
--- a/src/app/api/labs/ocr/commit/route.ts
+++ b/src/app/api/labs/ocr/commit/route.ts
@@ -25,6 +25,7 @@ import { auditLog } from "@/lib/auth/audit";
 import { prisma } from "@/lib/db";
 import { withIdempotency } from "@/lib/idempotency";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import { resolveOrMintBiomarker } from "@/lib/labs/biomarker-store";
 import { serialiseLabResult } from "@/lib/labs/serialise";
 import {
@@ -214,6 +215,24 @@ async function commitOcrRows(request: NextRequest) {
     fireAndForget(enqueueReminderSatisfy(user.id), {
       action: "reminder.satisfy.enqueue",
     });
+
+    // v1.31.0 — the labs arm of the data-arrival spine. This seam DOES know
+    // its whole panel (the commit builds a per-row `linkable` list), so it
+    // emits once with the newest draw date and the true inserted count rather
+    // than leaning on the singleton key to collapse per-row emits.
+    const newestTakenAt = linkable.reduce<Date | null>(
+      (acc, r) => (!acc || r.takenAt > acc ? r.takenAt : acc),
+      null,
+    );
+    if (newestTakenAt) {
+      void emitDataArrival({
+        userId: user.id,
+        kind: "labs_panel",
+        newestSampleAt: newestTakenAt,
+        insertedCount: inserted.length,
+        source: "ocr",
+      }).catch(() => {});
+    }
   }
 
   return apiSuccess({ inserted, skipped });
diff --git a/src/app/api/labs/route.ts b/src/app/api/labs/route.ts
index 1bab6a5fb..2e30996a1 100644
--- a/src/app/api/labs/route.ts
+++ b/src/app/api/labs/route.ts
@@ -14,6 +14,7 @@ import { auditLog } from "@/lib/auth/audit";
 import { prisma } from "@/lib/db";
 import { withIdempotency } from "@/lib/idempotency";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import { resolveOrMintBiomarker } from "@/lib/labs/biomarker-store";
 import {
   type ResolvedBiomarker,
@@ -276,5 +277,18 @@ async function postLabResult(request: NextRequest) {
     action: "reminder.satisfy.enqueue",
   });
 
+  // v1.31.0 — the labs arm of the data-arrival spine. A "panel" has no
+  // first-class entity in the schema (it is a nullable label plus a `takenAt`),
+  // so the day-scoped singleton key IS the panel grouping: pasting twelve
+  // markers from one draw fires twelve emits that collapse into one arrival.
+  void emitDataArrival({
+    userId: user.id,
+    kind: "labs_panel",
+    newestSampleAt: created.takenAt,
+    insertedCount: 1,
+    refId: created.panel ?? undefined,
+    source: "manual",
+  }).catch(() => {});
+
   return apiSuccess(serialiseLabResult(created, biomarker), 201);
 }
diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts
index 96a4e60e6..e561d42d0 100644
--- a/src/app/api/measurements/batch/route.ts
+++ b/src/app/api/measurements/batch/route.ts
@@ -66,6 +66,8 @@ import { checkRateLimit } from "@/lib/rate-limit";
 import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
+import { groupRowsByArrivalKind } from "@/lib/arrivals/measurement-kind";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
@@ -177,6 +179,39 @@ type BatchEntry = z.infer<typeof batchEntrySchema>;
  */
 type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped";
 
+/**
+ * v1.31.0 — the rows this write actually INSERTED, in the shape the arrival
+ * kind grouper takes.
+ *
+ * Restricting to `status === "inserted"` is what keeps a re-sync silent: an
+ * `updated` row is the `stats:` overwrite contract re-stating a value the
+ * record already had, and a `duplicate` row landed nothing. The per-entry
+ * statuses are reconciled against the insert race upstream, so the count is
+ * accurate; in a rare concurrent-write tie it can be off on WHICH row, which
+ * cannot change the outcome because the day-scoped singleton key collapses the
+ * result either way.
+ */
+function insertedArrivalRows(
+  prepared: ReadonlyArray<{
+    index: number;
+    row: Prisma.MeasurementCreateManyInput;
+  }>,
+  results: ReadonlyArray<{ status: EntryStatus } | undefined>,
+): Array<{ type: MeasurementType; measuredAt: Date }> {
+  const rows: Array<{ type: MeasurementType; measuredAt: Date }> = [];
+  for (const p of prepared) {
+    if (results[p.index]?.status !== "inserted") continue;
+    rows.push({
+      type: p.row.type as MeasurementType,
+      measuredAt:
+        p.row.measuredAt instanceof Date
+          ? p.row.measuredAt
+          : new Date(p.row.measuredAt as string),
+    });
+  }
+  return rows;
+}
+
 // v1.5.0 issue #213 — `stats:*` externalIds carry per-day cumulative
 // totals that the iOS HealthKit observer re-posts as the day progresses.
 // Treating those as immutable duplicates freezes the day's tile at the
@@ -828,6 +863,35 @@ async function postBatch(request: NextRequest): Promise<Response> {
             : new Date(p.row.measuredAt as string),
         ),
     ).catch(() => {});
+
+    // v1.31.0 — the measurement arm of the data-arrival spine.
+    //
+    // ONE emit per kind per request, never one per row. The batch route is the
+    // Apple-Health / iOS ingest path, so it is also the single biggest backfill
+    // path in the product: a ten-year export arrives here in 500-row batches.
+    // Emitting per row would put thousands of classifier calls (and thousands
+    // of dropped-event annotations) on the hot path for no gain, since the
+    // day-scoped singleton key collapses them to one job anyway.
+    //
+    // Blood pressure is deliberately one arrival, not two: it is stored as two
+    // rows (SYS + DIA) sharing a `measuredAt`, and a reader cares about the
+    // reading, not the arms.
+    //
+    // Only rows the write actually INSERTED count — `updatedCount` rows are the
+    // `stats:` overwrite contract re-stating a value the record already had,
+    // which is not news. If nothing was inserted the classifier returns `noop`
+    // and nothing is enqueued.
+    for (const [kind, group] of groupRowsByArrivalKind(
+      insertedArrivalRows(prepared, results),
+    )) {
+      void emitDataArrival({
+        userId: user.id,
+        kind,
+        newestSampleAt: group.newestAt,
+        insertedCount: group.count,
+        source: "batch",
+      }).catch(() => {});
+    }
   }
 
   return apiSuccess({
diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts
index 3d64bb0f8..cc06ff008 100644
--- a/src/app/api/measurements/route.ts
+++ b/src/app/api/measurements/route.ts
@@ -32,6 +32,11 @@ import {
 import { withIdempotency } from "@/lib/idempotency";
 import { encryptNote, shapeMeasurementNotes } from "@/lib/crypto/note-cipher";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
+import {
+  ARRIVAL_MEASUREMENT_KIND,
+  groupRowsByArrivalKind,
+} from "@/lib/arrivals/measurement-kind";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { runSafetyFloorCheck } from "@/lib/illness/safety-floor-check";
@@ -846,6 +851,21 @@ async function postMeasurement(request: NextRequest) {
       action: "reminder.satisfy.enqueue",
     });
 
+    // v1.31.0 — the interactive arm of the data-arrival spine. This route is
+    // the combined BP + pulse form, so it lands both BP arms in one
+    // transaction; they are one reading and produce one arrival, which the
+    // kind map already collapses. Every row here is a real insert (the
+    // transaction throws rather than de-duplicating), so the count is exact.
+    for (const [kind, group] of groupRowsByArrivalKind(results)) {
+      void emitDataArrival({
+        userId: user.id,
+        kind,
+        newestSampleAt: group.newestAt,
+        insertedCount: group.count,
+        source: "manual",
+      }).catch(() => {});
+    }
+
     // v1.18.6 — absolute clinical safety-floor check on the just-written
     // readings (confirm-gated, module-gated, never diagnoses). The combined
     // BP form lands both arms here, so the BP floor sees a whole reading.
@@ -998,6 +1018,22 @@ async function postMeasurement(request: NextRequest) {
     action: "reminder.satisfy.enqueue",
   });
 
+  // v1.31.0 — the single-entry arm of the data-arrival spine. One row, so at
+  // most one kind; a type the spine does not track maps to null and emits
+  // nothing.
+  {
+    const kind = ARRIVAL_MEASUREMENT_KIND[measurement.type];
+    if (kind) {
+      void emitDataArrival({
+        userId: user.id,
+        kind,
+        newestSampleAt: measurement.measuredAt,
+        insertedCount: 1,
+        source: "manual",
+      }).catch(() => {});
+    }
+  }
+
   // v1.18.6 — absolute clinical safety-floor check (confirm-gated,
   // module-gated, never diagnoses). A single-entry POST carries only one arm
   // of a BP reading, so the BP floor only fires through the combined-form /
diff --git a/src/app/api/workouts/batch/route.ts b/src/app/api/workouts/batch/route.ts
index 05fe3c853..86c5b964a 100644
--- a/src/app/api/workouts/batch/route.ts
+++ b/src/app/api/workouts/batch/route.ts
@@ -68,6 +68,7 @@ import { withIdempotency } from "@/lib/idempotency";
 import { checkRateLimit } from "@/lib/rate-limit";
 import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import {
   createBatchWorkoutSchema,
   MAX_WORKOUTS_PER_BATCH,
@@ -118,6 +119,65 @@ interface EntryResult {
   reason?: string;
 }
 
+/**
+ * v1.31.0 — emit one data arrival per workout this batch actually inserted.
+ *
+ * `createMany` does not return ids, so they are recovered with a single indexed
+ * lookup on the `(userId, source, externalId)` dedup key. Entries without an
+ * external id cannot be looked up and are emitted without a `refId`; the spine
+ * then falls back to a day-scoped key for them, which is an honest degradation
+ * (no external id means no stable per-workout referent to react to anyway).
+ *
+ * Fully best-effort — every failure path returns quietly. A reaction is never
+ * worth failing an ingest that already succeeded.
+ */
+async function emitWorkoutArrivals(
+  userId: string,
+  prepared: ReadonlyArray<{
+    index: number;
+    row: Prisma.WorkoutCreateManyInput;
+    dedupKey: { source: string; externalId: string } | null;
+  }>,
+  results: ReadonlyArray<EntryResult | undefined>,
+): Promise<void> {
+  const inserted = prepared.filter(
+    (p) => results[p.index]?.status === "inserted",
+  );
+  if (inserted.length === 0) return;
+
+  const externalIds = inserted
+    .map((p) => p.dedupKey?.externalId)
+    .filter((id): id is string => typeof id === "string");
+
+  const idByExternalId = new Map<string, string>();
+  if (externalIds.length > 0) {
+    const rows = await prisma.workout.findMany({
+      where: { userId, externalId: { in: externalIds } },
+      select: { id: true, externalId: true },
+    });
+    for (const row of rows) {
+      if (row.externalId) idByExternalId.set(row.externalId, row.id);
+    }
+  }
+
+  for (const p of inserted) {
+    const startedAt =
+      p.row.startedAt instanceof Date
+        ? p.row.startedAt
+        : new Date(p.row.startedAt as string);
+    if (Number.isNaN(startedAt.getTime())) continue;
+    const externalId = p.dedupKey?.externalId;
+    await emitDataArrival({
+      userId,
+      kind: "workout",
+      newestSampleAt: startedAt,
+      insertedCount: 1,
+      refId: externalId ? idByExternalId.get(externalId) : undefined,
+      source: "batch",
+    });
+  }
+}
+
 export const POST = apiHandler(withIdempotency<[NextRequest]>(postBatch));
 
 async function postBatch(request: NextRequest): Promise<Response> {
@@ -673,6 +733,20 @@ async function postBatch(request: NextRequest): Promise<Response> {
   // workout-derived metrics.
   if (insertedCount > 0) {
     invalidateUserMeasurements(user.id);
+
+    // v1.31.0 — the workout arm of the data-arrival spine.
+    //
+    // Unlike every other kind, workouts emit ONE arrival PER workout, keyed by
+    // the workout's own id. Two sessions in a day are two events and each
+    // deserves its own reaction; a day-scoped key would silently collapse the
+    // second one.
+    //
+    // That needs the ids, and `createMany` does not return them, so this costs
+    // one extra indexed lookup — but only on a batch that actually inserted
+    // something, and only for entries carrying an external id. The classifier
+    // still runs per emit, so a historical import walks straight into
+    // `backfill` and enqueues nothing regardless of how many rows it wrote.
+    void emitWorkoutArrivals(user.id, prepared, results).catch(() => {});
   }
 
   return apiSuccess({
diff --git a/src/lib/arrivals/emit-shared.ts b/src/lib/arrivals/emit-shared.ts
new file mode 100644
index 000000000..c48e9b37d
--- /dev/null
+++ b/src/lib/arrivals/emit-shared.ts
@@ -0,0 +1,224 @@
+/**
+ * Generator-free enqueue contract for the `data-arrival` queue.
+ *
+ * Mirrors the `morning-digest-refresh-shared.ts` split exactly: queue name +
+ * payload + `boss.send` live here, with NO import of the worker module, so a
+ * sleep sync, a workout batch, or a lab write never drags the reaction tree
+ * (and its transitive insight imports) into its own module graph. The worker
+ * lives in `@/lib/jobs/data-arrival` and re-exports the queue name from here so
+ * there is one source of truth.
+ *
+ * Cost discipline at the seam, because this code runs in front of every ingest
+ * path in the product:
+ *
+ *   - Called ONCE per write batch, never per row. The caller passes the newest
+ *     sample it inserted and how many rows it inserted; the spine never walks a
+ *     batch.
+ *   - No row reads. The only lookups are `resolveUserTimezone` and
+ *     `resolveModuleMap`, both process-cached per user, and both already on the
+ *     hot path of the seams this joins.
+ *   - Fire-and-forget and fully try/caught. An emit failure can NEVER fail the
+ *     ingest that called it — the discipline of `maybeEnqueueMorningRefresh`
+ *     (`morning-refresh-trigger.ts`, its outer catch). Callers still
+ *     `void emitDataArrival(...).catch(() => {})` for defence in depth.
+ *   - Non-salient never reaches the queue. A backfill is annotated and dropped
+ *     at the seam, so trigger volume is observable before any surface consumes
+ *     the spine.
+ */
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+import { annotate } from "@/lib/logging/context";
+import { resolveModuleMap } from "@/lib/modules/gate";
+import { resolveUserTimezone } from "@/lib/tz/resolver";
+
+import { arrivalLocalDate, classifyArrival } from "./salience";
+import type { ArrivalKind, DataArrival } from "./types";
+
+export const DATA_ARRIVAL_QUEUE = "data-arrival";
+
+/**
+ * Device double-post window for the one kind whose key is not day-scoped.
+ *
+ * `workout` is keyed by the workout id rather than the local date, because two
+ * different workouts on one day are two different events and must both react.
+ * That key is re-fireable, so it takes a time window the way
+ * `insight-status-generate-shared.ts` does. Five minutes covers a watch
+ * posting the same session twice during a sync retry.
+ *
+ * Every OTHER kind is day-scoped and takes NO window — see the table in
+ * `arrivalSendOptions` for why a wall-clock throttle is the wrong tool there.
+ */
+const WORKOUT_SINGLETON_SECONDS = 300;
+
+export interface EmitArrivalInput {
+  userId: string;
+  kind: ArrivalKind;
+  /** The newest sample this write actually inserted. */
+  newestSampleAt: Date;
+  /**
+   * Rows INSERTED by this write. An upsert that merely updated an existing row
+   * MUST pass 0 — that is what keeps a re-sync silent.
+   */
+  insertedCount: number;
+  /** Workout id for `workout`; the panel's collection day for `labs_panel`. */
+  refId?: string;
+  /** Transport token, for annotations only. Never used for a decision. */
+  source: string;
+  /** Injectable for tests; production always uses the real clock. */
+  now?: Date;
+}
+
+/**
+ * The module each kind depends on. A user who turned the domain off gets no
+ * events for it — checked at the seam so a disabled module costs one cached
+ * map lookup rather than a queued job.
+ *
+ * `weight` and `blood_pressure` are CORE domains (always on, see
+ * `CORE_DOMAIN_KEYS` in `@/lib/modules/registry`), so they name no gate.
+ */
+const KIND_MODULE: Partial<Record<ArrivalKind, "sleep" | "workouts" | "labs">> =
+  {
+    sleep_night: "sleep",
+    workout: "workouts",
+    labs_panel: "labs",
+  };
+
+/**
+ * Per-kind de-duplication key + window.
+ *
+ * The day-scoped kinds take a day-unique key and NO `singletonSeconds`, which
+ * is the shape `morning-digest-refresh-shared.ts` documents at length: a
+ * `singletonSeconds` window is a `floor()` over WALL-CLOCK time, so it can
+ * neither express a user's local date nor line up with one. It would both split
+ * a single local day across two buckets and merge two local days into one. The
+ * local date already lives in the key; the queue policy just has to honour it.
+ *
+ * Honouring it is not automatic. Under pg-boss's default `standard` policy NO
+ * index covers `singleton_key`, so a bare key coalesces NOTHING. The queue is
+ * therefore registered `exclusive` in `reminder/register-status.ts`; if that
+ * entry is ever removed these keys go silently inert. The two belong together.
+ */
+function arrivalSendOptions(
+  kind: ArrivalKind,
+  userId: string,
+  localDate: string,
+  refId: string | undefined,
+): Record<string, unknown> {
+  const base = {
+    // The worker is cheap and provider-free, so it takes the morning-refresh
+    // retry policy rather than the slower LLM one.
+    retryLimit: 2,
+    retryDelay: 60,
+    retryBackoff: true,
+  };
+
+  if (kind === "workout") {
+    return {
+      ...base,
+      singletonKey: `arrival:${userId}:workout:${refId ?? localDate}`,
+      singletonSeconds: WORKOUT_SINGLETON_SECONDS,
+    };
+  }
+
+  return {
+    ...base,
+    singletonKey: `arrival:${userId}:${kind}:${localDate}`,
+  };
+}
+
+/**
+ * Classify a write, and enqueue a `DataArrival` only if it is genuinely news.
+ *
+ * Returns nothing and throws nothing. Every exit annotates, so the volume of
+ * emitted vs. skipped events is visible in wide events before any surface
+ * consumes the spine.
+ */
+export async function emitDataArrival(input: EmitArrivalInput): Promise<void> {
+  const { userId, kind, newestSampleAt, insertedCount, refId, source } = input;
+  const now = input.now ?? new Date();
+
+  try {
+    const moduleKey = KIND_MODULE[kind];
+    if (moduleKey) {
+      const modules = await resolveModuleMap(userId);
+      if (modules[moduleKey] === false) {
+        annotate({
+          action: { name: `arrival.${kind}.skipped` },
+          meta: { reason: "module_off", source },
+        });
+        return;
+      }
+    }
+
+    const tz = await resolveUserTimezone(userId);
+    const classification = classifyArrival({
+      kind,
+      newestSampleAt,
+      insertedCount,
+      now,
+      tz,
+    });
+
+    if (classification !== "salient") {
+      // The diagnostic gate. A mass import lands here for every one of its
+      // batches, so the meta carries enough to tell a backfill apart from a
+      // timezone mistake without a code change.
+      annotate({
+        action: { name: `arrival.${kind}.skipped` },
+        meta: {
+          reason: classification,
+          source,
+          tz,
+          inserted: insertedCount,
+          sample_local_date:
+            classification === "backfill"
+              ? arrivalLocalDate(newestSampleAt, tz)
+              : undefined,
+        },
+      });
+      return;
+    }
+
+    const localDate = arrivalLocalDate(now, tz);
+    const payload: DataArrival = {
+      userId,
+      kind,
+      salience: "salient",
+      localDate,
+      occurredAt: newestSampleAt.toISOString(),
+      refId,
+      count: insertedCount,
+      source,
+    };
+
+    const boss = getGlobalBoss();
+    if (!boss) {
+      // A web process without an embedded worker. Not an error: the surfaces
+      // this spine feeds all degrade to their deterministic form, and the next
+      // landing on a worker-bearing process re-emits.
+      annotate({
+        action: { name: `arrival.${kind}.skipped` },
+        meta: { reason: "no_boss", source, local_date: localDate },
+      });
+      return;
+    }
+
+    await boss.send(
+      DATA_ARRIVAL_QUEUE,
+      payload,
+      arrivalSendOptions(kind, userId, localDate, refId),
+    );
+
+    annotate({
+      action: { name: `arrival.${kind}.emitted` },
+      meta: {
+        salience: "salient",
+        count: insertedCount,
+        source,
+        local_date: localDate,
+      },
+    });
+  } catch {
+    // Never let a reaction trigger fail an ingest. The data is already written;
+    // the reaction is garnish, and the next landing is the catch-net.
+  }
+}
diff --git a/src/lib/arrivals/measurement-kind.ts b/src/lib/arrivals/measurement-kind.ts
new file mode 100644
index 000000000..dd533116c
--- /dev/null
+++ b/src/lib/arrivals/measurement-kind.ts
@@ -0,0 +1,60 @@
+/**
+ * Measurement type → arrival kind mapping, shared by the three measurement
+ * write seams (the batch route, and both arms of the interactive POST).
+ *
+ * Kept in one place because the mapping encodes a product judgement that must
+ * not drift between seams: blood pressure is stored as TWO rows (a systolic and
+ * a diastolic sharing one `measuredAt`) but is ONE reading to every reader, so
+ * both types map to the same kind and a write that lands both produces a single
+ * arrival.
+ *
+ * Types absent from the map are simply not part of the spine. That is the
+ * default, and adding one is a deliberate act: a new kind also needs a surface
+ * that consumes it and, if it can cause a provider call, a stated per-user
+ * daily bound.
+ */
+import type { MeasurementType } from "@/generated/prisma/client";
+
+import type { ArrivalKind } from "./types";
+
+export const ARRIVAL_MEASUREMENT_KIND: Partial<
+  Record<MeasurementType, ArrivalKind>
+> = {
+  WEIGHT: "weight",
+  BLOOD_PRESSURE_SYS: "blood_pressure",
+  BLOOD_PRESSURE_DIA: "blood_pressure",
+};
+
+export interface ArrivalKindGroup {
+  newestAt: Date;
+  count: number;
+}
+
+/**
+ * Collapse just-written measurement rows into at most one group per arrival
+ * kind, carrying the newest timestamp and how many rows landed.
+ *
+ * The newest timestamp is what the classifier tests for recency, so a write
+ * mixing a backfilled row with a fresh one is judged on the fresh one — which
+ * is correct: the record genuinely did just gain something current.
+ */
+export function groupRowsByArrivalKind(
+  rows: ReadonlyArray<{ type: MeasurementType; measuredAt: Date }>,
+): Array<[ArrivalKind, ArrivalKindGroup]> {
+  const groups = new Map<ArrivalKind, ArrivalKindGroup>();
+  for (const row of rows) {
+    const kind = ARRIVAL_MEASUREMENT_KIND[row.type];
+    if (!kind) continue;
+    if (Number.isNaN(row.measuredAt.getTime())) continue;
+    const existing = groups.get(kind);
+    if (!existing) {
+      groups.set(kind, { newestAt: row.measuredAt, count: 1 });
+      continue;
+    }
+    existing.count++;
+    if (row.measuredAt.getTime() > existing.newestAt.getTime()) {
+      existing.newestAt = row.measuredAt;
+    }
+  }
+  return [...groups];
+}
diff --git a/src/lib/arrivals/salience.ts b/src/lib/arrivals/salience.ts
new file mode 100644
index 000000000..2c37da3c3
--- /dev/null
+++ b/src/lib/arrivals/salience.ts
@@ -0,0 +1,105 @@
+/**
+ * The anti-storm core of the arrival spine: a pure, timezone-honest recency
+ * classifier that runs AT THE SEAM, before anything is enqueued.
+ *
+ * This is the one guardrail the whole spine rests on. Every ingest path in the
+ * product is also a BACKFILL path: a ten-year Apple Health export, a WHOOP
+ * re-sync after a token refresh, a 30-day catch-up after downtime, a `stats:`
+ * overwrite storm from the iOS client. Without a recency test at the seam,
+ * those would each turn into thousands of queued jobs and, downstream, provider
+ * calls. With it, they emit exactly zero events: every sample fails the test
+ * and is annotated, never enqueued.
+ *
+ * It generalizes `isLastNightLocal` (`@/lib/daily/morning-refresh-trigger`),
+ * the predicate the S4 sleep trigger has already proven against real backfills,
+ * to every arrival kind. Pure and timezone-explicit — no DB, no ambient clock —
+ * so the timezone-boundary cases are unit-testable directly.
+ */
+import { userDayKey } from "@/lib/tz/format";
+import type { ArrivalKind } from "./types";
+
+const MS_PER_DAY = 86_400_000;
+
+/**
+ * What the seam decided about a candidate arrival.
+ *
+ * - `salient` — new data from today or yesterday (locally); enqueue.
+ * - `backfill` — historical or future-dated data; annotate and drop.
+ * - `noop` — the write landed no new rows; annotate and drop.
+ */
+export type ArrivalClassification = "salient" | "backfill" | "noop";
+
+export interface ClassifyArrivalInput {
+  kind: ArrivalKind;
+  /** The newest sample the write actually inserted. */
+  newestSampleAt: Date;
+  /** Rows INSERTED — an upsert that only updated values must pass 0 here. */
+  insertedCount: number;
+  now: Date;
+  /** The user's PROFILE timezone. Never the server's, never UTC. */
+  tz: string;
+}
+
+/**
+ * Classify a candidate arrival. Deterministic and total — every input lands in
+ * exactly one of the three buckets.
+ *
+ * Ordering matters:
+ *
+ * 1. `insertedCount <= 0` → `noop`. A pure re-sync or a `stats:` overwrite
+ *    changed values but landed no new reading. Value changes already refresh
+ *    the cards through `@/lib/insights/status-invalidation`; the spine reacts
+ *    to NEW data only, so it must not fire here.
+ * 2. Future-dated sample → `backfill`. Clock skew on a device, not news.
+ *    Rejected the same way `isLastNightLocal` rejects it.
+ * 3. Local calendar day older than yesterday → `backfill`. This is the test
+ *    that makes a mass import free. "Older than yesterday" (rather than
+ *    "older than today") is deliberate: a sleep night, and a late-evening
+ *    workout synced after midnight, both belong to yesterday locally while
+ *    still being the freshest thing the user has.
+ * 4. Otherwise → `salient`.
+ *
+ * Kind is carried for readability and future per-kind rules, but every kind
+ * shares the recency test today. The one per-kind refinement the spec calls for
+ * — `weight` is salient only as the FIRST reading of the local day — is NOT
+ * decided here: it needs a lookup, and the seam must stay free of reads. The
+ * worker enforces it structurally instead, via the `ArrivalReaction` unique row
+ * on `[userId, kind, localDate]`; a second weigh-in enqueues one cheap job that
+ * de-duplicates against that row and fans nothing out.
+ */
+export function classifyArrival(
+  input: ClassifyArrivalInput,
+): ArrivalClassification {
+  const { newestSampleAt, insertedCount, now, tz } = input;
+
+  if (!Number.isFinite(insertedCount) || insertedCount <= 0) return "noop";
+  if (
+    !(newestSampleAt instanceof Date) ||
+    Number.isNaN(newestSampleAt.getTime())
+  )
+    return "noop";
+
+  if (newestSampleAt.getTime() > now.getTime()) return "backfill";
+
+  const todayKey = userDayKey(now, tz);
+  const yesterdayKey = userDayKey(new Date(now.getTime() - MS_PER_DAY), tz);
+  const sampleKey = userDayKey(newestSampleAt, tz);
+
+  if (sampleKey !== todayKey && sampleKey !== yesterdayKey) return "backfill";
+
+  return "salient";
+}
+
+/**
+ * The `localDate` an arrival is filed under: TODAY in the user's timezone, not
+ * the sample's own day.
+ *
+ * The day key is a claim key, not a timestamp — it answers "which day does this
+ * reaction belong to", and a sleep night that ended at 06:00 today, a workout
+ * from 23:40 yesterday synced at 00:20, and this morning's weight are all part
+ * of the same day's story. Filing by the sample's own day would split them and
+ * defeat the once-per-kind-per-day claim the unique row exists to make.
+ */
+export function arrivalLocalDate(now: Date, tz: string): string {
+  return userDayKey(now, tz);
+}
diff --git a/src/lib/arrivals/types.ts b/src/lib/arrivals/types.ts
new file mode 100644
index 000000000..e7425136d
--- /dev/null
+++ b/src/lib/arrivals/types.ts
@@ -0,0 +1,71 @@
+/**
+ * The data-arrival event spine — wire types.
+ *
+ * Every ingest path in the product can emit a `DataArrival` when NEW data
+ * lands. One worker consumes them and drives the reaction surfaces. The spine
+ * sits directly in front of every write seam, so its types are deliberately
+ * closed and small: a free-form kind would let a future ingest path enqueue an
+ * unbounded event class that no reader understands, and a fat payload would
+ * tempt a seam into reading rows it does not already hold.
+ *
+ * Mirrors the discipline of `@/lib/daily/priority-item`: wire-serialisable
+ * primitives only (the payload crosses a pg-boss JSON column), no Date objects,
+ * no Prisma types.
+ */
+
+/**
+ * The closed set of arrival kinds. GROWS BY PR ONLY — adding a member is a
+ * deliberate act that must also name the surface consuming it and, if it can
+ * cause a provider call, its per-user daily bound.
+ */
+export const ARRIVAL_KINDS = [
+  /** Last night's sleep completed, from any of the seven transports. */
+  "sleep_night",
+  /** A finished workout row was inserted (not re-synced). */
+  "workout",
+  /** The first weight reading of the local day. */
+  "weight",
+  /** A fresh blood-pressure reading. */
+  "blood_pressure",
+  /** A new lab panel landed (manual, OCR confirm, or auto-stage review). */
+  "labs_panel",
+] as const;
+
+export type ArrivalKind = (typeof ARRIVAL_KINDS)[number];
+
+/**
+ * Salience tier of an emitted event. Only `"salient"` is ever enqueued today;
+ * the union is kept open for a future `"digest_only"` tier that would ride the
+ * nightly pass instead of the queue. `"backfill"` and `"noop"` classifications
+ * never become events at all — they are annotated at the seam and dropped.
+ */
+export type ArrivalSalience = "salient";
+
+/**
+ * The queued payload. Everything here is known at the write seam without an
+ * extra read: the seam already holds the rows it just wrote.
+ */
+export interface DataArrival {
+  userId: string;
+  kind: ArrivalKind;
+  /** Deterministic, computed at the emit seam. Non-salient is never enqueued. */
+  salience: ArrivalSalience;
+  /** User-profile-tz day key of the NEWEST sample (YYYY-MM-DD). */
+  localDate: string;
+  /** ISO timestamp of the newest sample in the batch. */
+  occurredAt: string;
+  /** Kind-scoped referent: the workout id for `workout`, the panel date for labs. */
+  refId?: string;
+  /** Rows actually INSERTED (not upsert-updated) by the write that emitted. */
+  count: number;
+  /** Transport token ("withings", "apple", "manual", …) for annotations only. */
+  source: string;
+}
+
+/** Type guard for a kind read back off the wire. */
+export function isArrivalKind(value: unknown): value is ArrivalKind {
+  return (
+    typeof value === "string" &&
+    (ARRIVAL_KINDS as readonly string[]).includes(value)
+  );
+}
diff --git a/src/lib/arrivals/workout-emit.ts b/src/lib/arrivals/workout-emit.ts
new file mode 100644
index 000000000..816aa9480
--- /dev/null
+++ b/src/lib/arrivals/workout-emit.ts
@@ -0,0 +1,72 @@
+/**
+ * The provider-sync workout seam.
+ *
+ * The four provider syncs (WHOOP, Fitbit, Google Health, Strava) all persist a
+ * workout with `prisma.workout.upsert`, and an upsert cannot say whether it
+ * created a row or updated one. That distinction is the whole point of the
+ * spine: a re-sync must be silent. WHOOP alone re-posts the same recent workout
+ * on every poll, so without it a single day's session would emit a fresh
+ * arrival — and dispatch a fresh downstream reaction — every polling interval,
+ * indefinitely.
+ *
+ * `wasJustCreated` recovers the distinction for free, from the row the upsert
+ * already returned, rather than paying an existence probe per workout on what
+ * is also the backfill path. See its docblock for why the comparison is sound.
+ */
+import { emitDataArrival } from "./emit-shared";
+
+/** The subset of a `Workout` row this seam needs back from an upsert. */
+export interface UpsertedWorkoutRow {
+  id: string;
+  startedAt: Date;
+  createdAt: Date;
+  updatedAt: Date;
+}
+
+/**
+ * True when the upsert CREATED this row rather than updating an existing one.
+ *
+ * `createdAt` is `@default(now())` and `updatedAt` is `@updatedAt`, so Prisma
+ * writes both from the same statement on a create and they are byte-identical.
+ * Any subsequent update bumps `updatedAt` alone, leaving it strictly greater.
+ * The two can therefore only be equal on a row that has never been updated,
+ * which — on an upsert path where every update writes the full row — means the
+ * upsert just created it.
+ *
+ * The alternative is an indexed existence probe before every upsert, which is
+ * the pattern `import-apple-health-export.ts` uses because it can batch the
+ * probe across a chunk. These seams upsert one workout at a time, so a probe
+ * would double the query count on exactly the path (a provider backfill) that
+ * most needs to stay cheap.
+ */
+export function wasJustCreated(row: {
+  createdAt: Date;
+  updatedAt: Date;
+}): boolean {
+  return row.createdAt.getTime() === row.updatedAt.getTime();
+}
+
+/**
+ * Emit a workout arrival for a just-upserted row, but only if the upsert
+ * actually created it. Best-effort: never throws, never fails the sync.
+ *
+ * The recency classifier still runs inside `emitDataArrival`, so a provider
+ * backfill that legitimately CREATES hundreds of historical rows still emits
+ * zero events — the created-vs-updated test and the recency test are two
+ * independent gates, and a backfill fails the second one.
+ */
+export async function emitWorkoutArrivalIfCreated(
+  userId: string,
+  row: UpsertedWorkoutRow,
+  source: string,
+): Promise<void> {
+  if (!wasJustCreated(row)) return;
+  await emitDataArrival({
+    userId,
+    kind: "workout",
+    newestSampleAt: row.startedAt,
+    insertedCount: 1,
+    refId: row.id,
+    source,
+  });
+}
diff --git a/src/lib/crypto/encrypted-columns.ts b/src/lib/crypto/encrypted-columns.ts
index ccd7c0dfe..318c40236 100644
--- a/src/lib/crypto/encrypted-columns.ts
+++ b/src/lib/crypto/encrypted-columns.ts
@@ -170,6 +170,13 @@ export const ENCRYPTED_COLUMNS: readonly EncryptedColumn[] = [
   // ───── Insight narratives (Bytes column) ─────
   { model: "InsightNarrative", field: "encryptedContent", kind: "bytes" },
 
+  // ───── v1.31.0 data-arrival spine (Bytes column) ─────
+  // Nullable: the reaction marker is written on every salient arrival, but the
+  // generated line only exists where a provider was reachable and in budget.
+  // `rotateBytesColumn` skips a NULL/empty payload, so a provider-less
+  // install's rows rotate as a clean no-op.
+  { model: "ArrivalReaction", field: "lineEncrypted", kind: "bytes" },
+
   // ───── v1.18.1 clinical-spine notes (Bytes columns) ─────
   { model: "LabResult", field: "noteEncrypted", kind: "bytes" },
   { model: "Biomarker", field: "contextEncrypted", kind: "bytes" },
diff --git a/src/lib/daily/morning-refresh-trigger.ts b/src/lib/daily/morning-refresh-trigger.ts
index 53e59d420..e62402931 100644
--- a/src/lib/daily/morning-refresh-trigger.ts
+++ b/src/lib/daily/morning-refresh-trigger.ts
@@ -28,6 +28,7 @@ import { resolveModuleMap } from "@/lib/modules/gate";
 import { userDayKey } from "@/lib/tz/format";
 import { resolveUserTimezone } from "@/lib/tz/resolver";
 import { enqueueMorningDigestRefresh } from "@/lib/jobs/morning-digest-refresh-shared";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 
 const MS_PER_DAY = 86_400_000;
 
@@ -143,6 +144,31 @@ export async function maybeEnqueueMorningRefresh(
         sleep_samples: sleepMeasuredAts.length,
       },
     });
+
+    // v1.31.0 — the sleep arm of the data-arrival spine.
+    //
+    // Sleep gets exactly ONE seam for all seven transports, right here, because
+    // the four gates above ARE the sleep salience rules: module off, no rows,
+    // not-last-night (the same tz-honest recency predicate the spine's
+    // classifier generalises), already-refreshed. Reaching this line means the
+    // arrival is salient by construction, so the seven sleep-write call sites
+    // inherit the spine untouched and nothing is duplicated.
+    //
+    // It still routes through `emitDataArrival` rather than enqueuing directly:
+    // the classifier agreeing a second time is free, and the uniform
+    // `arrival.*` annotations are what make trigger volume comparable across
+    // kinds.
+    const newestSleepAt = sleepMeasuredAts.reduce((a, b) =>
+      b.getTime() > a.getTime() ? b : a,
+    );
+    void emitDataArrival({
+      userId,
+      kind: "sleep_night",
+      newestSampleAt: newestSleepAt,
+      insertedCount: sleepMeasuredAts.length,
+      source: "sleep",
+      now,
+    }).catch(() => {});
   } catch {
     // Never let a freshness trigger fail an ingest. The next sleep landing and
     // the nightly cron are the catch-net; the digest stays honestly
diff --git a/src/lib/documents/commit.ts b/src/lib/documents/commit.ts
index 676ec5308..8264c1133 100644
--- a/src/lib/documents/commit.ts
+++ b/src/lib/documents/commit.ts
@@ -18,6 +18,7 @@
  */
 import { encryptToBytes } from "@/lib/ai/coach/bytes-codec";
 import { prisma } from "@/lib/db";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import { decryptFactData } from "@/lib/documents/store";
 import { resolveOrMintBiomarker } from "@/lib/labs/biomarker-store";
 import type { ExtractedFact } from "@/generated/prisma/client";
@@ -104,6 +105,23 @@ async function commitObservation(
       noteEncrypted: null,
     },
   });
+
+  // v1.31.0 — the labs arm of the data-arrival spine. A "panel" has no
+  // first-class entity in the schema (it is a nullable label plus a `takenAt`),
+  // so the day-scoped singleton key IS the panel grouping: pasting twelve
+  // markers from one draw fires twelve emits that collapse into one arrival.
+  // Hooked HERE rather than in the confirm route because the route's
+  // `CommittedRecordRef` carries only the id — the draw date and panel label
+  // are lost by the time it returns.
+  void emitDataArrival({
+    userId,
+    kind: "labs_panel",
+    newestSampleAt: created.takenAt,
+    insertedCount: 1,
+    refId: created.panel ?? undefined,
+    source: "document",
+  }).catch(() => {});
+
   return { recordType: "labResult", recordId: created.id };
 }
 
diff --git a/src/lib/fitbit/sync-workout.ts b/src/lib/fitbit/sync-workout.ts
index a430e85f7..091e5d066 100644
--- a/src/lib/fitbit/sync-workout.ts
+++ b/src/lib/fitbit/sync-workout.ts
@@ -28,6 +28,7 @@ import {
   type FitbitResourceSyncOptions,
 } from "./sync";
 import { prisma } from "@/lib/db";
+import { emitWorkoutArrivalIfCreated } from "@/lib/arrivals/workout-emit";
 import { annotate, getEvent } from "@/lib/logging/context";
 import { resolveUserTimezone } from "@/lib/tz/resolver";
 
@@ -76,7 +77,7 @@ export async function syncUserWorkout(
     if (!w) continue; // no usable time span — not a workout
 
     try {
-      await prisma.workout.upsert({
+      const saved = await prisma.workout.upsert({
         where: {
           userId_source_externalId: {
             userId,
@@ -109,7 +110,16 @@ export async function syncUserWorkout(
           maxHeartRate: w.maxHeartRate,
           minHeartRate: w.minHeartRate,
         },
+        select: {
+          id: true,
+          startedAt: true,
+          createdAt: true,
+          updatedAt: true,
+        },
       });
+      // v1.31.0 — data-arrival spine. Only a genuinely NEW workout reacts; a
+      // re-sync of an already-stored session is not news.
+      void emitWorkoutArrivalIfCreated(userId, saved, "fitbit").catch(() => {});
       imported++;
     } catch (err) {
       getEvent()?.addWarning(`fitbit: failed to upsert workout: ${err}`);
diff --git a/src/lib/google-health/sync-workout.ts b/src/lib/google-health/sync-workout.ts
index d4e6f3035..02a4b3ba1 100644
--- a/src/lib/google-health/sync-workout.ts
+++ b/src/lib/google-health/sync-workout.ts
@@ -27,6 +27,7 @@ import {
   type GoogleHealthResourceSyncOptions,
 } from "./sync";
 import { prisma } from "@/lib/db";
+import { emitWorkoutArrivalIfCreated } from "@/lib/arrivals/workout-emit";
 import { annotate, getEvent } from "@/lib/logging/context";
 import { resolveUserTimezone } from "@/lib/tz/resolver";
 
@@ -66,7 +67,7 @@ export async function syncUserWorkout(
     if (!w) continue; // no usable time span — not a workout
 
     try {
-      await prisma.workout.upsert({
+      const saved = await prisma.workout.upsert({
         where: {
           userId_source_externalId: {
             userId,
@@ -99,7 +100,18 @@ export async function syncUserWorkout(
           maxHeartRate: w.maxHeartRate,
           minHeartRate: w.minHeartRate,
         },
+        select: {
+          id: true,
+          startedAt: true,
+          createdAt: true,
+          updatedAt: true,
+        },
       });
+      // v1.31.0 — data-arrival spine. Only a genuinely NEW workout reacts; a
+      // re-sync of an already-stored session is not news.
+      void emitWorkoutArrivalIfCreated(userId, saved, "google-health").catch(
+        () => {},
+      );
       imported++;
     } catch (err) {
       getEvent()?.addWarning(`google-health: failed to upsert workout: ${err}`);
diff --git a/src/lib/jobs/data-arrival.ts b/src/lib/jobs/data-arrival.ts
new file mode 100644
index 000000000..67582c091
--- /dev/null
+++ b/src/lib/jobs/data-arrival.ts
@@ -0,0 +1,205 @@
+/**
+ * The `data-arrival` worker — the consumer end of the arrival spine.
+ *
+ * It runs once per SALIENT arrival (the seam classifier has already dropped
+ * every backfill), claims the day's reaction marker, and fans out to the
+ * deterministic reaction surfaces.
+ *
+ * Two properties are load-bearing and must survive every future edit:
+ *
+ * 1. **No provider import, structurally.** This module — and everything
+ *    reachable from it — imports no AI client and no provider resolver. The
+ *    spine cannot spend a token by construction; the only AI it can cause is
+ *    through explicitly budget-gated sub-queues that it merely ENQUEUES. A
+ *    module-graph test (`__tests__/data-arrival-provider-isolation.test.ts`)
+ *    fails if a provider module ever becomes reachable from here.
+ *
+ * 2. **A refusal is `skipped`, never `failed`.** Every business refusal — the
+ *    module is off, the day's marker already exists, a later slice's cap is
+ *    exhausted — returns a status. It does NOT throw. This is not a stylistic
+ *    choice: pg-boss retries a failed job, and retrying against a ceiling that
+ *    does not move until the local day rolls over is an unbounded loop. Only a
+ *    genuine transient fault (pool exhaustion, a dropped connection) is allowed
+ *    to escape and earn its two backed-off retries.
+ *
+ * The marker is claimed FIRST, before any fan-out. That ordering is deliberate:
+ * 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 means a
+ * downstream failure costs at most the retries of ONE job, never a growing
+ * queue.
+ */
+import type { Job } from "pg-boss";
+
+import { invalidateUserDashboardSnapshot } from "@/lib/cache/invalidate";
+import { annotate } from "@/lib/logging/context";
+import { withBackgroundEvent } from "@/lib/logging/background";
+
+import { DATA_ARRIVAL_QUEUE } from "@/lib/arrivals/emit-shared";
+import { isArrivalKind, type DataArrival } from "@/lib/arrivals/types";
+
+import { getWorkerPrisma, workerLog } from "./reminder/shared";
+
+export { DATA_ARRIVAL_QUEUE };
+
+/**
+ * What one arrival job did. `skipped` and `processed` are both SUCCESS — the
+ * distinction is observability, not control flow.
+ */
+export type ArrivalOutcome =
+  | { status: "skipped"; reason: string }
+  | { status: "processed"; actions: string[]; dedup: boolean };
+
+/**
+ * Claim the day's reaction row for this (user, kind, localDate).
+ *
+ * `createMany` + `skipDuplicates` compiles to INSERT … ON CONFLICT DO NOTHING,
+ * which is exactly the primitive wanted here and — unlike `upsert` — reports
+ * whether THIS caller won the race. A returned count of 1 means we own today's
+ * reaction for this kind; 0 means someone else already does.
+ */
+async function claimReaction(
+  prisma: ReturnType<typeof getWorkerPrisma>,
+  arrival: DataArrival,
+): Promise<{ claimed: boolean }> {
+  const occurredAt = new Date(arrival.occurredAt);
+
+  const inserted = await prisma.arrivalReaction.createMany({
+    data: [
+      {
+        userId: arrival.userId,
+        kind: arrival.kind,
+        localDate: arrival.localDate,
+        occurredAt,
+        refId: arrival.refId ?? null,
+      },
+    ],
+    skipDuplicates: true,
+  });
+
+  if (inserted.count > 0) return { claimed: true };
+
+  // Lost the claim — another arrival of this kind already landed today. Keep
+  // the marker's timestamp moving forward so the "just in" surface shows the
+  // NEWEST arrival rather than the first one, but never move it backwards (a
+  // slightly-out-of-order sync must not rewind the chip).
+  await prisma.arrivalReaction.updateMany({
+    where: {
+      userId: arrival.userId,
+      kind: arrival.kind,
+      localDate: arrival.localDate,
+      occurredAt: { lt: occurredAt },
+    },
+    data: { occurredAt },
+  });
+
+  return { claimed: false };
+}
+
+/**
+ * Process one arrival. Kind-dispatched, entirely deterministic, no provider
+ * call anywhere on the path.
+ */
+export async function runDataArrival(
+  prisma: ReturnType<typeof getWorkerPrisma>,
+  arrival: DataArrival,
+): Promise<ArrivalOutcome> {
+  if (!isArrivalKind(arrival.kind)) {
+    // A payload from a future version, or a hand-inserted row. Refusing is
+    // correct; failing would retry it twice for nothing.
+    return { status: "skipped", reason: "unknown_kind" };
+  }
+
+  const { claimed } = await claimReaction(prisma, arrival);
+  const actions: string[] = [claimed ? "claimed" : "marker_refreshed"];
+
+  switch (arrival.kind) {
+    case "sleep_night":
+      // Nothing extra. `morning-digest-refresh` already owns the
+      // provisional → final flip for the day, enqueued by the same seam that
+      // emitted this arrival; duplicating it here would double the work and
+      // race its marker. The row this worker just wrote is what powers the
+      // "just in" surface.
+      actions.push("sleep_marker_only");
+      break;
+
+    case "workout":
+      // DISPATCH POINT — the per-workout Activity Insight consumes this.
+      // It is deliberately NOT gated on `claimed`: two workouts in one day are
+      // two events and both deserve a paragraph, whereas the day-scoped marker
+      // can only be claimed once. Its own once-per-workout key, duration
+      // floor, hard daily cap, input hash and token-ledger reservation bound
+      // the spend — none of which belong in the spine.
+      actions.push("workout_pending_insight");
+      break;
+
+    case "weight":
+    case "blood_pressure":
+    case "labs_panel":
+      // Marker only. The Today surface reads the row directly.
+      actions.push("marker_only");
+      break;
+  }
+
+  if (claimed) {
+    // DISPATCH POINT — the day's single reaction line consumes this.
+    // Gated on `claimed` precisely BECAUSE the unique row is the throttle:
+    // there is no code path to a second generation for this kind today.
+    actions.push("line_pending");
+  }
+
+  // Best-effort and same-process only. `ServerCache` is per-process
+  // in-memory, so a worker eviction does not reach the web process; the short
+  // read TTLs plus the client's foreground poll are the real freshness floor.
+  // Called anyway because in a single-container deployment it IS the web
+  // process, and it costs nothing when it is not.
+  try {
+    invalidateUserDashboardSnapshot(arrival.userId);
+  } catch {
+    // A cache miss is never worth failing a job over.
+  }
+
+  return { status: "processed", actions, dedup: !claimed };
+}
+
+export async function handleDataArrival(
+  jobs: Job<DataArrival>[],
+): Promise<void> {
+  await withBackgroundEvent("job.data_arrival", async (evt) => {
+    for (const job of jobs) {
+      const arrival = job.data;
+      try {
+        const outcome = await runDataArrival(getWorkerPrisma(), arrival);
+
+        if (outcome.status === "skipped") {
+          annotate({
+            action: { name: `arrival.${arrival.kind}.skipped` },
+            meta: { reason: outcome.reason, source: arrival.source },
+          });
+          evt.addMeta("arrival", `${arrival.kind}:skipped:${outcome.reason}`);
+          continue;
+        }
+
+        annotate({
+          action: { name: `arrival.${arrival.kind}.processed` },
+          meta: {
+            actions: outcome.actions.join(","),
+            dedup: outcome.dedup,
+            source: arrival.source,
+            local_date: arrival.localDate,
+          },
+        });
+        evt.addMeta(
+          "arrival",
+          `${arrival.kind}:processed:${outcome.actions.join("|")}`,
+        );
+      } catch (err) {
+        // Reached only by a genuine transient fault — every business refusal
+        // returned a status above. Rethrow so the queue's two backed-off
+        // retries apply; the marker claim is idempotent, so a retry is safe.
+        workerLog("error", "[data-arrival] pass failed", err);
+        throw err;
+      }
+    }
+  });
+}
diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts
index 105367bdb..c936342c4 100644
--- a/src/lib/jobs/reminder/cleanup-handlers.ts
+++ b/src/lib/jobs/reminder/cleanup-handlers.ts
@@ -131,6 +131,40 @@ export async function handlePushAttemptCleanup(
   });
 }
 
+// v1.31.0 — daily prune for the data-arrival spine's reaction markers.
+//
+// Fourteen days rather than the ledgers' 90: a reaction marker is a
+// same-day surface ("just in", today's one generated line), so a row stops
+// being read the moment its local day rolls over. Two weeks is purely a
+// debugging margin — long enough to reconstruct what the spine reacted to
+// across a reported incident, short enough that a chatty account's markers
+// never accumulate. The `created_at` index makes the trailing-edge scan
+// cheap, exactly as it does for the push-attempt ledger above.
+const ARRIVAL_REACTION_RETENTION_DAYS = 14;
+
+export interface ArrivalReactionCleanupPayload {
+  triggeredAt: string;
+}
+
+export async function handleArrivalReactionCleanup(
+  jobs: Job<ArrivalReactionCleanupPayload>[],
+) {
+  void jobs;
+  await withBackgroundEvent("job.arrival_reaction_cleanup", async (evt) => {
+    const p = getWorkerPrisma();
+    try {
+      const cutoff = new Date();
+      cutoff.setUTCDate(cutoff.getUTCDate() - ARRIVAL_REACTION_RETENTION_DAYS);
+      const deleted = await p.arrivalReaction.deleteMany({
+        where: { createdAt: { lt: cutoff } },
+      });
+      evt.addMeta("arrival_reaction_cleanup_deleted", deleted.count);
+    } catch (err) {
+      evt.addWarning(`arrival-reaction-cleanup failed: ${err}`);
+    }
+  });
+}
+
 export interface MeasurementTombstoneCleanupPayload {
   triggeredAt: string;
 }
diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts
index c258a79ca..6575dec33 100644
--- a/src/lib/jobs/reminder/register-maintenance.ts
+++ b/src/lib/jobs/reminder/register-maintenance.ts
@@ -140,6 +140,8 @@ import {
   handleMoodReminderCleanup,
   PushAttemptCleanupPayload,
   handlePushAttemptCleanup,
+  ArrivalReactionCleanupPayload,
+  handleArrivalReactionCleanup,
   MeasurementTombstoneCleanupPayload,
   handleMeasurementTombstoneCleanup,
   handleRateLimitCleanup,
@@ -231,6 +233,13 @@ const PUSH_ATTEMPT_CLEANUP_QUEUE = "push-attempt-cleanup";
 
 const PUSH_ATTEMPT_CLEANUP_CRON = "35 3 * * *";
 
+// v1.31.0 — daily prune for the data-arrival spine's reaction markers.
+// Slots at 04:10, after the 03:xx retention block and the 04:00 MCP-token
+// prune, so the maintenance window stays in a readable order.
+const ARRIVAL_REACTION_CLEANUP_QUEUE = "arrival-reaction-cleanup";
+
+const ARRIVAL_REACTION_CLEANUP_CRON = "10 4 * * *";
+
 const MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE = "measurement-tombstone-cleanup";
 
 const MEASUREMENT_TOMBSTONE_CLEANUP_CRON = "40 3 * * *";
@@ -307,6 +316,9 @@ const allQueues = [
   // v1.4.49 — push-attempt ledger cleanup. The daily schedule below would
   // silently no-op without this entry.
   PUSH_ATTEMPT_CLEANUP_QUEUE,
+  // v1.31.0 — data-arrival reaction-marker prune. Without this entry the
+  // daily schedule silently no-ops and the markers accumulate forever.
+  ARRIVAL_REACTION_CLEANUP_QUEUE,
   // v1.7.0 — soft-deleted measurement tombstone prune. Without this entry
   // the daily schedule silently no-ops and pruned-past-retention tombstones
   // pile up forever.
@@ -419,6 +431,8 @@ const schedules: ScheduleEntry[] = [
   [INTAKE_SLOT_DEDUP_QUEUE, INTAKE_SLOT_DEDUP_CRON],
   // v1.4.49 — daily 03:35 Europe/Berlin prune for push_attempts.
   [PUSH_ATTEMPT_CLEANUP_QUEUE, PUSH_ATTEMPT_CLEANUP_CRON],
+  // v1.31.0 — daily 04:10 Europe/Berlin prune for arrival_reactions.
+  [ARRIVAL_REACTION_CLEANUP_QUEUE, ARRIVAL_REACTION_CLEANUP_CRON],
   // v1.7.0 — daily 03:40 Europe/Berlin prune for expired measurement
   // tombstones.
   [MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, MEASUREMENT_TOMBSTONE_CLEANUP_CRON],
@@ -597,6 +611,14 @@ export async function registerMaintenanceQueues(
     { localConcurrency: 1 },
     handlePushAttemptCleanup,
   );
+  // v1.31.0 — daily prune of the data-arrival reaction markers. Single-flight
+  // like every other cleanup queue: two ticks racing the same DELETE is
+  // wasted work.
+  await boss.work<ArrivalReactionCleanupPayload>(
+    ARRIVAL_REACTION_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleArrivalReactionCleanup,
+  );
   // v1.7.0 — daily prune of expired measurement tombstones. Single-flight
   // like every other cleanup queue.
   await boss.work<MeasurementTombstoneCleanupPayload>(
diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts
index 9b49e8e34..c8cab6683 100644
--- a/src/lib/jobs/reminder/register-status.ts
+++ b/src/lib/jobs/reminder/register-status.ts
@@ -29,6 +29,8 @@ import {
   runMorningDigestRefresh,
   type MorningDigestRefreshPayload,
 } from "@/lib/jobs/morning-digest-refresh";
+import { DATA_ARRIVAL_QUEUE, handleDataArrival } from "@/lib/jobs/data-arrival";
+import type { DataArrival } from "@/lib/arrivals/types";
 import {
   DAILY_BRIEFING_QUEUE,
   DAILY_BRIEFING_CRON,
@@ -171,6 +173,12 @@ const allQueues = [
   // send-only queue driven by sleep ingest; without this entry pg-boss never
   // provisions the queue and every enqueue silently drops.
   MORNING_DIGEST_REFRESH_QUEUE,
+  // v1.31.0 — the data-arrival spine. Every salient ingest emits here; the
+  // worker claims the day's reaction marker and fans out to the reaction
+  // surfaces. No cron schedule — a send-only queue driven by ingest; without
+  // this entry pg-boss never provisions the queue and every emit silently
+  // drops (the v1.4.37 dead-queue class).
+  DATA_ARRIVAL_QUEUE,
   // v1.10.0 — computed scores (WX-C / WX-E). Nightly Recovery / Stress /
   // Strain compute + store. The queue MUST be registered here or pg-boss
   // never provisions it and the nightly schedule silently never fires.
@@ -275,9 +283,30 @@ const schedules: ScheduleEntry[] = [
  * differently-shaped constraint over a working one for no gain, so they are
  * left alone.
  *
- * That leaves exactly one queue here whose key is inert today.
+ * That leaves exactly two queues here whose keys would otherwise be inert.
  */
 const queuePolicies: QueuePolicyTable = {
+  // v1.31.0 — the arrival spine's day-scoped keys
+  // (`arrival:<user>:<kind>:<localDate>`) carry the SAME requirement as the
+  // morning refresh below, for the same reason: a user's local date cannot be
+  // expressed as a wall-clock `singletonSeconds` window, so the date lives in
+  // the key and the policy has to honour it. Under `standard` the key
+  // coalesces nothing at all, and a device posting a batch per minute would
+  // queue a job per batch instead of one per day.
+  //
+  // `exclusive` rather than `short` because the handler is self-converging:
+  // it claims a durable unique row, so a second run while the first is active
+  // is pure duplicated work, and the next ingest re-emits anyway.
+  //
+  // The `workout` kind is the deliberate exception — it is keyed per workout
+  // id and DOES pass `singletonSeconds`, so it is constrained by pg-boss's
+  // `job_i4` index regardless of this policy. Both shapes coexist on one
+  // queue safely: they populate different columns.
+  [DATA_ARRIVAL_QUEUE]: {
+    policy: "exclusive",
+    reason:
+      "Day-scoped arrival keys carry the user's local date; under standard they coalesce nothing and a chatty ingest queues a job per batch. The worker claims a durable unique row, so a concurrent second run is pure duplicated work.",
+  },
   // The sleep-arrival trigger has seven write seams that can each enqueue for
   // the same user and the same local date. The key is
   // `morning-refresh:<user>:<localDate>`, and `exclusive` turns that into a
@@ -494,6 +523,15 @@ export async function registerStatusQueues(
       });
     },
   );
+  // v1.31.0 — the data-arrival spine. Single-flight: the worker's whole job is
+  // to claim a durable unique row, so two slots would only race each other for
+  // a claim exactly one of them can win. It makes no provider call — a
+  // module-graph test pins that — so one slot costs nothing in throughput.
+  await boss.work<DataArrival>(
+    DATA_ARRIVAL_QUEUE,
+    { localConcurrency: 1 },
+    handleDataArrival,
+  );
   // S5 — daily-briefing fallback slot. Single-flight; the `push_attempts`
   // frequency cap makes an overlapping tick a no-op (a second same-day attempt
   // is suppressed), and the tick only loads a digest for opted-in users at
diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts
index 54199cc01..8927ab4c5 100644
--- a/src/lib/measurements/import-apple-health-export.ts
+++ b/src/lib/measurements/import-apple-health-export.ts
@@ -40,6 +40,7 @@ import {
   canonicalDailyTimestamp,
 } from "@/lib/measurements/drain-per-sample-cumulative";
 import { resolveHkWorkoutSportType } from "@/lib/measurements/hk-workout-activity-type-map";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import { validateMeasurementRange } from "@/lib/validations/measurement";
 import {
   CycleImportAccumulator,
@@ -496,10 +497,18 @@ export async function streamParseExportXml(
     });
     const existingKey = new Set(existingRows.map((r) => r.externalId));
 
+    // v1.31.0 — data-arrival spine. An export is the archetypal backfill, so
+    // this tracks only the NEWEST row the chunk CREATED and emits once per
+    // flushed chunk rather than once per row. A ten-year export therefore
+    // costs a handful of classifier calls that all resolve to `backfill`,
+    // instead of thousands. A genuinely fresh export still reacts, because a
+    // same-day workout is by definition the newest one it created.
+    let newestCreated: { id: string; startedAt: Date } | null = null;
+
     for (const row of chunk) {
       const rowStart = Date.now();
       const exists = existingKey.has(row.externalId);
-      await prisma.workout.upsert({
+      const saved = await prisma.workout.upsert({
         where: {
           userId_source_externalId: {
             userId,
@@ -530,12 +539,33 @@ export async function streamParseExportXml(
           externalSourceVersion: row.externalSourceVersion,
           metadata: row.metadata ?? undefined,
         },
+        select: { id: true, startedAt: true },
       });
       workouts.durationMs += Date.now() - rowStart;
-      if (exists) workouts.updated += 1;
-      else workouts.inserted += 1;
+      if (exists) {
+        workouts.updated += 1;
+      } else {
+        workouts.inserted += 1;
+        if (
+          !newestCreated ||
+          saved.startedAt.getTime() > newestCreated.startedAt.getTime()
+        ) {
+          newestCreated = saved;
+        }
+      }
       rowsUpserted += 1;
     }
+
+    if (newestCreated) {
+      void emitDataArrival({
+        userId,
+        kind: "workout",
+        newestSampleAt: newestCreated.startedAt,
+        insertedCount: 1,
+        refId: newestCreated.id,
+        source: "apple_export",
+      }).catch(() => {});
+    }
   };
 
   const flushCumulativeBuckets = async (): Promise<void> => {
diff --git a/src/lib/strava/sync.ts b/src/lib/strava/sync.ts
index 6a7cdc2d2..0ca94d7e3 100644
--- a/src/lib/strava/sync.ts
+++ b/src/lib/strava/sync.ts
@@ -24,6 +24,7 @@
  * cron tick resumes from the cursor.
  */
 import { prisma } from "@/lib/db";
+import { emitWorkoutArrivalIfCreated } from "@/lib/arrivals/workout-emit";
 import { annotate, getEvent } from "@/lib/logging/context";
 import {
   recordSyncFailure,
@@ -262,7 +263,7 @@ export async function upsertStravaWorkouts(
       metadata: r.metadata,
     };
     try {
-      await prisma.workout.upsert({
+      const saved = await prisma.workout.upsert({
         where: {
           userId_source_externalId: {
             userId,
@@ -277,7 +278,16 @@ export async function upsertStravaWorkouts(
           ...data,
         },
         update: data,
+        select: {
+          id: true,
+          startedAt: true,
+          createdAt: true,
+          updatedAt: true,
+        },
       });
+      // v1.31.0 — data-arrival spine. Only a genuinely NEW workout reacts; a
+      // re-sync of an already-stored session is not news.
+      void emitWorkoutArrivalIfCreated(userId, saved, "strava").catch(() => {});
       imported += 1;
     } catch (err) {
       getEvent()?.addWarning(`strava: failed to upsert workout: ${err}`);
diff --git a/src/lib/whoop/sync-workout.ts b/src/lib/whoop/sync-workout.ts
index 14cdfea06..303da4ea7 100644
--- a/src/lib/whoop/sync-workout.ts
+++ b/src/lib/whoop/sync-workout.ts
@@ -35,6 +35,7 @@ import {
 } from "./sync";
 import { mapWhoopSportType } from "./sport-map";
 import { prisma } from "@/lib/db";
+import { emitWorkoutArrivalIfCreated } from "@/lib/arrivals/workout-emit";
 import { getEvent } from "@/lib/logging/context";
 import type { Prisma } from "@/generated/prisma/client";
 
@@ -162,7 +163,7 @@ export async function upsertWhoopWorkout(
   };
 
   try {
-    await prisma.workout.upsert({
+    const saved = await prisma.workout.upsert({
       where: {
         userId_source_externalId: {
           userId,
@@ -177,7 +178,16 @@ export async function upsertWhoopWorkout(
         ...row,
       },
       update: row,
+      select: {
+        id: true,
+        startedAt: true,
+        createdAt: true,
+        updatedAt: true,
+      },
     });
+    // v1.31.0 — data-arrival spine. Only a genuinely NEW workout reacts;
+    // WHOOP re-posts recent workouts on every poll and an update is not news.
+    void emitWorkoutArrivalIfCreated(userId, saved, "whoop").catch(() => {});
     return 1;
   } catch (err) {
     getEvent()?.addWarning(`WHOOP: failed to upsert workout: ${err}`);