diff --git a/docs/Integrations/ownerrez/api-markdown.md b/docs/Integrations/ownerrez/api-markdown.md
index 86273f5c..26f5faa9 100644
--- a/docs/Integrations/ownerrez/api-markdown.md
+++ b/docs/Integrations/ownerrez/api-markdown.md
@@ -14,6 +14,49 @@
 - Errors return JSON with a `messages` array of human-readable strings. Status codes: 400 = bad request / validation, 401 = authentication, 403 = permission, 404 = not found.
 - Prefer the current API (v2). Legacy versions remain documented below but should not be used for new integrations.
 
+## Verified contract details (2026-08-13)
+
+Pulled from `https://api.ownerrez.com/openapi/v2.json` because the orientation
+notes above have no per-endpoint parameter list, and guessing filled the gap
+wrongly once already. Re-verify against the spec before trusting any of this.
+
+### Paging — every list endpoint
+
+The response wrapper (`PageableEnumerableOf*`, used by all 22 list endpoints) is:
+
+```
+{ items: [...], limit: 100, offset: 0, next_page_url: "https://..." | null }
+```
+
+- `next_page_url` is the continuation mechanism. **Null means the collection is
+  finished** — including on a full page. Absent is a different answer from null.
+- There is **no `next_page_token` and no `total_count`.** `lib/integrations/types.ts`
+  declared both; neither string occurs anywhere in the spec. The pager read
+  `next_page_token`, got `undefined`, and stopped after one page — and since it
+  also sent no `limit`, that one page was OwnerRez's default of 20. One
+  production org's first sync imported exactly 20 bookings as a result.
+- `limit` (default 20, max 100) and `offset` are **declared on zero operations**
+  in the spec even though the response echoes both back. They work, but do not
+  assume a requested `limit` was honored — prefer `next_page_url`, and compare
+  short-page checks against the server-reported `page.limit`, not the one you asked for.
+
+### GET /v2/bookings query parameters
+
+| Param | Meaning |
+|---|---|
+| `property_ids` | comma-separated integers |
+| `from` | bookings that **depart on or after** this date (property timezone) |
+| `to` | bookings that **arrive on or before** this date (property timezone) |
+| `since_utc` | created or changed since (UTC) — a **modification-time cursor, not a stay-date filter** |
+| `status` | `Active` \| `Pending` \| `Canceled` |
+| `include_guest`, `include_charges`, `include_tags`, `include_fields`, `include_door_codes`, `include_cancellation_policy`, `include_agreements` | booleans |
+
+`from`/`to` together are an interval **overlap** filter, not containment: a stay
+straddling a window boundary is returned by both adjacent windows. That is what
+makes a progressive historical backfill possible (walk `to` backwards in fixed
+steps); the boundary duplicates are harmless because bookings upsert on the
+OwnerRez booking id.
+
 ## Documentation
 
 - [OpenAPI 3.0 spec](/openapi/v2.json): the full machine-readable contract for v2.
diff --git a/lib/inngest/functions/ownerrez/incremental-sync.ts b/lib/inngest/functions/ownerrez/incremental-sync.ts
index 7baa3f8a..1c99a95e 100644
--- a/lib/inngest/functions/ownerrez/incremental-sync.ts
+++ b/lib/inngest/functions/ownerrez/incremental-sync.ts
@@ -53,6 +53,11 @@ import type { GetStepTools }            from 'inngest'
 import { createServiceClient }          from '@/lib/supabase/server'
 import { fetchTurnoverCreatedEvents } from '@/lib/inngest/turnover-created-events'
 import { OwnerRezApiClient }  from '@/lib/integrations/providers/ownerrez-api'
+import {
+  planBackfillWindow,
+  advanceBackfill,
+  readBackfillState,
+} from '@/lib/integrations/providers/ownerrez-backfill'
 import { getRedis, upstashConfigured } from '@/lib/redis'
 import { RateLimitError, TokenRevokedError, translateSyncError } from '@/lib/integrations/types'
 import { logAuditEvent }                from '@/lib/audit'
@@ -996,10 +1001,172 @@ export const ownerRezConnectionSync = inngest.createFunction(
     await runPostSyncFanOut(step, { orgId, userId, affectedIds, logger })
 
     const synced = Boolean(syncResult && 'affectedPropertyIds' in syncResult)
-    return { connectionId, synced }
+
+    // ── Historical backfill: one older stay-date window per run ──────────────
+    //
+    // The initial sync deliberately takes only a recent window so a new PM is
+    // not kept waiting on years of history (see ownerrez-backfill.ts). This
+    // walks the rest backwards, one window per incremental run, until it
+    // reaches the horizon and stops for good.
+    //
+    // Gated on `synced`: if the live sync just skipped or failed, the
+    // connection is inactive, rate limited or degraded, and adding a second
+    // multi-page fetch on top of that helps nobody. It simply waits an hour.
+    const backfill = await runBackfillPhase(step, { connectionId, userId, orgId, synced, logger })
+
+    return { connectionId, synced, backfill }
   }
 )
 
+interface BackfillOutcome {
+  outcome:               string
+  bookingsToPostRevenue: SyncSuccess['bookingsToPostRevenue']
+}
+
+/**
+ * The backfill phase: claim one historical window, then post revenue for
+ * whatever it imported. Structured like runPostSyncFanOut — step tooling stays
+ * at the function's top level, never inside a step.run callback.
+ *
+ * Revenue for backfilled stays IS posted; historical owner P&L is the reason to
+ * import them at all. Turnovers deliberately are NOT — those are units of WORK,
+ * and generateTurnoversForProperty's own history floor is what stops a two-year
+ * import from manufacturing retroactive cleaning jobs. Backfilled properties
+ * therefore never enter runPostSyncFanOut.
+ */
+async function runBackfillPhase(
+  step: GetStepTools<typeof inngest>,
+  ctx:  { connectionId: string; userId: string; orgId: string | null
+          synced: boolean; logger: SyncLogger }
+): Promise<string> {
+  const backfill = await step.run('backfill-history', () => runHistoricalBackfill(ctx))
+
+  // Optional-chained for the same reason `syncResult &&` is, at the call site:
+  // a step's result is only as present as the step that produced it.
+  const revenue = backfill?.bookingsToPostRevenue ?? []
+
+  if (revenue.length > 0 && ctx.orgId) {
+    await step.sendEvent(
+      'post-backfill-revenue',
+      revenue.map((b) => ({
+        name: 'booking/confirmed' as const,
+        data: {
+          booking_id:          b.bookingId,
+          property_id:         b.propertyId,
+          org_id:              ctx.orgId as string,
+          source:              'ownerrez' as const,
+          actual_total_amount: b.actualTotalAmount,
+        },
+      }))
+    )
+  }
+
+  return backfill?.outcome ?? 'skipped'
+}
+
+const NO_BACKFILL = (outcome: string): BackfillOutcome =>
+  ({ outcome, bookingsToPostRevenue: [] })
+
+/**
+ * Claims and persists ONE historical stay-date window for a connection.
+ *
+ * Never throws. The backfill is strictly best-effort catch-up work running
+ * behind a live sync that already succeeded — letting a rate limit or a bad
+ * page here fail the whole run would turn "we did not fetch some 2024 bookings
+ * this hour" into "this connection's hourly sync is red", and Inngest would
+ * retry the entire function including the live sync that was already fine.
+ * Progress only advances on success, so a swallowed failure simply retries the
+ * same window next hour.
+ */
+async function runHistoricalBackfill(params: {
+  connectionId: string
+  userId:       string
+  orgId:        string | null
+  synced:       boolean
+  logger:       SyncLogger
+}): Promise<BackfillOutcome> {
+  const { connectionId, userId, orgId, synced, logger } = params
+
+  if (!synced || !orgId) return NO_BACKFILL('skipped_not_synced')
+
+  try {
+    const supabase = createServiceClient({ system: 'inngest:ownerrez-backfill' })
+
+    const connRes = await supabase
+      .from('integration_connections')
+      .select('id, user_id, org_id, external_user_id, metadata, status')
+      .eq('id', connectionId)
+      .maybeSingle()
+
+    const conn = unwrap(connRes, {
+      site:  'inngest.ownerrez-connection-sync.backfill-reload',
+      orgId: orgId,
+    })
+
+    // org_id first so the optional chain covers the null-connection case too:
+    // a truthy conn?.org_id already proves conn itself is non-null, which is
+    // what lets conn.status be read plainly on the next clause.
+    if (!conn?.org_id || conn.status !== 'active') {
+      return NO_BACKFILL('skipped_inactive')
+    }
+
+    const window = planBackfillWindow(readBackfillState(conn.metadata), new Date())
+    if (!window) return NO_BACKFILL('complete')
+
+    // property_ids is required alongside the date bounds for the same reason
+    // the live sync needs it: OwnerRez wants at least one scoping parameter,
+    // and this keeps the window to properties FieldStay actually knows about.
+    const propertyIds = await loadConnectedPropertyIds(supabase, conn.org_id)
+    if (!propertyIds.length) return NO_BACKFILL('skipped_no_properties')
+
+    const bookings = await new OwnerRezApiClient(userId).getBookings({
+      propertyIds,
+      from:         window.from,
+      to:           window.to,
+      includeGuest: true,
+    })
+
+    const activeConn: ActiveConnection = { ...conn, org_id: conn.org_id }
+    const persisted = await persistBookings(supabase, activeConn, bookings, logger)
+
+    // Property lookup failed. Do NOT advance the cursor — the same reasoning as
+    // the live sync's bail-out, plus advancing here would skip this window
+    // permanently since nothing revisits it.
+    if (!persisted) return NO_BACKFILL('persist_failed')
+
+    const advanced = advanceBackfill(window, new Date())
+    await mergeIntegrationConnectionMetadata({
+      userId,
+      providerId: PROVIDER,
+      patch: {
+        backfill_oldest_covered: advanced.oldestCovered,
+        backfill_complete:       advanced.complete,
+      },
+    })
+
+    logger.info(
+      `[OwnerRez:${userId}] backfilled ${window.from}..${window.to} — ` +
+      `${bookings.length} booking(s)${advanced.complete ? ', walk complete' : ''}`
+    )
+
+    return {
+      outcome:               advanced.complete ? 'window_done_complete' : 'window_done',
+      bookingsToPostRevenue: persisted.bookingsToPostRevenue,
+    }
+  } catch (err) {
+    logger.warn(
+      `[OwnerRez:${userId}] historical backfill failed, will retry next run: ` +
+      (err instanceof Error ? err.message : String(err))
+    )
+    reportError(err, {
+      site:  'inngest.ownerrez-connection-sync.backfill-history',
+      orgId: orgId ?? undefined,
+      extra: { connection_id: connectionId },
+    })
+    return NO_BACKFILL('failed')
+  }
+}
+
 /**
  * Fire a PM notification about a broken connection — throttled to once per
  * 4 hours per connection via an org_milestones timestamp. Shared by the
diff --git a/lib/inngest/functions/ownerrez/initial-sync.ts b/lib/inngest/functions/ownerrez/initial-sync.ts
index 7eafd29c..0d7f7b9a 100644
--- a/lib/inngest/functions/ownerrez/initial-sync.ts
+++ b/lib/inngest/functions/ownerrez/initial-sync.ts
@@ -23,6 +23,7 @@ import {
   selectOwnerRezBookingsToPostRevenue,
 } from '@/lib/integrations/providers/ownerrez'
 import { upsertBookingsReturningIds } from './upsert-bookings'
+import { initialHistoryFrom } from '@/lib/integrations/providers/ownerrez-backfill'
 import { logAuditEvent }        from '@/lib/audit'
 import {
   applyMasterChecklistToProperty,
@@ -52,6 +53,13 @@ import { fetchAllRows } from '@/lib/inngest/paginate'
 import { tryUnwrap, unwrapList } from '@/lib/supabase/unwrap'
 const PROVIDER = 'ownerrez'
 
+/**
+ * Bookings per booking/confirmed sendEvent step. Sized to keep one step's
+ * payload comfortably small while making the step count independent of
+ * portfolio size — see the batching note at the call site.
+ */
+const REVENUE_EVENT_CHUNK = 200
+
 async function writeSyncCount(
   user_id: string,
   field: 'properties_found' | 'bookings_found',
@@ -94,6 +102,27 @@ export const ownerRezInitialSync = inngest.createFunction(
     id:      'ownerrez-initial-sync',
     name:    'OwnerRez Initial Sync',
     retries: 3,
+    // Two caps, and they answer different questions.
+    //
+    // The unkeyed one is capacity: this is the heaviest OwnerRez consumer we
+    // have — it paginates every property, listing and booking for a brand-new
+    // account — and the 300-request/5-minute OwnerRez budget is shared by every
+    // tenant on the same deployment IP. Left uncapped, several signups landing
+    // together would spend that budget on each other and take the incremental
+    // syncs down with them. 3 matches ownerRezConnectionSync's cap for the
+    // same reason.
+    //
+    // The keyed one is correctness: never two initial syncs for one connection
+    // at once. This step chain seeds checklists, generates turnovers and seeds
+    // assets from amenities, and not all of that is safe to interleave with
+    // itself. A reconnect, a double-clicked Connect button, or a re-fired
+    // event would otherwise race. Keyed concurrency QUEUES the second run
+    // rather than dropping it, which is what a genuine reconnect wants —
+    // `idempotency` would silently discard it instead.
+    concurrency: [
+      { limit: 3 },
+      { limit: 1, key: 'event.data.user_id' },
+    ],
   },
   { event: 'integration/ownerrez.connected' as const },
   async ({ event, step, logger }) => {
@@ -515,6 +544,7 @@ export const ownerRezInitialSync = inngest.createFunction(
             cursor: new Date().toISOString(), count: 0,
             affectedPropertyIds: [] as string[],
             bookingsToPostRevenue: [] as { bookingId: string; propertyId: string; actualTotalAmount: number | null }[],
+            historyFrom: initialHistoryFrom(new Date()),
           }
         }
 
@@ -525,8 +555,28 @@ export const ownerRezInitialSync = inngest.createFunction(
 
         let bookings: OwnerRezBooking[]
 
+        // BOUNDED to a recent window, plus everything upcoming.
+        //
+        // This call used to pass no date bounds at all, i.e. "every booking
+        // this account has ever had". That was harmless only because the pager
+        // was broken and stopped at 20 records; with pagination fixed it became
+        // a request to page through a portfolio's entire history — thousands of
+        // rows against a request budget shared by every tenant — before the PM
+        // sees anything. Older history is walked backwards afterwards, one
+        // window per incremental sync (ownerrez-backfill.ts).
+        //
+        // `from` with no `to` is the important shape: `from` means "departs on
+        // or after", so every FUTURE booking is still included. Bounding the
+        // upper end would drop upcoming stays, which are the whole point of a
+        // first sync.
+        const historyFrom = initialHistoryFrom(new Date(fetchStartedAt))
+
         try {
-          bookings = await client.getBookings({ propertyIds: fetchPropsResult.ids, includeGuest: true })
+          bookings = await client.getBookings({
+            propertyIds:  fetchPropsResult.ids,
+            from:         historyFrom,
+            includeGuest: true,
+          })
         } catch (err) {
           if (err instanceof RateLimitError) {
             throw err
@@ -601,7 +651,7 @@ export const ownerRezInitialSync = inngest.createFunction(
           reportError(countErr, { site: 'inngest.ownerrez-initial-sync.fetch-bookings' })
         }
 
-        return { cursor: fetchStartedAt, count: bookings.length, affectedPropertyIds, bookingsToPostRevenue }  // MEDIUM-3: pre-fetch timestamp
+        return { cursor: fetchStartedAt, count: bookings.length, affectedPropertyIds, bookingsToPostRevenue, historyFrom }  // MEDIUM-3: pre-fetch timestamp
       })
 
       // ── Post booking revenue for newly-confirmed guest-stay bookings ───────
@@ -612,17 +662,30 @@ export const ownerRezInitialSync = inngest.createFunction(
       // 2026-07-15; booking-events.ts's handleBookingConfirmed still falls
       // back to the avg_nightly_rate estimate whenever this is null (e.g. a
       // booking whose charges genuinely didn't resolve to a positive total).
-      for (const b of fetchBookingsResult.bookingsToPostRevenue) {
-        await step.sendEvent(`post-booking-revenue-${b.bookingId}`, {
-          name: 'booking/confirmed' as const,
-          data: {
-            booking_id:          b.bookingId,
-            property_id:         b.propertyId,
-            org_id,
-            source:              'ownerrez' as const,
-            actual_total_amount: b.actualTotalAmount,
-          },
-        })
+      // BATCHED, one step per chunk rather than one step per booking.
+      //
+      // This was `for (const b of ...) await step.sendEvent(...)`, which makes
+      // a distinct Inngest step — with its own memoized state carried for the
+      // rest of the run — for every single booking. At 20 bookings that was
+      // invisible; bounded to a 90-day window it is hundreds, and unbounded it
+      // would have been thousands once pagination started working. step.sendEvent
+      // takes an array, so a chunk costs one step regardless of its size.
+      const revenueEvents = fetchBookingsResult.bookingsToPostRevenue.map((b) => ({
+        name: 'booking/confirmed' as const,
+        data: {
+          booking_id:          b.bookingId,
+          property_id:         b.propertyId,
+          org_id,
+          source:              'ownerrez' as const,
+          actual_total_amount: b.actualTotalAmount,
+        },
+      }))
+
+      for (let i = 0; i < revenueEvents.length; i += REVENUE_EVENT_CHUNK) {
+        const chunk = revenueEvents.slice(i, i + REVENUE_EVENT_CHUNK)
+        // Chunk index, not booking id: the step id must stay stable across
+        // retries, and it does because the source list is memoized upstream.
+        await step.sendEvent(`post-booking-revenue-${i / REVENUE_EVENT_CHUNK}`, chunk)
       }
 
       // ── Step 3: Update sync metadata ────────────────────────────────────────
@@ -639,6 +702,13 @@ export const ownerRezInitialSync = inngest.createFunction(
               last_sync_status: 'success',
               last_sync_error:  null,
               last_sync_count:  fetchBookingsResult.count,
+              // Seeds the historical backfill walk at the exact lower edge of
+              // the window this run actually fetched — NOT at "90 days before
+              // whenever the first backfill happens to run". Recomputing it
+              // later would open a gap the width of the delay between the two,
+              // and nothing ever revisits a skipped window.
+              backfill_oldest_covered: fetchBookingsResult.historyFrom,
+              backfill_complete:       false,
             },
           })
         } catch (err) {
diff --git a/lib/integrations/providers/ownerrez-api.ts b/lib/integrations/providers/ownerrez-api.ts
index f8a6ed87..8a09a5cd 100644
--- a/lib/integrations/providers/ownerrez-api.ts
+++ b/lib/integrations/providers/ownerrez-api.ts
@@ -30,6 +30,16 @@ import { getRedis, upstashConfigured } from '@/lib/redis'
 const BASE_URL   = 'https://api.ownerrez.com'
 const PROVIDER   = 'ownerrez'
 
+// OwnerRez list endpoints default to 20 records per page and cap at 100.
+// Sending nothing takes the default, which is how one production org's first
+// sync imported exactly 20 bookings and stopped.
+const PAGE_SIZE = 100
+
+// A loop guard, not a coverage ceiling: 1000 × 100 = 100,000 records, well past
+// any real account, and the shared 270-request/5-min IP budget throws long
+// before this does. Exceeding it means pagination is not terminating.
+const MAX_PAGES = 1000
+
 // ── Shared IP rate-limit budget tracker ──────────────────────────────────────
 //
 // OwnerRez limits 300 requests per 5-minute rolling window per IP address —
@@ -104,6 +114,28 @@ export class OwnerRezApiClient {
     path: string,
     params?: Record<string, string | number | undefined>,
     options?: { method?: string; body?: string }
+  ): Promise<T> {
+    const url = new URL(`${BASE_URL}${path}`)
+    if (params) {
+      for (const [k, v] of Object.entries(params)) {
+        if (v !== undefined) url.searchParams.set(k, String(v))
+      }
+    }
+    return this.fetchUrl<T>(url, path, options)
+  }
+
+  /**
+   * The transport half of fetch(), split out so fetchAllPages can follow the
+   * absolute `next_page_url` OwnerRez returns without rebuilding it from parts.
+   *
+   * `url` MUST already be origin-checked by the caller — see assertOwnerRezUrl.
+   * `label` is the path used in log/error messages only; it never affects the
+   * request, and exists so a followed page URL still reports as its endpoint.
+   */
+  private async fetchUrl<T>(
+    url: URL,
+    label: string,
+    options?: { method?: string; body?: string }
   ): Promise<T> {
     // HIGH-2: check shared IP budget before making the request.
     // Throws RateLimitError proactively at 270/300 to prevent exhausting the pool
@@ -119,12 +151,7 @@ export class OwnerRezApiClient {
       throw new TokenRevokedError(this.userId)
     }
 
-    const url = new URL(`${BASE_URL}${path}`)
-    if (params) {
-      for (const [k, v] of Object.entries(params)) {
-        if (v !== undefined) url.searchParams.set(k, String(v))
-      }
-    }
+    const path = label
 
     const res = await globalThis.fetch(url.toString(), {
       method:  options?.method ?? 'GET',
@@ -231,29 +258,108 @@ export class OwnerRezApiClient {
     params?: Record<string, string | number | undefined>
   ): Promise<T[]> {
     const results: T[] = []
-    let nextPageToken: string | null | undefined = undefined
     let pageCount = 0
-    const MAX_PAGES = 200  // 200 × 100 items = 20,000 results — generous ceiling
 
-    do {
+    // First page: BASE_URL + path + caller params, plus our page size. Every
+    // page after it comes from the server's own next_page_url.
+    let url: URL | null = new URL(`${BASE_URL}${path}`)
+    if (params) {
+      for (const [k, v] of Object.entries(params)) {
+        if (v !== undefined) url.searchParams.set(k, String(v))
+      }
+    }
+    if (!url.searchParams.has('limit')) {
+      url.searchParams.set('limit', String(PAGE_SIZE))
+    }
+
+    while (url) {
       pageCount++
       if (pageCount > MAX_PAGES) {
-        console.error(`[OwnerRez] fetchAllPages: exceeded ${MAX_PAGES} pages — aborting to prevent infinite loop`)
-        break
+        // THROW, never return what we have. Returning a partial list as if it
+        // were complete is the same silent-truncation failure this function was
+        // fixed for — callers upsert the result and treat anything absent from
+        // it as deleted, so a quiet short read is worse than a loud failure.
+        throw new Error(
+          `[OwnerRez:${this.userId}] ${path}: pagination exceeded ${MAX_PAGES} pages ` +
+          `(${results.length} records) — refusing to return a partial result`
+        )
       }
 
-      const pageParams = { ...params } as Record<string, string | number | undefined>
-      if (nextPageToken) pageParams['page_token'] = nextPageToken
-
-      const page = await this.fetch<OwnerRezPagedResponse<T>>(path, pageParams)
+      const page: OwnerRezPagedResponse<T> = await this.fetchUrl<OwnerRezPagedResponse<T>>(url, path)
       const items = Array.isArray(page?.items) ? page.items : []
       results.push(...items)
-      nextPageToken = page?.next_page_token ?? null
-    } while (nextPageToken)
+
+      url = this.nextPageUrl(page, url, items.length, path)
+    }
 
     return results
   }
 
+  /**
+   * Resolves the next page URL, or null when the collection is exhausted.
+   *
+   * `next_page_url` is OwnerRez's documented mechanism and is authoritative
+   * when present — critically, it works even if `limit` is ignored, which
+   * matters because OwnerRez's OpenAPI spec declares `limit`/`offset` on ZERO
+   * operations even though the response echoes both back. If we trusted our
+   * requested page size instead, a server that quietly capped us at 20 while
+   * we asked for 100 would look like a short final page and we would stop
+   * early — the exact bug this replaces, in a new disguise.
+   *
+   * The offset fallback covers the other direction: a response that omits
+   * next_page_url but is clearly full. Over-fetching one duplicate page is
+   * harmless (every caller upserts by OwnerRez id); stopping early is not.
+   */
+  private nextPageUrl<T>(
+    page: OwnerRezPagedResponse<T>,
+    current: URL,
+    received: number,
+    path: string
+  ): URL | null {
+    // ABSENT and explicitly NULL are different answers, and collapsing them is
+    // a bug (caught by this function's own test): OwnerRez documents null as
+    // "there are no more pages", so a null on a FULL page still ends the
+    // collection. Only a field that is missing entirely means "this server
+    // doesn't tell me", which is what the offset fallback below is for.
+    if (page?.next_page_url !== undefined) {
+      return page.next_page_url ? this.assertOwnerRezUrl(page.next_page_url, path) : null
+    }
+
+    // No continuation field at all: only keep going if the page came back full.
+    // Note the `?? PAGE_SIZE` sits INSIDE Number() — Number(null) is 0, which
+    // ?? would happily accept, and a pageSize of 0 makes `received < pageSize`
+    // permanently false and the loop unbounded.
+    const requested = Number(current.searchParams.get('limit') ?? PAGE_SIZE)
+    const pageSize  = page?.limit ?? (requested > 0 ? requested : PAGE_SIZE)
+    if (received === 0 || received < pageSize) return null
+
+    const offset = (page?.offset ?? Number(current.searchParams.get('offset') ?? 0)) + received
+    const next = new URL(current.toString())
+    next.searchParams.set('offset', String(offset))
+    return next
+  }
+
+  /**
+   * A URL taken from a response body is attacker-influenced in principle, and
+   * fetchUrl attaches this tenant's OAuth bearer token to whatever it is given.
+   * A next_page_url pointing off-host would therefore hand that token to a third
+   * party, so the origin is checked before it is ever followed.
+   */
+  private assertOwnerRezUrl(candidate: string, path: string): URL {
+    let parsed: URL
+    try {
+      parsed = new URL(candidate, BASE_URL)
+    } catch {
+      throw new Error(`[OwnerRez:${this.userId}] ${path}: unparseable next_page_url`)
+    }
+    if (parsed.origin !== BASE_URL) {
+      throw new Error(
+        `[OwnerRez:${this.userId}] ${path}: next_page_url points off-host (${parsed.origin}) — refusing to follow`
+      )
+    }
+    return parsed
+  }
+
   // ── Public methods ─────────────────────────────────────────────────────────
 
   async getProperties(): Promise<OwnerRezProperty[]> {
@@ -295,13 +401,32 @@ export class OwnerRezApiClient {
     return this.fetchAllPages<OwnerRezListing>('/v2/listings', queryParams)
   }
 
+  /**
+   * `from`/`to` are STAY-date bounds and `sinceUtc` is a MODIFICATION-time
+   * cursor — they answer different questions and compose fine together.
+   *
+   *   from — bookings that DEPART on or after this date (property timezone)
+   *   to   — bookings that ARRIVE on or before this date
+   *
+   * Together they select every stay OVERLAPPING the window, not only those
+   * contained in it, so adjacent windows both return a stay that straddles
+   * their shared edge. That is what makes the historical backfill safe to walk
+   * in windows (see lib/integrations/providers/ownerrez-backfill.ts).
+   *
+   * Verified 2026-08-13 against OwnerRez's OpenAPI contract; the parameter
+   * table is recorded in docs/Integrations/ownerrez/api-markdown.md.
+   */
   async getBookings(params: {
     propertyIds?:  number[]
     sinceUtc?:     string
+    from?:         string
+    to?:           string
     includeGuest?: boolean
   }): Promise<OwnerRezBooking[]> {
     const queryParams: Record<string, string | number | undefined> = {}
     if (params.sinceUtc) queryParams['since_utc'] = params.sinceUtc
+    if (params.from)     queryParams['from']      = params.from
+    if (params.to)       queryParams['to']        = params.to
     if (params.propertyIds?.length) {
       queryParams['property_ids'] = params.propertyIds.join(',')
     }
diff --git a/lib/integrations/providers/ownerrez-backfill.ts b/lib/integrations/providers/ownerrez-backfill.ts
new file mode 100644
index 00000000..570735dc
--- /dev/null
+++ b/lib/integrations/providers/ownerrez-backfill.ts
@@ -0,0 +1,154 @@
+// lib/integrations/providers/ownerrez-backfill.ts
+// ============================================================================
+// Window planning for OwnerRez's progressive historical booking backfill.
+//
+// WHY THIS EXISTS
+//
+// initial-sync's fetch-bookings step calls getBookings() with property ids and
+// nothing else — no date bounds at all — so it asks OwnerRez for a portfolio's
+// ENTIRE booking history in one step. That was invisible while the pager was
+// broken (it stopped after 20 records regardless), and fixing pagination turned
+// it into a real cost: a 50-property manager with several years of history is
+// thousands of bookings, fetched 100 at a time, against a 300-request/5-minute
+// budget shared by every tenant on the deployment IP — all before the PM sees
+// their first screen.
+//
+// So the initial sync now takes a bounded recent window, and older history is
+// walked backwards one window per incremental sync until it reaches a horizon.
+//
+// WHY from/to AND NOT since_utc
+//
+// since_utc is a MODIFICATION-time cursor — "created or changed since" — which
+// is right for incremental sync and useless for reaching back through history:
+// an old booking that never changed has no recent modification time. `from`/`to`
+// are stay-date bounds, verified 2026-08-13 against OwnerRez's OpenAPI contract
+// and recorded in docs/Integrations/ownerrez/api-markdown.md:
+//
+//     from — bookings that DEPART on or after this date
+//     to   — bookings that ARRIVE on or before this date
+//
+// Together they are an interval OVERLAP filter, not containment. A stay
+// straddling a window edge is returned by both adjacent windows, which is why
+// the windows below share their boundary date rather than trying to abut
+// exactly: bookings upsert on the OwnerRez id, so a duplicate is free and a gap
+// is not.
+//
+// This module is pure — dates in, dates out, no clock of its own — so the
+// walk is testable without driving an Inngest function.
+// ============================================================================
+
+/** How much history the INITIAL sync pulls inline, before any backfill. */
+export const INITIAL_HISTORY_DAYS = 90
+
+/** How much older history each incremental sync claims, one window per run. */
+export const BACKFILL_WINDOW_DAYS = 90
+
+/**
+ * How far back the walk goes in total. Two years covers prior-year comparison
+ * in owner reporting, which is the furthest back any FieldStay screen looks.
+ * At one 90-day window per hourly incremental sync a connection reaches it in
+ * about 8 hours.
+ */
+export const BACKFILL_HORIZON_DAYS = 730
+
+export interface OwnerRezBackfillState {
+  /**
+   * The oldest stay date already covered — the `from` of the last window
+   * fetched. Null means no backfill has run yet, in which case the initial
+   * sync's own window is the starting edge.
+   */
+  oldestCovered: string | null
+  /** True once the horizon is reached; the walk never restarts on its own. */
+  complete: boolean
+}
+
+export interface BackfillWindow {
+  /** Inclusive stay-date lower bound (YYYY-MM-DD). */
+  from: string
+  /** Inclusive stay-date upper bound (YYYY-MM-DD). */
+  to:   string
+}
+
+/** Calendar date in UTC, as OwnerRez's date-only bounds expect. */
+export function isoDate(d: Date): string {
+  return d.toISOString().slice(0, 10)
+}
+
+function shiftDays(from: Date, days: number): Date {
+  return new Date(from.getTime() - days * 86_400_000)
+}
+
+/**
+ * The lower stay-date bound for the INITIAL sync's booking fetch.
+ *
+ * Deliberately no upper bound at the call site: `from` alone means "departs on
+ * or after", which keeps every future booking. Bounding the future would be a
+ * bug — upcoming stays are the entire point of the first sync.
+ */
+export function initialHistoryFrom(now: Date): string {
+  return isoDate(shiftDays(now, INITIAL_HISTORY_DAYS))
+}
+
+/**
+ * The next window to fetch, or null when the walk is finished.
+ *
+ * The returned window's `to` is the previous window's `from` — they share that
+ * boundary date on purpose, see the overlap note in the header.
+ */
+export function planBackfillWindow(
+  state: OwnerRezBackfillState,
+  now:   Date,
+): BackfillWindow | null {
+  if (state.complete) return null
+
+  const horizon = isoDate(shiftDays(now, BACKFILL_HORIZON_DAYS))
+
+  // No backfill yet ⇒ start from wherever the initial sync's window ended.
+  const to = state.oldestCovered ?? initialHistoryFrom(now)
+
+  // Already at or past the horizon: nothing left to claim. (`<=` on ISO
+  // date strings is a valid chronological comparison — fixed width, zero
+  // padded, most-significant first.)
+  if (to <= horizon) return null
+
+  const candidate = isoDate(shiftDays(new Date(`${to}T00:00:00.000Z`), BACKFILL_WINDOW_DAYS))
+  const from = candidate < horizon ? horizon : candidate
+
+  return { from, to }
+}
+
+/**
+ * State after a window has been fetched successfully.
+ *
+ * `complete` is set when the window reached the horizon, so the walk stops
+ * rather than re-fetching the same terminal window on every run forever.
+ */
+export function advanceBackfill(
+  window: BackfillWindow,
+  now:    Date,
+): OwnerRezBackfillState {
+  const horizon = isoDate(shiftDays(now, BACKFILL_HORIZON_DAYS))
+  return {
+    oldestCovered: window.from,
+    complete:      window.from <= horizon,
+  }
+}
+
+/**
+ * Reads the state back out of integration_connections.metadata, which is
+ * untyped jsonb — anything in there could be any shape, including from an
+ * older build that never wrote these keys.
+ */
+export function readBackfillState(metadata: unknown): OwnerRezBackfillState {
+  const meta = (metadata !== null && typeof metadata === 'object' && !Array.isArray(metadata))
+    ? metadata as Record<string, unknown>
+    : {}
+
+  const oldest = meta['backfill_oldest_covered']
+  const done   = meta['backfill_complete']
+
+  return {
+    oldestCovered: typeof oldest === 'string' && oldest.length > 0 ? oldest : null,
+    complete:      done === true,
+  }
+}
diff --git a/lib/integrations/types.ts b/lib/integrations/types.ts
index 005fa9c5..e1256ab2 100644
--- a/lib/integrations/types.ts
+++ b/lib/integrations/types.ts
@@ -245,10 +245,26 @@ export interface OwnerRezUser {
   email:    string
 }
 
+// ✅ Confirmed 2026-08-13 against OwnerRez's published OpenAPI 3.0 contract
+// (https://api.ownerrez.com/openapi/v2.json → PageableEnumerableOf*, the
+// wrapper returned by all 22 list endpoints).
+//
+// The previous shape guessed `total_count` and `next_page_token`. NEITHER
+// name appears anywhere in that spec — zero occurrences of either string.
+// The continuation field is `next_page_url`, so `next_page_token` read as
+// undefined on every response and fetchAllPages' `while (nextPageToken)`
+// exited after ONE page, with no `limit` sent so it took OwnerRez's default
+// of 20. Live confirmation: the first OwnerRez sync for one production org
+// created exactly 20 bookings in a single minute — the only burst of that
+// size in the table. Silent: a 200, a well-formed body, no truncation signal.
 export interface OwnerRezPagedResponse<T> {
-  total_count:     number
-  items:           T[]
-  next_page_token?: string | null
+  items:  T[]
+  /** Records per page. Echoed back by OwnerRez; default 20, max 100. */
+  limit?:  number
+  /** Current offset from the start of the collection. */
+  offset?: number
+  /** Absolute URL of the next page. Null/absent means the collection is done. */
+  next_page_url?: string | null
 }
 
 // ✅ Confirmed live 2026-07-15 against a real GET /v2/reviews response
diff --git a/lib/turnovers/generator.ts b/lib/turnovers/generator.ts
index dc5386ae..515a286b 100644
--- a/lib/turnovers/generator.ts
+++ b/lib/turnovers/generator.ts
@@ -7,6 +7,16 @@ import { fetchAllRows } from '@/lib/inngest/paginate'
 import { reportError } from '@/lib/observability/report-error'
 import { tryUnwrap, unwrap, unwrapList } from '@/lib/supabase/unwrap'
 
+/**
+ * How far back a checkout can be and still produce a turnover.
+ *
+ * Generous on purpose — a turnover from three weeks ago may legitimately still
+ * be open and awaiting completion — but finite, so importing historical
+ * bookings never manufactures work that is already over. See the note at the
+ * booking read in generateTurnoversForProperty.
+ */
+export const TURNOVER_HISTORY_FLOOR_DAYS = 45
+
 export interface GeneratedTurnover {
   id:                string
   property_id:       string
@@ -257,6 +267,26 @@ export async function generateTurnoversForProperty(
   // not unique (two bookings can start the same day across a property's
   // history), and range() over a non-unique sort key can skip or repeat rows
   // across page boundaries.
+  // FLOORED to recent checkouts. A turnover is a unit of WORK — a cleaning job
+  // someone is meant to do — so manufacturing one for a stay that ended months
+  // ago creates a task nobody will ever perform, and it lands in
+  // pending_assignment where it looks like a real backlog. Production already
+  // carried 17 such rows before this floor existed, generated from ordinary
+  // historical bookings arriving through a sync.
+  //
+  // This became load-bearing with OwnerRez's historical backfill: that walk
+  // deliberately imports up to two years of past stays, and every one of them
+  // would otherwise become a retroactive cleaning job the next time anything
+  // regenerated this property. Skipping generation during the backfill alone
+  // would not have worked — this function re-reads ALL of a property's
+  // bookings, so the next ordinary sync would generate them anyway.
+  //
+  // Existing rows are untouched; only new creation is floored. The pair pass
+  // is unaffected in substance: a stay whose checkout predates the floor is
+  // exactly the one whose turnover we do not want.
+  const historyFloor = new Date(Date.now() - TURNOVER_HISTORY_FLOOR_DAYS * 86_400_000)
+    .toISOString().slice(0, 10)
+
   const bookings = await fetchAllRows<{
     id: string; checkin_date: string; checkout_date: string
     checkin_time: string | null; checkout_time: string | null
@@ -267,6 +297,7 @@ export async function generateTurnoversForProperty(
       .eq('property_id', propertyId)
       .eq('is_block', false)
       .in('status', ['confirmed', 'tentative'])
+      .gte('checkout_date', historyFloor)
       .order('checkin_date', { ascending: true })
       .order('id')
       .range(from, to),
diff --git a/unit/guardrails/n-plus-one-loops.test.ts b/unit/guardrails/n-plus-one-loops.test.ts
index d3a4ebf5..c2046442 100644
--- a/unit/guardrails/n-plus-one-loops.test.ts
+++ b/unit/guardrails/n-plus-one-loops.test.ts
@@ -133,10 +133,10 @@ const EXCEPTIONS: Record<string, string> = {
     'Per-section insert (parent-before-child, same reasoning as clone-actions.ts:122) — additionally guarded by a template-signature equality check just above that skips the whole delete-then-recreate rebuild when nothing changed.',
   'lib/inngest/functions/cron/guest-pii-retention.ts:137':
     'Per-secret delete_vault_secret RPC call — each is a distinct external Vault secret; structurally cannot be batched into one call any more than "one API call per distinct external resource" ever can. Bounded since the 2026-07-30 scalability pass: the loop now iterates one BOOKING_BATCH_SIZE page inside a per-batch step, not an org\'s entire un-anonymized booking history.',
-  'lib/inngest/functions/ownerrez/initial-sync.ts:210':
+  'lib/inngest/functions/ownerrez/initial-sync.ts:239':
     'Per-property conditional field patch (bedrooms/bathrooms/square_footage) — each property\'s patch object contains different values, so it is not a uniform batched update. Pre-fetch of existing rows just above IS already batched via .in(\'external_id\', ids).',
   'lib/guidebook/sync.ts:168':
-    'Per-property conditional guidebook-config patch — same shape as ownerrez/initial-sync.ts:210 (differing patch per row); the read side just above is already batched via .in(\'property_id\', ids).',
+    'Per-property conditional guidebook-config patch — same shape as ownerrez/initial-sync.ts:239 (differing patch per row); the read side just above is already batched via .in(\'property_id\', ids).',
   'lib/properties/upsert-normalized.ts:172':
     'Per-property conditional cleaning_cost backfill — same differing-patch-per-row shape as the two entries above.',
   'lib/inngest/functions/turnover-events.ts:354':
diff --git a/unit/inngest/ownerrez-incremental-sync.test.ts b/unit/inngest/ownerrez-incremental-sync.test.ts
index 28690f43..e1fb32c2 100644
--- a/unit/inngest/ownerrez-incremental-sync.test.ts
+++ b/unit/inngest/ownerrez-incremental-sync.test.ts
@@ -399,7 +399,7 @@ describe('ownerRezConnectionSync (per-connection handler)', () => {
     expect(patch.sync_cursor).not.toBe(patch.last_synced_at)
 
     expect(generateTurnoversForProperty).toHaveBeenCalledWith('prop_1', 'org_1', supabase)
-    expect(result).toEqual({ connectionId: 'conn_1', synced: true })
+    expect(result).toEqual({ connectionId: 'conn_1', synced: true, backfill: 'skipped' })
 
     vi.useRealTimers()
   })
@@ -415,7 +415,7 @@ describe('ownerRezConnectionSync (per-connection handler)', () => {
     const result = await invokeHandler(ownerRezConnectionSync, { event: SYNC_EVENT, step, logger: makeLogger() })
 
     expect(supabase.upsertSpy).not.toHaveBeenCalled()
-    expect(result).toEqual({ connectionId: 'conn_1', synced: false })
+    expect(result).toEqual({ connectionId: 'conn_1', synced: false, backfill: 'skipped' })
   })
 
   it('skips the bookings upsert entirely when the property lookup query fails, instead of overwriting property_id with null', async () => {
@@ -440,7 +440,7 @@ describe('ownerRezConnectionSync (per-connection handler)', () => {
     // a failed lookup must not silently mark this run as synced.
     expect(supabase.updateSpy).not.toHaveBeenCalled()
     expect(supabase.rpc).not.toHaveBeenCalled()
-    expect(result).toEqual({ connectionId: 'conn_1', synced: false })
+    expect(result).toEqual({ connectionId: 'conn_1', synced: false, backfill: 'skipped' })
   })
 
   it('marks the connection revoked, fires integration/connection.error, and surfaces a non-retriable failure', async () => {
@@ -605,3 +605,143 @@ describe('ownerRezIncrementalSync — Upstash not configured (preview)', () => {
   })
 })
 
+
+// ============================================================================
+// Progressive historical backfill.
+//
+// The initial sync deliberately fetches only a recent window (it used to ask
+// for a portfolio's entire history in one step, which was survivable only
+// because the pager stopped at 20 records). This walks the rest backwards, one
+// stay-date window per run, using OwnerRez's from/to bounds.
+//
+// The planner's arithmetic is covered in unit/integrations/ownerrez-backfill.test.ts;
+// what is checked HERE is the wiring — that a window actually reaches
+// getBookings as from/to, that progress is persisted so the next run advances,
+// and that a failure cannot either fail the live sync or skip a window silently.
+// ============================================================================
+describe('ownerRezConnectionSync — historical backfill', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-08-13T12:00:00.000Z'))
+  })
+  afterEach(() => vi.useRealTimers())
+
+  const backfillStep = () => makeAllowlistStep([
+    'check-circuit-breaker',
+    'sync-connection',
+    'backfill-history',
+  ])
+
+  /** Connection rows: one for the live sync's reload, one for the backfill's. */
+  const connRows = (metadata: Record<string, unknown>) => ([
+    { data: { ...CONN_ROW, metadata }, error: null },
+    { data: { ...CONN_ROW, metadata }, error: null },
+  ])
+
+  it('asks OwnerRez for a STAY-DATE window, not a modification-time cursor', async () => {
+    // since_utc cannot reach back through history at all — an old booking that
+    // never changed has no recent modification time. This is the whole reason
+    // the backfill exists as a separate call.
+    const mockClient = baseMocks()
+    const supabase = makeSupabase({
+      integration_connections: connRows({ sync_cursor: '2026-08-01T00:00:00.000Z' }),
+      properties:              [
+        { data: [{ external_id: '777' }], error: null },
+        { data: [{ external_id: '777' }], error: null },
+      ],
+      bookings:                [{ data: [], error: null }],
+    })
+    ;(createServiceClient as ReturnType<typeof vi.fn>).mockReturnValue(supabase)
+
+    await invokeHandler(ownerRezConnectionSync, {
+      event: SYNC_EVENT, step: backfillStep(), logger: makeLogger(),
+    })
+
+    const backfillCall = mockClient.getBookings.mock.calls
+      .map((c) => c[0] as Record<string, unknown>)
+      .find((a) => a.from !== undefined)
+
+    expect(backfillCall).toBeDefined()
+    // 90 days of history from the initial sync, then the next 90 back.
+    expect(backfillCall).toMatchObject({ to: '2026-05-15', from: '2026-02-14' })
+    expect(backfillCall?.sinceUtc).toBeUndefined()
+  })
+
+  it('persists progress so the next run claims an OLDER window', async () => {
+    // Without this write the walk re-fetches the same window forever.
+    const supabase = makeSupabase({
+      integration_connections: connRows({
+        sync_cursor: '2026-08-01T00:00:00.000Z',
+        backfill_oldest_covered: '2026-02-14',
+      }),
+      properties: [
+        { data: [{ external_id: '777' }], error: null },
+        { data: [{ external_id: '777' }], error: null },
+      ],
+      bookings:   [{ data: [], error: null }],
+    })
+    baseMocks()
+    ;(createServiceClient as ReturnType<typeof vi.fn>).mockReturnValue(supabase)
+
+    await invokeHandler(ownerRezConnectionSync, {
+      event: SYNC_EVENT, step: backfillStep(), logger: makeLogger(),
+    })
+
+    const merge = findMetadataMergeCall(supabase.rpc, 'backfill_oldest_covered')
+    expect(merge).toBeDefined()
+    const patch = (merge?.[1] as { p_patch: Record<string, unknown> }).p_patch
+    expect(patch.backfill_oldest_covered).toBe('2025-11-16')
+    expect(patch.backfill_complete).toBe(false)
+  })
+
+  it('stops for good once the walk is marked complete', async () => {
+    const mockClient = baseMocks()
+    const supabase = makeSupabase({
+      integration_connections: connRows({
+        sync_cursor: '2026-08-01T00:00:00.000Z',
+        backfill_complete: true,
+      }),
+      properties: [{ data: [{ external_id: '777' }], error: null }],
+      bookings:   [{ data: [], error: null }],
+    })
+    ;(createServiceClient as ReturnType<typeof vi.fn>).mockReturnValue(supabase)
+
+    const result = await invokeHandler(ownerRezConnectionSync, {
+      event: SYNC_EVENT, step: backfillStep(), logger: makeLogger(),
+    })
+
+    expect(result).toMatchObject({ backfill: 'complete' })
+    expect(mockClient.getBookings.mock.calls.every((c) => (c[0] as Record<string, unknown>).from === undefined))
+      .toBe(true)
+  })
+
+  it('does NOT advance progress when the window fetch fails', async () => {
+    // Advancing on failure would skip that window permanently — nothing
+    // revisits it. Retrying the same window next hour is the correct cost.
+    const mockClient = baseMocks()
+    mockClient.getBookings.mockImplementation(async (args: Record<string, unknown>) => {
+      if (args?.from) throw new Error('OwnerRez 500')
+      return []
+    })
+    const supabase = makeSupabase({
+      integration_connections: connRows({ sync_cursor: '2026-08-01T00:00:00.000Z' }),
+      properties: [
+        { data: [{ external_id: '777' }], error: null },
+        { data: [{ external_id: '777' }], error: null },
+      ],
+      bookings:   [{ data: [], error: null }],
+    })
+    ;(createServiceClient as ReturnType<typeof vi.fn>).mockReturnValue(supabase)
+
+    const result = await invokeHandler(ownerRezConnectionSync, {
+      event: SYNC_EVENT, step: backfillStep(), logger: makeLogger(),
+    })
+
+    expect(findMetadataMergeCall(supabase.rpc, 'backfill_oldest_covered')).toBeUndefined()
+    // And the LIVE sync still succeeded — backfill is catch-up work behind it,
+    // never a reason to turn an otherwise-healthy hourly run red.
+    expect(result).toMatchObject({ synced: true, backfill: 'failed' })
+    expect(reportError).toHaveBeenCalled()
+  })
+})
diff --git a/unit/integrations/ownerrez-backfill.test.ts b/unit/integrations/ownerrez-backfill.test.ts
new file mode 100644
index 00000000..c760c588
--- /dev/null
+++ b/unit/integrations/ownerrez-backfill.test.ts
@@ -0,0 +1,179 @@
+import { describe, it, expect } from 'vitest'
+
+import {
+  planBackfillWindow,
+  advanceBackfill,
+  initialHistoryFrom,
+  readBackfillState,
+  isoDate,
+  INITIAL_HISTORY_DAYS,
+  BACKFILL_WINDOW_DAYS,
+  BACKFILL_HORIZON_DAYS,
+  type OwnerRezBackfillState,
+} from '@/lib/integrations/providers/ownerrez-backfill'
+
+// ============================================================================
+// The backfill walk. Pure, so the whole thing is driven by a fixed clock.
+//
+// The property that matters most is COVERAGE: consecutive windows must leave no
+// gap in stay dates, because a gap is a permanently missing slice of history —
+// nothing ever revisits it. The walk is allowed to overlap (from/to is an
+// interval-overlap filter and bookings upsert by id), so every assertion below
+// checks for gaps, never for exact abutment.
+// ============================================================================
+
+const NOW = new Date('2026-08-13T12:00:00.000Z')
+const fresh: OwnerRezBackfillState = { oldestCovered: null, complete: false }
+
+/** Runs the walk to completion and returns every window, in order. */
+function walkToCompletion(now = NOW, maxIterations = 100) {
+  let state = fresh
+  const windows = []
+  for (let i = 0; i < maxIterations; i++) {
+    const w = planBackfillWindow(state, now)
+    if (!w) break
+    windows.push(w)
+    state = advanceBackfill(w, now)
+  }
+  return { windows, state }
+}
+
+describe('initialHistoryFrom', () => {
+  it('is INITIAL_HISTORY_DAYS before now', () => {
+    expect(initialHistoryFrom(NOW)).toBe('2026-05-15')
+    expect(initialHistoryFrom(NOW)).toBe(
+      isoDate(new Date(NOW.getTime() - INITIAL_HISTORY_DAYS * 86_400_000)),
+    )
+  })
+})
+
+describe('planBackfillWindow', () => {
+  it('starts where the initial sync stopped, so no history is skipped', () => {
+    // The seam that would otherwise silently lose 90 days: the first backfill
+    // window's upper bound must be the initial sync's lower bound.
+    expect(planBackfillWindow(fresh, NOW)?.to).toBe(initialHistoryFrom(NOW))
+  })
+
+  it('claims one BACKFILL_WINDOW_DAYS window per call', () => {
+    const w = planBackfillWindow(fresh, NOW)!
+    const span = (Date.parse(`${w.to}T00:00:00Z`) - Date.parse(`${w.from}T00:00:00Z`)) / 86_400_000
+    expect(span).toBe(BACKFILL_WINDOW_DAYS)
+  })
+
+  it('resumes from oldestCovered on the next run', () => {
+    const first  = planBackfillWindow(fresh, NOW)!
+    const second = planBackfillWindow(advanceBackfill(first, NOW), NOW)!
+    expect(second.to).toBe(first.from)
+  })
+
+  it('returns null once complete, and never restarts on its own', () => {
+    expect(planBackfillWindow({ oldestCovered: '2025-01-01', complete: true }, NOW)).toBeNull()
+  })
+
+  it('returns null when oldestCovered is already at the horizon', () => {
+    const horizon = isoDate(new Date(NOW.getTime() - BACKFILL_HORIZON_DAYS * 86_400_000))
+    expect(planBackfillWindow({ oldestCovered: horizon, complete: false }, NOW)).toBeNull()
+  })
+
+  it('clamps the final window to the horizon instead of overshooting', () => {
+    // Two days short of the horizon: the last window must be 2 days, not 90.
+    const nearly = isoDate(new Date(NOW.getTime() - (BACKFILL_HORIZON_DAYS - 2) * 86_400_000))
+    const w = planBackfillWindow({ oldestCovered: nearly, complete: false }, NOW)!
+    expect(w.from).toBe(isoDate(new Date(NOW.getTime() - BACKFILL_HORIZON_DAYS * 86_400_000)))
+    expect(w.to).toBe(nearly)
+  })
+})
+
+describe('the full walk', () => {
+  it('terminates', () => {
+    const { windows, state } = walkToCompletion()
+    expect(state.complete).toBe(true)
+    // (730 - 90) / 90 = 7.1 -> 8 windows, the last one clamped.
+    expect(windows).toHaveLength(8)
+  })
+
+  it('leaves NO GAP in stay-date coverage', () => {
+    // The defect this guards: any gap is history that nothing ever revisits.
+    const { windows } = walkToCompletion()
+    for (let i = 1; i < windows.length; i++) {
+      expect(windows[i].to).toBe(windows[i - 1].from)
+    }
+  })
+
+  it('reaches exactly the horizon and no further', () => {
+    const { windows } = walkToCompletion()
+    const horizon = isoDate(new Date(NOW.getTime() - BACKFILL_HORIZON_DAYS * 86_400_000))
+    expect(windows[windows.length - 1].from).toBe(horizon)
+  })
+
+  it('covers the whole span from the horizon up to the initial window', () => {
+    const { windows } = walkToCompletion()
+    expect(windows[0].to).toBe(initialHistoryFrom(NOW))
+    expect(windows[windows.length - 1].from)
+      .toBe(isoDate(new Date(NOW.getTime() - BACKFILL_HORIZON_DAYS * 86_400_000)))
+  })
+
+  it('every window is non-empty and ordered', () => {
+    const { windows } = walkToCompletion()
+    for (const w of windows) expect(w.from < w.to).toBe(true)
+  })
+
+  it('is idempotent under a replayed run — re-planning without advancing repeats the same window', () => {
+    // Inngest retries a step; planning must not consume a window as a side effect.
+    const a = planBackfillWindow(fresh, NOW)
+    const b = planBackfillWindow(fresh, NOW)
+    expect(a).toEqual(b)
+  })
+})
+
+describe('advanceBackfill', () => {
+  it('records the window it just covered', () => {
+    const w = planBackfillWindow(fresh, NOW)!
+    expect(advanceBackfill(w, NOW).oldestCovered).toBe(w.from)
+  })
+
+  it('does not mark complete mid-walk', () => {
+    const w = planBackfillWindow(fresh, NOW)!
+    expect(advanceBackfill(w, NOW).complete).toBe(false)
+  })
+
+  it('marks complete when the window reached the horizon', () => {
+    const horizon = isoDate(new Date(NOW.getTime() - BACKFILL_HORIZON_DAYS * 86_400_000))
+    expect(advanceBackfill({ from: horizon, to: '2025-01-01' }, NOW).complete).toBe(true)
+  })
+})
+
+describe('readBackfillState', () => {
+  it('reads a well-formed metadata blob', () => {
+    expect(readBackfillState({ backfill_oldest_covered: '2026-01-01', backfill_complete: true }))
+      .toEqual({ oldestCovered: '2026-01-01', complete: true })
+  })
+
+  it('treats an older connection with no backfill keys as a fresh walk', () => {
+    expect(readBackfillState({ sync_cursor: 'x' })).toEqual({ oldestCovered: null, complete: false })
+  })
+
+  it.each([
+    ['null',      null],
+    ['undefined', undefined],
+    ['an array',  ['nope']],
+    ['a string',  'nope'],
+    ['a number',  42],
+  ])('survives metadata that is %s', (_label, value) => {
+    expect(readBackfillState(value)).toEqual({ oldestCovered: null, complete: false })
+  })
+
+  it.each([
+    ['an empty string', ''],
+    ['a number',        20260101],
+    ['null',            null],
+  ])('ignores a non-string oldest cursor (%s) rather than trusting it', (_label, value) => {
+    expect(readBackfillState({ backfill_oldest_covered: value }).oldestCovered).toBeNull()
+  })
+
+  it('only accepts a literal true for complete', () => {
+    // A truthy-but-not-true value must not silently end the walk.
+    expect(readBackfillState({ backfill_complete: 'yes' }).complete).toBe(false)
+    expect(readBackfillState({ backfill_complete: 1 }).complete).toBe(false)
+  })
+})
diff --git a/unit/integrations/ownerrez-pagination.test.ts b/unit/integrations/ownerrez-pagination.test.ts
new file mode 100644
index 00000000..7ea69547
--- /dev/null
+++ b/unit/integrations/ownerrez-pagination.test.ts
@@ -0,0 +1,236 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+
+vi.mock('server-only', () => ({}))
+
+// The transport is the unit under test, so only its collaborators are mocked.
+vi.mock('@/lib/integrations/vault', () => ({
+  readIntegrationToken: vi.fn(async () => 'test-token'),
+}))
+vi.mock('@/lib/supabase/server', () => ({
+  createServiceClient: vi.fn(() => { throw new Error('not expected in these tests') }),
+}))
+vi.mock('@/lib/supabase/unwrap', () => ({ unwrap: vi.fn() }))
+vi.mock('@/lib/observability/report-error', () => ({ reportError: vi.fn() }))
+// upstashConfigured() === false short-circuits the shared IP budget check, so
+// these tests exercise pagination without a Redis double.
+vi.mock('@/lib/redis', () => ({
+  getRedis:           vi.fn(),
+  upstashConfigured:  vi.fn(() => false),
+}))
+
+import { OwnerRezApiClient } from '@/lib/integrations/providers/ownerrez-api'
+
+// ============================================================================
+// OwnerRez pagination — the layer that had NO test at all, which is how it
+// shipped reading two fields that do not exist in OwnerRez's API.
+//
+// `OwnerRezPagedResponse` declared `total_count` and `next_page_token`. Neither
+// string appears anywhere in https://api.ownerrez.com/openapi/v2.json (zero
+// occurrences of each); the real wrapper is { items, limit, offset,
+// next_page_url }. So `next_page_token` was undefined on every response, the
+// do/while exited after one page, and no `limit` was sent — leaving OwnerRez's
+// default of 20. It affected getBookings, getListings, getGuests and getReviews
+// alike, and produced a 200 with a well-formed body and no truncation signal.
+//
+// Live confirmation before the fix: one production org's first OwnerRez sync
+// created exactly 20 bookings inside a single minute — the only burst of that
+// size anywhere in the table.
+//
+// Every existing OwnerRez test mocks the client CLASS (getBookings and friends)
+// and so sits entirely above this code. These drive globalThis.fetch instead.
+// ============================================================================
+
+const BASE = 'https://api.ownerrez.com'
+
+interface PageSpec {
+  items:          unknown[]
+  limit?:         number
+  offset?:        number
+  next_page_url?: string | null
+}
+
+/** Records every URL requested and replays `pages` in order. */
+function mockPages(pages: PageSpec[]) {
+  const urls: string[] = []
+  const fetchMock = vi.fn(async (url: string) => {
+    urls.push(url)
+    const page = pages[urls.length - 1]
+    if (!page) throw new Error(`unexpected extra request #${urls.length} to ${url}`)
+    return {
+      ok:      true,
+      status:  200,
+      headers: new Headers(),
+      json:    async () => page,
+      text:    async () => JSON.stringify(page),
+    }
+  })
+  globalThis.fetch = fetchMock as unknown as typeof globalThis.fetch
+  return { urls, fetchMock }
+}
+
+const rows = (n: number, from = 0) =>
+  Array.from({ length: n }, (_, i) => ({ id: from + i }))
+
+let originalFetch: typeof globalThis.fetch
+
+beforeEach(() => {
+  originalFetch = globalThis.fetch
+  process.env.OWNERREZ_CLIENT_ID = 'test-client-id'
+})
+
+afterEach(() => {
+  globalThis.fetch = originalFetch
+  vi.restoreAllMocks()
+})
+
+describe('OwnerRez pagination', () => {
+  it('follows next_page_url across pages and returns EVERY record', async () => {
+    // The regression test. Against the old code this returns 100, not 250.
+    const { urls } = mockPages([
+      { items: rows(100, 0),   limit: 100, offset: 0,   next_page_url: `${BASE}/v2/bookings?limit=100&offset=100` },
+      { items: rows(100, 100), limit: 100, offset: 100, next_page_url: `${BASE}/v2/bookings?limit=100&offset=200` },
+      { items: rows(50,  200), limit: 100, offset: 200, next_page_url: null },
+    ])
+
+    const all = await new OwnerRezApiClient('user-1').getBookings({})
+
+    expect(all).toHaveLength(250)
+    expect(urls).toHaveLength(3)
+    // Contiguous and in order — no page dropped, none replayed.
+    expect((all as { id: number }[]).map((b) => b.id)).toEqual(rows(250).map((r) => r.id))
+  })
+
+  it('asks for the 100-record maximum instead of taking the default 20', async () => {
+    const { urls } = mockPages([{ items: rows(3), limit: 100, offset: 0, next_page_url: null }])
+
+    await new OwnerRezApiClient('user-1').getBookings({})
+
+    expect(new URL(urls[0]).searchParams.get('limit')).toBe('100')
+  })
+
+  it('preserves caller params on the first page', async () => {
+    const { urls } = mockPages([{ items: [], next_page_url: null }])
+
+    await new OwnerRezApiClient('user-1').getBookings({
+      propertyIds:  [7, 9],
+      sinceUtc:     '2026-01-01T00:00:00Z',
+      includeGuest: true,
+    })
+
+    const q = new URL(urls[0]).searchParams
+    expect(q.get('property_ids')).toBe('7,9')
+    expect(q.get('since_utc')).toBe('2026-01-01T00:00:00Z')
+    expect(q.get('include_guest')).toBe('true')
+  })
+
+  it('stops at a null next_page_url even when the page is full', async () => {
+    // next_page_url is authoritative when present: a full final page is still
+    // final if the server says so.
+    const { urls } = mockPages([{ items: rows(100), limit: 100, offset: 0, next_page_url: null }])
+
+    const all = await new OwnerRezApiClient('user-1').getReviews()
+
+    expect(all).toHaveLength(100)
+    expect(urls).toHaveLength(1)
+  })
+
+  it('stops on an empty page without looping', async () => {
+    const { urls } = mockPages([{ items: [], limit: 100, offset: 0 }])
+
+    await expect(new OwnerRezApiClient('user-1').getGuests()).resolves.toEqual([])
+    expect(urls).toHaveLength(1)
+  })
+
+  describe('offset fallback when next_page_url is absent', () => {
+    it('keeps paging while pages come back full', async () => {
+      const { urls } = mockPages([
+        { items: rows(100, 0),   limit: 100, offset: 0 },
+        { items: rows(40,  100), limit: 100, offset: 100 },
+      ])
+
+      const all = await new OwnerRezApiClient('user-1').getListings()
+
+      expect(all).toHaveLength(140)
+      expect(new URL(urls[1]).searchParams.get('offset')).toBe('100')
+    })
+
+    it('trusts the server-reported limit over the one we requested', async () => {
+      // If OwnerRez quietly caps us at 20 while we ask for 100, a short-page
+      // test against OUR requested size stops after one page — the original
+      // bug wearing a different hat. page.limit is what settles it.
+      const { urls } = mockPages([
+        { items: rows(20, 0),  limit: 20, offset: 0 },
+        { items: rows(20, 20), limit: 20, offset: 20 },
+        { items: rows(5,  40), limit: 20, offset: 40 },
+      ])
+
+      const all = await new OwnerRezApiClient('user-1').getBookings({})
+
+      expect(all).toHaveLength(45)
+      expect(urls).toHaveLength(3)
+    })
+
+    it('stops on a short page', async () => {
+      const { urls } = mockPages([{ items: rows(99), limit: 100, offset: 0 }])
+
+      await expect(new OwnerRezApiClient('user-1').getBookings({})).resolves.toHaveLength(99)
+      expect(urls).toHaveLength(1)
+    })
+  })
+
+  describe('next_page_url is response-supplied, so its origin is checked', () => {
+    it('refuses to follow an off-host URL and never sends the token there', async () => {
+      const { urls } = mockPages([
+        { items: rows(1), limit: 100, offset: 0, next_page_url: 'https://evil.example.com/v2/bookings?limit=100' },
+      ])
+
+      await expect(new OwnerRezApiClient('user-1').getBookings({}))
+        .rejects.toThrow(/off-host/)
+
+      // The point of the guard: the bearer token was never sent to that host.
+      expect(urls).toHaveLength(1)
+      expect(urls.every((u) => u.startsWith(BASE))).toBe(true)
+    })
+
+    it('accepts a relative next_page_url, resolved against the API base', async () => {
+      const { urls } = mockPages([
+        { items: rows(1), limit: 100, offset: 0, next_page_url: '/v2/bookings?limit=100&offset=100' },
+        { items: [],      limit: 100, offset: 100, next_page_url: null },
+      ])
+
+      await new OwnerRezApiClient('user-1').getBookings({})
+
+      expect(urls[1]).toBe(`${BASE}/v2/bookings?limit=100&offset=100`)
+    })
+
+    it('rejects an unparseable next_page_url rather than fetching it', async () => {
+      mockPages([
+        { items: rows(1), limit: 100, offset: 0, next_page_url: 'http://[oops' },
+      ])
+
+      await expect(new OwnerRezApiClient('user-1').getBookings({}))
+        .rejects.toThrow(/next_page_url/)
+    })
+  })
+
+  it('THROWS past the page ceiling instead of returning a partial list', async () => {
+    // A server that never terminates. Returning results.length here would hand
+    // callers a short list they would treat as complete — and OwnerRez callers
+    // reconcile deletions against exactly that list.
+    let n = 0
+    globalThis.fetch = vi.fn(async () => {
+      n++
+      return {
+        ok: true, status: 200, headers: new Headers(),
+        json: async () => ({
+          items: rows(100), limit: 100, offset: n * 100,
+          next_page_url: `${BASE}/v2/bookings?limit=100&offset=${n * 100}`,
+        }),
+        text: async () => '',
+      }
+    }) as unknown as typeof globalThis.fetch
+
+    await expect(new OwnerRezApiClient('user-1').getBookings({}))
+      .rejects.toThrow(/refusing to return a partial result/)
+  })
+})