diff --git a/lib/core/fetchCommit.ts b/lib/core/fetchCommit.ts
new file mode 100644
index 0000000..4ee299d
--- /dev/null
+++ b/lib/core/fetchCommit.ts
@@ -0,0 +1,173 @@
+import type { PageResponse, QueryResponse } from '../adapters/adapter.js'
+import { rebaseResponseData, type FetchRebasePlan } from './fetchRebase.js'
+import { queryPolicy } from './queryMaintenance.js'
+import { commitQuery } from './queryResults.js'
+import {
+  entityKey,
+  type EntityKey,
+  type ItemId,
+  type ProcessedCacheEvent,
+  type ProcessedProjectionEvent,
+  type Query,
+  type QueuedEvent,
+  type ServiceState,
+  type TraceCause,
+} from './queryTypes.js'
+import { applyEventsToService, diffCompleteSet, isCompleteSetQuery } from './windowMaintenance.js'
+
+export type StoreResponse<TMeta> =
+  | QueryResponse<unknown, TMeta | undefined>
+  | PageResponse<unknown[], TMeta>
+
+interface CompleteSetCommit {
+  previousEntities: Map<EntityKey, unknown>
+  nextItemIds: Set<EntityKey>
+  ignoredItemIds: ReadonlySet<EntityKey>
+}
+
+interface EntityAccess {
+  getId: (item: unknown) => ItemId | undefined
+  isItemStale: (current: unknown, next: unknown) => boolean
+}
+
+export interface FetchCommitPlan<TMeta> {
+  query: Omit<Query<unknown, TMeta>, 'rows'> & { fetchedAt: number }
+  entityEvents: QueuedEvent[]
+  completeSet: CompleteSetCommit | null
+  replayEvents: readonly ProcessedCacheEvent[]
+  deferredMutationLaneKeys: readonly string[]
+}
+
+/** Plan against the confirmed lane bases and current overlays without mutating the cache. */
+export function buildFetchCommit<TMeta>({
+  service,
+  query,
+  result,
+  source,
+  rebasePlan,
+  pendingMutationEvents,
+  getId,
+  isItemStale,
+  meta,
+  fetchedAt,
+  cause,
+}: EntityAccess & {
+  service: ServiceState<TMeta>
+  query: Query<unknown, TMeta>
+  result: StoreResponse<TMeta>
+  source: 'cache' | 'server'
+  rebasePlan: FetchRebasePlan
+  pendingMutationEvents: readonly ProcessedProjectionEvent[]
+  meta: TMeta
+  fetchedAt: number
+  cause?: TraceCause
+}): FetchCommitPlan<TMeta> {
+  const policy = queryPolicy(query.maintenance)
+  const mode = policy.responseMode
+  const queryMutationEvents = mode === 'snapshot' ? [] : pendingMutationEvents
+  const latestEventById = new Map(rebasePlan.latestEventById)
+  for (const event of queryMutationEvents) latestEventById.set(event.itemId, event)
+  const ignoredItemIds = new Set(rebasePlan.itemIds)
+  // Every pending mutation protects its canonical entity from a fetch overwrite,
+  // including snapshots, whose returned rows deliberately ignore those mutations.
+  for (const event of pendingMutationEvents) ignoredItemIds.add(event.itemId)
+
+  const rebased = rebaseResponseData({
+    data: result.data,
+    mode,
+    latestEventById,
+    entities: service.entities,
+    getId,
+    isItemStale,
+    canKeepCurrentItem: item =>
+      !(
+        query.desc.method === 'find' &&
+        policy.rebaseMembership &&
+        query.maintenance.matches(item) === false
+      ),
+  })
+  const entityEvents: QueuedEvent[] = []
+  // Projections never replace complete canonical entities, even at the same revision.
+  if (source === 'server' && !query.maintenance.isProjection) {
+    for (const item of rebased.items) {
+      const id = getId(item)
+      if (id === undefined || ignoredItemIds.has(entityKey(id))) continue
+      entityEvents.push({
+        mode: 'server',
+        source: 'fetch',
+        serviceName: query.desc.serviceName,
+        type: service.entities.has(entityKey(id)) ? 'updated' : 'created',
+        item,
+        ...(cause === undefined ? {} : { cause }),
+      })
+    }
+  }
+
+  let completeSet: CompleteSetCommit | null = null
+  if (isCompleteSetQuery(query)) {
+    const previousEntities = new Map<EntityKey, unknown>()
+    for (const itemId of query.rows.ids) {
+      const entity = service.entities.get(itemId)
+      if (entity !== undefined) previousEntities.set(itemId, entity)
+    }
+    const nextItemIds = new Set(rebased.itemIds)
+    // A stale complete response cannot delete a newly created row or resurrect a removal.
+    for (const [itemId, event] of latestEventById) {
+      if (event.type === 'removed') nextItemIds.delete(itemId)
+      else nextItemIds.add(itemId)
+    }
+    completeSet = { previousEntities, nextItemIds, ignoredItemIds }
+  }
+
+  const events = [...rebasePlan.events, ...queryMutationEvents]
+  const replayEvents =
+    mode === 'entity'
+      ? events
+      : mode === 'fetch-owned'
+        ? events.filter(event => event.mode !== 'server' || event.source !== 'realtime')
+        : []
+  const pageInfo = 'pageInfo' in result ? result.pageInfo : undefined
+  return {
+    query: {
+      ...query,
+      fetchedAt,
+      state: {
+        status: 'success',
+        data: rebased.data,
+        meta,
+        ...(pageInfo ? { pageInfo } : {}),
+        isFetching: false,
+        error: null,
+      },
+    },
+    entityEvents,
+    completeSet,
+    replayEvents,
+    deferredMutationLaneKeys: queryMutationEvents.map(event => event.mutationLaneKey),
+  }
+}
+
+/** Apply entities and query membership before the store publishes any effects. */
+export function applyFetchCommit<TMeta>(
+  service: ServiceState<TMeta>,
+  plan: FetchCommitPlan<TMeta>,
+  access: EntityAccess,
+): ProcessedCacheEvent[] {
+  const { query, completeSet } = plan
+  const processedEvents: ProcessedCacheEvent[] = []
+  applyEventsToService({
+    service,
+    serviceName: query.desc.serviceName,
+    events: plan.entityEvents,
+    ...access,
+    processedEvents,
+  })
+  if (completeSet) {
+    service.materialized = { queryId: query.queryId, fetchedAt: query.fetchedAt }
+  }
+  commitQuery(service, query)
+  // Complete-set diffs establish membership; ordinary fetches only establish row values.
+  return completeSet
+    ? diffCompleteSet({ service, serviceName: query.desc.serviceName, ...completeSet })
+    : processedEvents
+}
diff --git a/lib/core/fetchRebase.ts b/lib/core/fetchRebase.ts
index 337b60a..bca07b7 100644
--- a/lib/core/fetchRebase.ts
+++ b/lib/core/fetchRebase.ts
@@ -1,3 +1,4 @@
+import type { FetchResponseMode } from './queryMaintenance.js'
 import { CappedBuffer } from './cappedBuffer.js'
 import { entityKey, type EntityKey, type ItemId, type ProcessedCacheEvent } from './queryTypes.js'
 
@@ -155,8 +156,6 @@ export interface RebasedResponse {
   readonly itemIds: ReadonlySet<EntityKey>
 }
 
-export type FetchResponseMode = 'entity' | 'projection' | 'snapshot' | 'fetch-owned'
-
 function overlayProjectionItem(
   responseItem: unknown,
   event: ProcessedCacheEvent,
diff --git a/lib/core/figbird.ts b/lib/core/figbird.ts
index edff427..ff0c07d 100644
--- a/lib/core/figbird.ts
+++ b/lib/core/figbird.ts
@@ -34,12 +34,12 @@ import {
 import { resolveQueryInput, type PreparedQuery, type QueryInput } from './queryDefinition.js'
 import {
   explainQueryNode,
-  isServerMaintained,
   type ClassificationReason,
   type ExplainReport,
   type QueryNodeClass,
 } from './queryClassification.js'
 export type { ExplainNode, ExplainReport } from './queryClassification.js'
+import { queryPolicy } from './queryMaintenance.js'
 import { QueryRef } from './queryRef.js'
 import {
   DEFAULT_GC_TIME,
@@ -79,7 +79,7 @@ import type {
 } from './schema.js'
 import { foreignKeys, resolveServicePath } from './schema.js'
 import type { ValueComparator } from './sort.js'
-import { isWithinStaleTime, validatePrefetchStaleTime, validateStaleTime } from './staleTime.js'
+import { validatePrefetchStaleTime, validateStaleTime } from './staleTime.js'
 import { createTransactionContext, type TransactionContext } from './transactions.js'
 
 type DescriptorWriteProjection<TItem> =
@@ -328,11 +328,6 @@ export class Figbird<
   dispose(): void {
     if (this.#disposed) return
     this.#disposed = true
-    for (const prefetch of this.#prefetches.values()) {
-      prefetch.timer.cancel()
-      prefetch.release()
-    }
-    this.#prefetches.clear()
     for (const ref of this.#windowQueryCache.values()) ref.dispose()
     for (const ref of this.#relationalQueryCache.values()) ref.dispose()
     this.#windowQueryCache.clear()
@@ -608,9 +603,6 @@ export class Figbird<
     }
   }
 
-  // Active speculative pins, keyed by query hash (see prefetch()).
-  #prefetches: Map<string, { at: number; release: () => void; timer: ClockTimer }> = new Map()
-
   /**
    * Speculatively warm a query — the idempotent, fire-and-forget sibling of `prepare()`.
    *
@@ -635,28 +627,7 @@ export class Figbird<
     options?: { staleTime?: number },
   ): void {
     const staleTime = validatePrefetchStaleTime(options?.staleTime ?? 30_000)
-    const ref = this.query(query)
-    const hash = ref.hash()
-
-    const now = this.clock.now()
-    const existing = this.#prefetches.get(hash)
-    if (existing && isWithinStaleTime(existing.at, staleTime, now)) return
-    if (existing) {
-      existing.timer.cancel()
-      existing.release()
-      this.#prefetches.delete(hash)
-    }
-
-    // The pin also carries the staleTime so a warm-in-store read within the window
-    // skips the SWR revalidation instead of re-fetching.
-    const release = ref.subscribe(() => {}, { staleTime, source: 'prefetch' })
-    const timer = this.clock.setTimeout(() => {
-      this.#prefetches.delete(hash)
-      release()
-    }, staleTime)
-    // Never keep a Node process alive for a speculative pin (browsers ignore this).
-    timer.unref()
-    this.#prefetches.set(hash, { at: now, release, timer })
+    this.query(query).prefetch(staleTime)
   }
 
   // Descriptor layer — the primitive the relational engine (and the deprecated
@@ -1053,7 +1024,7 @@ export class Figbird<
                 server: query.config.server,
                 allPages: 'allPages' in query.config && query.config.allPages === true,
                 localOperators: locallySupportedOperators(this.adapter, serviceName),
-                snapshot: query.config.realtime === 'disabled',
+                snapshot: queryPolicy(query.maintenance).responseMode === 'snapshot',
               })
             : null
         rows.push({
@@ -1073,13 +1044,7 @@ export class Figbird<
             : {}),
           classification: query.maintenance.classification,
           classificationReasons: explanation?.reasons ?? [],
-          realtimeStrategy:
-            query.config.realtime === 'disabled'
-              ? 'manual'
-              : query.config.realtime === 'refetch' ||
-                  isServerMaintained(query.maintenance.classification)
-                ? 'refetch'
-                : 'merge',
+          realtimeStrategy: queryPolicy(query.maintenance).realtimeStrategy,
           skipped: query.config.skip === true,
           status: query.state.status,
           isFetching: query.state.isFetching,
diff --git a/lib/core/queryMaintenance.ts b/lib/core/queryMaintenance.ts
index 3892f01..e2f4d5c 100644
--- a/lib/core/queryMaintenance.ts
+++ b/lib/core/queryMaintenance.ts
@@ -1,6 +1,8 @@
 import {
   classifyStoredQuery,
   isProjectionQuery,
+  isServerMaintained,
+  usesFetchOwnedRows,
   type StoredQueryClass,
 } from './queryClassification.js'
 import { buildComparator, type ValueComparator } from './sort.js'
@@ -11,6 +13,8 @@ import {
   type QueryConfig,
 } from './queryTypes.js'
 
+export type FetchResponseMode = 'entity' | 'projection' | 'snapshot' | 'fetch-owned'
+
 export interface QueryMaintenance {
   classification: StoredQueryClass
   matches: (item: unknown) => MatchResult
@@ -21,6 +25,29 @@ export interface QueryMaintenance {
   isProjection: boolean
 }
 
+/** Immutable execution rules kept out of subscriber-visible query state. */
+interface QueryPolicy {
+  responseMode: FetchResponseMode
+  /** Preferred row ownership; missing IDs still force a query to retain its own values. */
+  rowSource: 'entities' | 'values'
+  /** Whether event replay may change membership through the local matcher. */
+  mergeEvents: boolean
+  /** Keep visible row values current when a membership merge cannot be proved. */
+  replayVisibleEvents: boolean
+  /** Whether a newer entity can be rejected by the fetched query's local predicate. */
+  rebaseMembership: boolean
+  refetchEvents: boolean
+  realtimeStrategy: 'manual' | 'refetch' | 'merge'
+}
+
+const policies = new WeakMap<QueryMaintenance, Readonly<QueryPolicy>>()
+
+export function queryPolicy(maintenance: QueryMaintenance): Readonly<QueryPolicy> {
+  const policy = policies.get(maintenance)
+  if (!policy) throw new Error('figbird: query maintenance has not been compiled')
+  return policy
+}
+
 /** Split window operators off a query so the rest can feed the local matcher. */
 function splitWindow(q: Record<string, unknown> | undefined): {
   filters: Record<string, unknown> | undefined
@@ -60,12 +87,21 @@ export function compileQueryMaintenance({
     allPages: 'allPages' in config && config.allPages === true,
     localOperators,
   })
+  const isProjection = isProjectionQuery(query)
+  const responseMode: FetchResponseMode =
+    config.realtime === 'disabled'
+      ? 'snapshot'
+      : isProjection
+        ? 'projection'
+        : usesFetchOwnedRows(classification, config.realtime)
+          ? 'fetch-owned'
+          : 'entity'
   const { filters, sort, limit, skip } = splitWindow(query)
   const effectiveSort = sort ?? defaultSort
   // Local reads also serve queries with realtime disabled. Find matchers receive
   // only predicates here, while realtime matchers retain the original query input.
   let localMatcher: ((item: unknown) => MatchResult) | undefined
-  return {
+  const maintenance: QueryMaintenance = {
     classification,
     matches:
       config.realtime === 'merge' && classification !== 'server-authoritative'
@@ -76,6 +112,27 @@ export function compileQueryMaintenance({
     compare: effectiveSort ? buildComparator(effectiveSort, compare) : undefined,
     limit,
     skip,
-    isProjection: isProjectionQuery(query),
+    isProjection,
   }
+  policies.set(maintenance, {
+    responseMode,
+    rowSource:
+      responseMode !== 'entity' || config.fetchPolicy === 'network-only' ? 'values' : 'entities',
+    mergeEvents:
+      config.realtime === 'merge' &&
+      !(desc.method === 'find' && config.fetchPolicy === 'network-only'),
+    replayVisibleEvents:
+      config.realtime === 'refetch' ||
+      config.fetchPolicy === 'network-only' ||
+      isServerMaintained(classification),
+    rebaseMembership: config.realtime === 'merge' && !isServerMaintained(classification),
+    refetchEvents: config.realtime === 'refetch',
+    realtimeStrategy:
+      config.realtime === 'disabled'
+        ? 'manual'
+        : config.realtime === 'refetch' || isServerMaintained(classification)
+          ? 'refetch'
+          : 'merge',
+  })
+  return maintenance
 }
diff --git a/lib/core/queryResults.ts b/lib/core/queryResults.ts
index 68a89ee..a95a710 100644
--- a/lib/core/queryResults.ts
+++ b/lib/core/queryResults.ts
@@ -1,5 +1,5 @@
+import { queryPolicy } from './queryMaintenance.js'
 import { entityKey, type EntityKey, type Query, type ServiceState } from './queryTypes.js'
-import { usesFetchOwnedRows } from './queryClassification.js'
 
 /** Query membership is indexed by ID; fetch-owned results retain their returned values. */
 export interface QueryRows {
@@ -37,10 +37,7 @@ export function commitQuery<TMeta>(
   }
   const ownsValues =
     next.state.status !== 'success' ||
-    next.config.realtime === 'disabled' ||
-    usesFetchOwnedRows(next.maintenance.classification, next.config.realtime) ||
-    next.config.fetchPolicy === 'network-only' ||
-    next.maintenance.isProjection ||
+    queryPolicy(next.maintenance).rowSource === 'values' ||
     ids.length !== items.length
   const rows: QueryRows = { kind: ownsValues ? 'values' : 'entities', ids, data: items }
 
diff --git a/lib/core/queryStore.ts b/lib/core/queryStore.ts
index 76e0230..7f9acb4 100644
--- a/lib/core/queryStore.ts
+++ b/lib/core/queryStore.ts
@@ -1,43 +1,32 @@
 import { systemClock, type Clock, type ClockTimer } from './clock.js'
-import { compileQueryMaintenance } from './queryMaintenance.js'
+import { compileQueryMaintenance, queryPolicy } from './queryMaintenance.js'
+import { applyFetchCommit, buildFetchCommit, type StoreResponse } from './fetchCommit.js'
 import { queryPage, EMPTY_PAGE, type QueryPage } from '../adapters/queryPage.js'
 import { ReconcileScheduler, type ReconcilePreparation } from './reconcileScheduler.js'
 import { commitQuery, deleteQuery } from './queryResults.js'
 import { QueryRetention } from './queryRetention.js'
-import {
-  locallySupportedOperators,
-  type Adapter,
-  type PageResponse,
-  type QueryResponse,
-} from '../adapters/adapter.js'
+import { locallySupportedOperators, type Adapter, type QueryResponse } from '../adapters/adapter.js'
 import { isEphemeralQuery } from './queryIdentity.js'
 import { type FigbirdEventEmitter, type FetchReason, type TraceCause } from './events.js'
 import { QueryTelemetry } from './queryTelemetry.js'
 import type { MutationActivity } from './mutationTracker.js'
 import { MutationExecutor } from './mutationExecutor.js'
 import { type RegisteredMutation, type ScheduledMutationControl } from './mutationQueue.js'
-import {
-  FetchEventJournal,
-  planFetchRebase,
-  rebaseResponseData,
-  type FetchResponseMode,
-} from './fetchRebase.js'
+import { FetchEventJournal, planFetchRebase } from './fetchRebase.js'
 import { ABSENT, type ProjectionChange } from './mutationLanes.js'
 import type { ForeignKey } from './schema.js'
 import {
   applyEventsToService,
   applyVisibleEventToQuery,
   createServiceState,
-  diffCompleteSet,
   findInsertIndex,
-  groupEventsByService,
   isCompleteSetQuery,
+  groupEventsByService,
   replayFetchedQueryFromEvents,
   reapplyQueryFromEntities,
   updateQueriesFromEvents,
   type QueryMembershipScope,
 } from './windowMaintenance.js'
-import { isServerMaintained, usesFetchOwnedRows } from './queryClassification.js'
 import {
   entityKey,
   queryOfParams,
@@ -99,10 +88,6 @@ function isIdOnlyRealtimePayload(item: unknown, itemId: ItemId | undefined): boo
   )
 }
 
-type StoreResponse<TMeta> =
-  | QueryResponse<unknown, TMeta | undefined>
-  | PageResponse<unknown[], TMeta>
-
 type QueuedStoreEvent = QueuedEvent | RealtimeInvalidation
 
 interface AppliedEventEffect {
@@ -1247,7 +1232,7 @@ export class QueryStore<
     if (!service) return null
     if (
       !service.materialized &&
-      (query.config.realtime !== 'merge' ||
+      (!queryPolicy(query.maintenance).mergeEvents ||
         !this.#mutationExecutor.hasOptimisticCreate(desc.serviceName, desc.resourceId))
     )
       return null
@@ -1325,7 +1310,8 @@ export class QueryStore<
     service: ServiceState<TMeta>,
     query: Query<unknown, TMeta>,
   ): 'unavailable' | 'unchanged' | 'changed' {
-    if (query.config.realtime !== 'merge' || query.state.status !== 'success') return 'unavailable'
+    if (!queryPolicy(query.maintenance).mergeEvents || query.state.status !== 'success')
+      return 'unavailable'
     const local = this.#selectMaterializedFind(query)
     if (!local) return 'unavailable'
     const previous = query.state.data
@@ -1401,15 +1387,7 @@ export class QueryStore<
       const getId = this.#getIdReader(query.desc.serviceName)
       const data = result.data
       const responseItems = Array.isArray(data) ? data : data == null ? [] : [data]
-      const isProjection = query.maintenance.isProjection
-      const responseMode: FetchResponseMode =
-        query.config.realtime === 'disabled'
-          ? 'snapshot'
-          : isProjection
-            ? 'projection'
-            : usesFetchOwnedRows(query.maintenance.classification, query.config.realtime)
-              ? 'fetch-owned'
-              : 'entity'
+      const responseMode = queryPolicy(query.maintenance).responseMode
       const rebasePlan = planFetchRebase({
         responseItems,
         journalEvents,
@@ -1446,123 +1424,34 @@ export class QueryStore<
         )
       }
 
-      const activeOverlayEvents = this.#mutationExecutor.overlayEvents(query.desc.serviceName)
-      const overlayEvents = query.config.realtime === 'disabled' ? [] : activeOverlayEvents
-      const effectiveJournalEvents = [...rebasePlan.events, ...overlayEvents]
-      hadEffectiveJournalEvents = rebasePlan.events.length > 0
-      const latestEventById = new Map(rebasePlan.latestEventById)
-      const journaledItemIds = new Set(rebasePlan.itemIds)
-      for (const event of overlayEvents) {
-        latestEventById.set(event.itemId, event)
-        this.#mutationExecutor.deferQueryIds(event.mutationLaneKey, [queryId])
-      }
-      for (const event of activeOverlayEvents) journaledItemIds.add(event.itemId)
-
-      const isCompleteSet = isCompleteSetQuery(query)
-      const previousRootEntities: Map<EntityKey, unknown> | null = isCompleteSet ? new Map() : null
-      if (previousRootEntities) {
-        for (const itemId of query.rows.ids) {
-          const entity = service.entities.get(itemId)
-          if (entity !== undefined) previousRootEntities.set(itemId, entity)
-        }
-      }
-      const meta = (result as { meta?: TMeta }).meta
-      const pageInfo = 'pageInfo' in result ? result.pageInfo : undefined
-      const rebasedResponse = rebaseResponseData({
-        data,
-        mode: responseMode,
-        latestEventById,
-        entities: service.entities,
-        getId,
-        isItemStale: (current, next) => this.#adapter.isItemStale(current, next),
-        canKeepCurrentItem: item =>
-          !(
-            query.desc.method === 'find' &&
-            query.config.realtime === 'merge' &&
-            !isServerMaintained(query.maintenance.classification) &&
-            // An undecidable item keeps the server's word for it.
-            query.maintenance.matches(item) === false
-          ),
-      })
-      const nextItemIds = new Set(rebasedResponse.itemIds)
-
-      // Projected (`$select`) rows are correct for this query's own result but are
-      // not full entities — never write them to the entity cache, which must hold
-      // only complete rows for the materialized local-answer paths to be sound
-      // (isItemStale can't catch a projection: same updatedAt as the row it shadows).
-      const fetchedEvents: ProcessedCacheEvent[] = []
-      const fetchedRows: QueuedEvent[] = []
-      if (source === 'server' && !isProjection) {
-        for (const item of rebasedResponse.items) {
-          const id = getId(item)
-          if (id === undefined || journaledItemIds.has(entityKey(id))) continue
-          fetchedRows.push({
-            mode: 'server',
-            source: 'fetch',
-            serviceName: query.desc.serviceName,
-            type: service.entities.has(entityKey(id)) ? 'updated' : 'created',
-            item,
-            ...(cause === undefined ? {} : { cause }),
-          })
-        }
-      }
-      applyEventsToService({
+      const plan = buildFetchCommit({
         service,
-        serviceName: query.desc.serviceName,
-        events: fetchedRows,
+        query,
+        result,
+        source,
+        rebasePlan,
+        pendingMutationEvents: this.#mutationExecutor.overlayEvents(query.desc.serviceName),
         getId,
         isItemStale: (current, next) => this.#adapter.isItemStale(current, next),
-        processedEvents: fetchedEvents,
+        meta: ('meta' in result ? result.meta : undefined) || this.#adapter.emptyMeta(),
+        fetchedAt: this.#clock.now(),
+        ...(cause === undefined ? {} : { cause }),
       })
-
-      // `nextItemIds` is also the complete-set diff input. Rebase its membership
-      // to the last in-flight event so a stale root response cannot delete a
-      // created row or resurrect a removed one service-wide.
-      for (const [itemId, event] of latestEventById) {
-        if (event.type === 'removed') {
-          nextItemIds.delete(itemId)
-        } else if (isCompleteSet) {
-          nextItemIds.add(itemId)
-        }
+      for (const laneKey of plan.deferredMutationLaneKeys) {
+        this.#mutationExecutor.deferQueryIds(laneKey, [queryId])
       }
-
+      hadEffectiveJournalEvents = rebasePlan.events.length > 0
       shouldRefetch = this.#executions.get(queryId)?.followup !== undefined
-
-      // A successful unfiltered allPages fetch (`.all()` with no filters) means the
-      // complete row set is now local: mark the service materialized so matcher-
-      // decidable finds are answered from the cache (see #selectMaterializedFind). A *filtered*
-      // allPages fetch is complete only for its own filter — it must not materialize
-      // the service.
-      if (isCompleteSet) {
+      if (plan.completeSet) {
         const previousRoot = service.materialized?.queryId
         if (previousRoot && previousRoot !== queryId && this.#listenerCount(previousRoot) === 0) {
           this.#retention.retain(previousRoot)
         }
-        service.materialized = { queryId, fetchedAt: this.#clock.now() }
       }
-
-      commitQuery(service, {
-        ...query,
-        fetchedAt: this.#clock.now(),
-        state: {
-          status: 'success' as const,
-          data: rebasedResponse.data,
-          meta: meta || this.#adapter.emptyMeta(),
-          ...(pageInfo ? { pageInfo } : {}),
-          isFetching: false,
-          error: null,
-        },
+      const changes = applyFetchCommit(service, plan, {
+        getId,
+        isItemStale: (current, next) => this.#adapter.isItemStale(current, next),
       })
-
-      const changes = previousRootEntities
-        ? diffCompleteSet({
-            service,
-            serviceName: query.desc.serviceName,
-            previousEntities: previousRootEntities,
-            nextItemIds,
-            ignoredItemIds: journaledItemIds,
-          })
-        : fetchedEvents
       eventEffects.push(
         ...this.#updateQueriesForEvents({
           service,
@@ -1570,27 +1459,16 @@ export class QueryStore<
           processedEvents: changes.map(event =>
             cause === undefined ? event : { ...event, cause },
           ),
-          // A normal fetch proves the returned entity value, but not that it belongs
-          // to another query. A complete-set diff does prove service membership.
-          membershipScope: previousRootEntities ? 'all-matching' : 'visible-only',
+          membershipScope: plan.completeSet ? 'all-matching' : 'visible-only',
           touch,
           excludeQueryId: queryId,
         }),
       )
-
-      if (
-        effectiveJournalEvents.length > 0 &&
-        (responseMode === 'entity' || responseMode === 'fetch-owned')
-      ) {
+      if (plan.replayEvents.length > 0) {
         replayFetchedQueryFromEvents({
           service,
           queryId,
-          events:
-            responseMode === 'fetch-owned'
-              ? effectiveJournalEvents.filter(
-                  event => event.mode !== 'server' || event.source !== 'realtime',
-                )
-              : effectiveJournalEvents,
+          events: plan.replayEvents,
           touch,
           getId,
           itemAdded: meta => this.#adapter.itemAdded(meta),
@@ -1627,7 +1505,10 @@ export class QueryStore<
     if (shouldRefetch && shouldRunFollowup) {
       serverMaintainedQueriesToRefetch.delete(queryId)
       this.#queue(queryId, this.#takeFollowupFetchContext(queryId))
-    } else if (hadEffectiveJournalEvents && query?.config.realtime !== 'disabled') {
+    } else if (
+      hadEffectiveJournalEvents &&
+      (!query || queryPolicy(query.maintenance).responseMode !== 'snapshot')
+    ) {
       // Even exact replay cannot prove every server-maintained membership/order
       // edge. One gated trailing reconciliation guarantees convergence.
       serverMaintainedQueriesToRefetch.add(queryId)
@@ -1956,7 +1837,7 @@ export class QueryStore<
       for (const [queryId, query] of service.queries) {
         if (
           queryId !== excludeQueryId &&
-          query.config.realtime === 'merge' &&
+          queryPolicy(query.maintenance).mergeEvents &&
           query.state.status === 'success' &&
           this.#materializedFindSource(query)
         ) {
@@ -2237,7 +2118,8 @@ export class QueryStore<
     if (service) {
       for (const invalidation of invalidations) {
         for (const query of service.queries.values()) {
-          if (query.config.skip || query.config.realtime === 'disabled') continue
+          if (query.config.skip || queryPolicy(query.maintenance).responseMode === 'snapshot')
+            continue
           if (
             query.desc.method === 'get' &&
             entityKey(query.desc.resourceId) !== invalidation.itemId
@@ -2296,7 +2178,8 @@ export class QueryStore<
     if (!service) return []
     return [...service.queries.values()]
       .filter(
-        query => query.config.realtime === 'refetch' && this.#listenerCount(query.queryId) > 0,
+        query =>
+          queryPolicy(query.maintenance).refetchEvents && this.#listenerCount(query.queryId) > 0,
       )
       .map(query => query.queryId)
   }
@@ -2436,7 +2319,8 @@ export class QueryStore<
   #reconcilesAfterMissedEvents(query: Query<unknown, TMeta, unknown>): boolean {
     return (
       !query.config.skip &&
-      (query.config.realtime !== 'disabled' || this.#reconnectQueryIds.has(query.queryId))
+      (queryPolicy(query.maintenance).responseMode !== 'snapshot' ||
+        this.#reconnectQueryIds.has(query.queryId))
     )
   }
 
diff --git a/lib/core/relationalQuery.ts b/lib/core/relationalQuery.ts
index b029627..ac68c78 100644
--- a/lib/core/relationalQuery.ts
+++ b/lib/core/relationalQuery.ts
@@ -1,5 +1,5 @@
 import type { Clock } from './clock.js'
-import { QueryLifetime } from './queryLifetime.js'
+import { RelationalQueryLifetime } from './relationalQueryLifetime.js'
 import { ChunkedRelationQuery, type ChunkedRelationSnapshot } from './relationChunks.js'
 import { compileRelations, type RelationPlan, type RelationQueryPlan } from './relationPlan.js'
 import { hashObject } from './hash.js'
@@ -49,7 +49,6 @@ import {
 import type { AnySchema, Schema } from './schema.js'
 import { resolveServicePath } from './schema.js'
 import type { ValueComparator } from './sort.js'
-import { validateStaleTime } from './staleTime.js'
 
 export type { RelationalPaginationState } from './queryRoots.js'
 
@@ -161,15 +160,6 @@ type GatherResult = {
   assembly: Map<string, AssembledRelationData>
 }
 
-type RelationalListener =
-  | { source: 'subscriber'; staleTime: number }
-  | { source: 'prefetch'; staleTime: number; adoptableUntil: number | null }
-  | { source: 'prepare'; staleTime: number; preparationGeneration: number }
-
-type PreparedAdoption =
-  | { kind: 'idle'; adoptedThrough: number }
-  | { kind: 'wave'; generation: number }
-
 export interface InspectedRelationalQuery {
   key: string
   name?: string
@@ -230,13 +220,7 @@ export class RelationalQueryRef<
   // "comments.reactions"). A relation is "synced" once its entry exists here — even a
   // kind:'empty' entry counts, so loading detection doesn't hang on empty relations.
   #relationSubs: Map<string, RelationSub<TMeta>> = new Map()
-  #lifetime = new QueryLifetime<
-    (state: RelationalQueryState<T>) => void,
-    RelationalListener,
-    null
-  >()
-  #nextPreparationGeneration = 0
-  #preparedAdoption: PreparedAdoption = { kind: 'idle', adoptedThrough: 0 }
+  #lifetime: RelationalQueryLifetime<(state: RelationalQueryState<T>) => void>
   #processedEventUnsub: (() => void) | null = null
   #relationalFilterReconcileQueued = false
   // Strictest active subscriber freshness tolerance — applied to newly-created
@@ -267,11 +251,6 @@ export class RelationalQueryRef<
   #lastGatherWasPartial = false
   #assembleRelations: ReturnType<typeof createRelationAssembler> | null = null
 
-  // A Suspense read materializes and fetches the graph before React can commit its
-  // subscription. The first committed listener claims that fetch instead of treating
-  // the just-resolved data as stale and immediately repeating the whole graph.
-  #coldStartAwaitingSubscriber = false
-
   // Relation keys that already produced a fan-out warning — warn once per relation,
   // not on every sync pass.
   #fanOutWarnedKeys: Set<string> = new Set()
@@ -289,7 +268,6 @@ export class RelationalQueryRef<
 
   #onIdle: (() => void) | null
   #onEvict: (() => void) | null = null
-  #defaultStaleTime: number
   #name: string | undefined
   #rootOverride: RelationalRootOverride | null
   #relationPlans: RelationPlan[]
@@ -308,7 +286,7 @@ export class RelationalQueryRef<
     this.#queryId = `rq/${hashObject(options?.root ? { ast, root: options.root } : ast)}`
     this.#onIdle = options?.onIdle ?? null
     this.#onEvict = onEvict ?? null
-    this.#defaultStaleTime = options?.defaultStaleTime ?? 0
+    this.#lifetime = new RelationalQueryLifetime(host.clock, options?.defaultStaleTime ?? 0)
     this.#rootOverride = options?.root ?? null
   }
 
@@ -401,46 +379,13 @@ export class RelationalQueryRef<
       source?: 'subscriber' | 'prepare' | 'prefetch'
     },
   ): () => void {
-    const source = options?.source ?? 'subscriber'
-    const staleTime =
-      options?.staleTime === undefined
-        ? this.#defaultStaleTime
-        : validateStaleTime(options.staleTime, 'query(): staleTime')
-    const listener: RelationalListener =
-      source === 'prepare'
-        ? {
-            source,
-            staleTime,
-            preparationGeneration: ++this.#nextPreparationGeneration,
-          }
-        : source === 'prefetch'
-          ? { source, staleTime, adoptableUntil: this.#host.clock.now() + staleTime }
-          : { source, staleTime }
-    // Preparation makes one freshness decision for the destination's initial React
-    // commit. Every subscriber in that synchronous wave adopts it; later mounts use
-    // their own staleTime even if the router keeps the preparation pinned.
-    const adoptsPreparation = source === 'subscriber' && this.#claimPreparedAdoption()
-    const adoptsPrefetch = source === 'prepare' && this.#claimPrefetch()
-    const claimsColdStart = this.#coldStartAwaitingSubscriber
-    this.#lifetime.acquire(fn, listener)
+    const freshness = this.#lifetime.retain(fn, options)
     this.#staleTime = this.#lifetime.staleTime()
-
     if (!this.#root) {
       this.#setupRoot()
     } else {
       this.#root.setStaleTime(this.#staleTime)
-      if (claimsColdStart) {
-        // React StrictMode may subscribe, unsubscribe, and resubscribe in one turn.
-        // Keep the claim window open through that commit so neither subscription
-        // mistakes the Suspense fetch for stale data.
-        queueMicrotask(() => {
-          this.#coldStartAwaitingSubscriber = false
-        })
-      } else if (!adoptsPreparation) {
-        // Adopt the speculative read once, but still retry errors and pending
-        // invalidations. An explicit preparation staleTime takes precedence.
-        this.#ensureFresh(adoptsPrefetch && options?.staleTime === undefined ? Infinity : staleTime)
-      }
+      if (freshness.kind === 'check') this.#ensureFresh(freshness.staleTime)
     }
 
     // Don't call fn synchronously - useSyncExternalStore will call getSnapshot() instead
@@ -461,39 +406,11 @@ export class RelationalQueryRef<
     }
   }
 
-  #claimPreparedAdoption(): boolean {
-    const adoption = this.#preparedAdoption
-    const adoptedThrough = adoption.kind === 'wave' ? adoption.generation : adoption.adoptedThrough
-    let newestActive = adoptedThrough
-    for (const listener of this.#lifetime.owners.values()) {
-      if (listener.source === 'prepare') {
-        newestActive = Math.max(newestActive, listener.preparationGeneration)
-      }
-    }
-
-    if (newestActive > adoptedThrough) {
-      this.#preparedAdoption = { kind: 'wave', generation: newestActive }
-      queueMicrotask(() => {
-        const current = this.#preparedAdoption
-        if (current.kind === 'wave' && current.generation === newestActive) {
-          this.#preparedAdoption = { kind: 'idle', adoptedThrough: newestActive }
-        }
-      })
-    }
-
-    return this.#preparedAdoption.kind === 'wave'
-  }
-
-  #claimPrefetch(): boolean {
-    const now = this.#host.clock.now()
-    let claimed = false
-    for (const listener of this.#lifetime.owners.values()) {
-      if (listener.source === 'prefetch' && listener.adoptableUntil !== null) {
-        claimed ||= now < listener.adoptableUntil
-        listener.adoptableUntil = null
-      }
-    }
-    return claimed
+  /** @internal Retain an idempotent speculative lease until its freshness window expires. */
+  prefetch(staleTime: number): void {
+    this.#lifetime.prefetch(staleTime, () =>
+      this.subscribe(() => {}, { source: 'prefetch', staleTime }),
+    )
   }
 
   #beginGraphRun(): string | null {
@@ -887,6 +804,7 @@ export class RelationalQueryRef<
   }
 
   #setupRoot(): void {
+    this.#lifetime.graphStarted()
     this.#beginGraphRun()
     this.#subscribeToRelationalFilterInvalidations()
 
@@ -1399,8 +1317,9 @@ export class RelationalQueryRef<
    */
   #settleSuspense(snapshot: RelationalQueryState<T>): void {
     if (snapshot.status !== 'success' && snapshot.status !== 'error') return
-    const wasPending = this.#lifetime.reads.get('root')?.status === 'pending'
-    this.#lifetime.settle('root', null, snapshot.status === 'error' ? snapshot.error : null)
+    const wasPending = this.#lifetime.settleSuspense(
+      snapshot.status === 'error' ? snapshot.error : null,
+    )
     if (wasPending) this.#onIdle?.()
   }
 
@@ -1412,10 +1331,7 @@ export class RelationalQueryRef<
    * eviction, a retry interns a fresh ref and cold-starts.
    */
   releaseColdStart(): void {
-    if (
-      this.#lifetime.reads.get('root')?.status === 'settled' &&
-      this.#lifetime.owners.size === 0
-    ) {
+    if (this.#lifetime.canEvictAbandonedRead()) {
       this.#scheduleCleanup()
     }
   }
@@ -1427,21 +1343,15 @@ export class RelationalQueryRef<
    * to the keep-previous-data path rather than throwing again.
    */
   suspensePromise(): Promise<void> {
-    if (this.#lifetime.reads.get('root')?.status === 'settled') return Promise.resolve()
-    return this.#lifetime.read('root', null, () => {
-      if (!this.#root) {
-        this.#coldStartAwaitingSubscriber = this.#lifetime.owners.size === 0
-        this.#setupRoot()
-      }
+    return this.#lifetime.suspensePromise(() => {
+      if (!this.#root) this.#setupRoot()
       this.#settleSuspense(this.getSnapshot())
     })
   }
 
   /** @internal Release settled speculative work only when no owner adopted it. */
   canEvictAbandonedRead(): boolean {
-    return (
-      this.#lifetime.owners.size === 0 && this.#lifetime.reads.get('root')?.status === 'settled'
-    )
+    return this.#lifetime.canEvictAbandonedRead()
   }
 
   evictAbandonedRead(): boolean {
@@ -1476,14 +1386,7 @@ export class RelationalQueryRef<
     this.#lastGatherWasPartial = false
     this.#assembleRelations = null
     this.#staleTime = 0
-    if (this.#preparedAdoption.kind === 'wave') {
-      this.#preparedAdoption = {
-        kind: 'idle',
-        adoptedThrough: this.#preparedAdoption.generation,
-      }
-    }
     this.#lifetime.reset()
-    this.#coldStartAwaitingSubscriber = false
     this.#onEvict?.()
   }
 }
diff --git a/lib/core/relationalQueryLifetime.ts b/lib/core/relationalQueryLifetime.ts
new file mode 100644
index 0000000..22255a5
--- /dev/null
+++ b/lib/core/relationalQueryLifetime.ts
@@ -0,0 +1,153 @@
+import type { Clock, ClockTimer } from './clock.js'
+import { QueryLifetime } from './queryLifetime.js'
+import { isWithinStaleTime, validateStaleTime } from './staleTime.js'
+
+type QueryLease =
+  | { source: 'subscriber'; staleTime: number }
+  | { source: 'prefetch'; staleTime: number; adoptableUntil: number | null }
+  | { source: 'prepare'; staleTime: number; preparationGeneration: number }
+
+type PreparedAdoption =
+  | { kind: 'idle'; adoptedThrough: number }
+  | { kind: 'wave'; generation: number }
+
+type FreshnessDecision = { kind: 'adopt' } | { kind: 'check'; staleTime: number }
+
+interface PrefetchLease {
+  at: number
+  release: () => void
+  timer: ClockTimer
+}
+
+/** Owns graph leases, freshness adoption, and the prefetch deadline for one query identity. */
+export class RelationalQueryLifetime<TListener> extends QueryLifetime<TListener, QueryLease, null> {
+  readonly #clock: Clock
+  readonly #defaultStaleTime: number
+  #nextPreparationGeneration = 0
+  #preparedAdoption: PreparedAdoption = { kind: 'idle', adoptedThrough: 0 }
+  #coldStartAwaitingSubscriber = false
+  #prefetch: PrefetchLease | null = null
+
+  constructor(clock: Clock, defaultStaleTime: number) {
+    super()
+    this.#clock = clock
+    this.#defaultStaleTime = defaultStaleTime
+  }
+
+  retain(
+    listener: TListener,
+    options?: { source?: QueryLease['source']; staleTime?: number | undefined },
+  ): FreshnessDecision {
+    const source = options?.source ?? 'subscriber'
+    const staleTime =
+      options?.staleTime === undefined
+        ? this.#defaultStaleTime
+        : validateStaleTime(options.staleTime, 'query(): staleTime')
+    const lease: QueryLease =
+      source === 'prepare'
+        ? { source, staleTime, preparationGeneration: ++this.#nextPreparationGeneration }
+        : source === 'prefetch'
+          ? { source, staleTime, adoptableUntil: this.#clock.now() + staleTime }
+          : { source, staleTime }
+    const adoptsPreparation = source === 'subscriber' && this.#claimPreparedAdoption()
+    const adoptsPrefetch = source === 'prepare' && this.#claimPrefetch()
+    this.acquire(listener, lease)
+
+    if (this.#coldStartAwaitingSubscriber) {
+      // Keep adoption open through the commit's immediate StrictMode resubscription.
+      queueMicrotask(() => {
+        this.#coldStartAwaitingSubscriber = false
+      })
+      return { kind: 'adopt' }
+    }
+    return adoptsPreparation
+      ? { kind: 'adopt' }
+      : {
+          kind: 'check',
+          staleTime: adoptsPrefetch && options?.staleTime === undefined ? Infinity : staleTime,
+        }
+  }
+
+  prefetch(staleTime: number, retain: () => () => void): void {
+    const now = this.#clock.now()
+    if (this.#prefetch && isWithinStaleTime(this.#prefetch.at, staleTime, now)) return
+    this.#releasePrefetch()
+    const release = retain()
+    const timer = this.#clock.setTimeout(() => this.#releasePrefetch(), staleTime)
+    timer.unref()
+    this.#prefetch = { at: now, release, timer }
+  }
+
+  graphStarted(): void {
+    this.#coldStartAwaitingSubscriber = this.owners.size === 0
+  }
+
+  canEvictAbandonedRead(): boolean {
+    return this.owners.size === 0 && this.reads.get('root')?.status === 'settled'
+  }
+
+  suspensePromise(start: () => void): Promise<void> {
+    if (this.reads.get('root')?.status === 'settled') return Promise.resolve()
+    return this.read('root', null, start)
+  }
+
+  settleSuspense(error: Error | null): boolean {
+    const wasPending = this.reads.get('root')?.status === 'pending'
+    this.settle('root', null, error)
+    return wasPending
+  }
+
+  override reset(): void {
+    if (this.#preparedAdoption.kind === 'wave') {
+      this.#preparedAdoption = { kind: 'idle', adoptedThrough: this.#preparedAdoption.generation }
+    }
+    this.#coldStartAwaitingSubscriber = false
+    super.reset()
+  }
+
+  override dispose(): void {
+    this.#releasePrefetch()
+    super.dispose()
+  }
+
+  #releasePrefetch(): void {
+    const lease = this.#prefetch
+    this.#prefetch = null
+    if (!lease) return
+    lease.timer.cancel()
+    lease.release()
+  }
+
+  #claimPreparedAdoption(): boolean {
+    const adoption = this.#preparedAdoption
+    const adoptedThrough = adoption.kind === 'wave' ? adoption.generation : adoption.adoptedThrough
+    let newestActive = adoptedThrough
+    for (const lease of this.owners.values()) {
+      if (lease.source === 'prepare') {
+        newestActive = Math.max(newestActive, lease.preparationGeneration)
+      }
+    }
+    if (newestActive > adoptedThrough) {
+      this.#preparedAdoption = { kind: 'wave', generation: newestActive }
+      queueMicrotask(() => {
+        const current = this.#preparedAdoption
+        if (current.kind === 'wave' && current.generation === newestActive) {
+          this.#preparedAdoption = { kind: 'idle', adoptedThrough: newestActive }
+        }
+      })
+    }
+    return this.#preparedAdoption.kind === 'wave'
+  }
+
+  #claimPrefetch(): boolean {
+    const now = this.#clock.now()
+    let claimed = false
+    for (const lease of this.owners.values()) {
+      if (lease.source === 'prefetch' && lease.adoptableUntil !== null) {
+        claimed ||= now < lease.adoptableUntil
+        lease.adoptableUntil = null
+      }
+    }
+    return claimed
+  }
+}
diff --git a/lib/core/windowMaintenance.ts b/lib/core/windowMaintenance.ts
index 0b3130a..cae5347 100644
--- a/lib/core/windowMaintenance.ts
+++ b/lib/core/windowMaintenance.ts
@@ -3,8 +3,8 @@ import { sameValue } from './valueEquality.js'
 import { commitQuery } from './queryResults.js'
 /**
  * Window maintenance and realtime event application — the pure algebra the query
- * store orchestrates. Everything here is a plain function over explicit inputs
- * (service state, queries, queued events); no store instance, no side channels.
+ * store orchestrates. Event application uses service state, queries, queued events,
+ * and their compiled policies without a store instance.
  *
  * The heart of the module is `mergeEventIntoWindow`, which decides whether a
  * realtime event's effect on a server-window query ($limit/$skip/$sort) is
@@ -12,6 +12,7 @@ import { commitQuery } from './queryResults.js'
  * "refetch" when it is not. The soundness argument lives on the function.
  */
 
+import { queryPolicy } from './queryMaintenance.js'
 import { isServerMaintained } from './queryClassification.js'
 import { ItemRemovedError } from './errors.js'
 import {
@@ -438,8 +439,7 @@ export function updateQueriesFromEvents<TMeta>({
   for (const event of appliedItems) {
     for (const [queryId, query] of service.queries) {
       if (queryId === excludeQueryId || excludeQueryIds?.has(queryId)) continue
-      if (query.config.realtime !== 'merge') continue
-      if (query.desc.method === 'find' && query.config.fetchPolicy === 'network-only') continue
+      if (!queryPolicy(query.maintenance).mergeEvents) continue
       const visible = service.itemQueryIndex.get(event.itemId)?.has(queryId) ?? false
       if (membershipScope === 'visible-only' && !visible) continue
       if (
@@ -512,8 +512,7 @@ export function reapplyQueryFromEntities<TMeta>({
   itemRemoved: (meta: TMeta) => TMeta
 }): QueryReapplyResult {
   const query = service.queries.get(queryId)
-  if (!query || query.config.realtime !== 'merge') return 'ignored'
-  if (query.desc.method === 'find' && query.config.fetchPolicy === 'network-only') return 'ignored'
+  if (!query || !queryPolicy(query.maintenance).mergeEvents) return 'ignored'
   if (isServerMaintained(query.maintenance.classification)) return 'reconcile'
   if (query.desc.method !== 'find' || query.state.status !== 'success') return 'ignored'
   if (!Array.isArray(query.state.data)) return 'ignored'
@@ -605,18 +604,12 @@ export function replayFetchedQueryFromEvents<TMeta>({
   }
   for (const event of events) {
     const query = service.queries.get(queryId)
-    if (!query || query.config.realtime === 'disabled') return
-
-    const canMerge =
-      query.config.realtime === 'merge' &&
-      !(query.desc.method === 'find' && query.config.fetchPolicy === 'network-only')
-    const needsVisibleFallback =
-      query.config.realtime === 'refetch' ||
-      query.config.fetchPolicy === 'network-only' ||
-      isServerMaintained(query.maintenance.classification)
-    if (canMerge) {
+    if (!query || queryPolicy(query.maintenance).responseMode === 'snapshot') return
+
+    const { mergeEvents, replayVisibleEvents } = queryPolicy(query.maintenance)
+    if (mergeEvents) {
       const result = applyMergeEventToQuery(context, queryId, event)
-      if (result === 'applied' || !needsVisibleFallback) continue
+      if (result === 'applied' || !replayVisibleEvents) continue
     }
     applyVisibleEventEffect(
       context,
diff --git a/test/clock.ts b/test/clock.ts
index 62771a3..4dba561 100644
--- a/test/clock.ts
+++ b/test/clock.ts
@@ -13,6 +13,10 @@ export class TestClock implements Clock {
   #nextId = 0
   #timers = new Map<number, { at: number; callback: () => void }>()
 
+  get pendingTimers(): number {
+    return this.#timers.size
+  }
+
   now(): number {
     return this.#now
   }
diff --git a/test/fetch-races.test.ts b/test/fetch-races.test.ts
index 5000102..4870728 100644
--- a/test/fetch-races.test.ts
+++ b/test/fetch-races.test.ts
@@ -12,6 +12,7 @@ import {
 import { FetchEventJournal, MAX_FETCH_JOURNAL_EVENTS } from '../lib/core/fetchRebase'
 import type { ProcessedCacheEvent } from '../lib/core/queryTypes'
 import { mockFeathers, waitFor, type TestItem } from './helpers'
+import { deferred } from './mutation-test-helpers'
 
 interface Note extends TestItem {
   id: number
@@ -738,6 +739,37 @@ test('a mutation acknowledgement survives an in-flight complete-set fetch', asyn
   unsub()
 })
 
+test('snapshot fetches ignore pending mutations while protecting their canonical entities', async t => {
+  const original = { id: 1, content: 'original', rank: 1 }
+  const { figbird, notes } = createApp({ 1: original })
+  t.teardown(() => figbird.dispose())
+  const live = figbird.queryDesc({ serviceName: 'notes', method: 'get', resourceId: 1 })
+  t.teardown(live.subscribe(() => {}))
+  await waitFor(() => live.getSnapshot()?.status === 'success', 'the live entity')
+
+  const acknowledgement = deferred<Note>()
+  notes.patch = () => acknowledgement.promise
+  const pendingWrite = figbird.m.notes.patch(1, { content: 'optimistic' })
+  await flushTasks()
+
+  const snapshot = figbird.queryDesc(
+    { serviceName: 'notes', method: 'find' },
+    { realtime: 'disabled' },
+  )
+  t.teardown(snapshot.subscribe(() => {}))
+  await waitFor(() => snapshot.getSnapshot()?.status === 'success', 'the snapshot fetch')
+  t.deepEqual(snapshot.getSnapshot()?.data, [original])
+  t.like(live.getSnapshot()?.data, { content: 'optimistic' })
+  t.like(figbird.getState().get('notes')?.entities.get('1'), { content: 'optimistic' })
+
+  acknowledgement.resolve({ ...original, content: 'confirmed' })
+  await pendingWrite
+  await flushTasks()
+  t.deepEqual(snapshot.getSnapshot()?.data, [original], 'settlement leaves the snapshot intact')
+  t.like(live.getSnapshot()?.data, { content: 'confirmed' })
+  t.is(notes.counts.find, 1, 'snapshot queries are not deferred for settlement refetches')
+})
+
 test('snapshot and fetch-owned queries retain their own fetched rows', async t => {
   const original = { id: 1, content: 'original', rank: 1, updatedAt: 1 }
   const { figbird, notes } = createApp({ 1: original })
diff --git a/test/query-lifecycle.test.ts b/test/query-lifecycle.test.ts
new file mode 100644
index 0000000..d83e743
--- /dev/null
+++ b/test/query-lifecycle.test.ts
@@ -0,0 +1,94 @@
+import test from 'ava'
+import { FeathersAdapter } from '../lib/adapters/feathers.js'
+import { Figbird } from '../lib/core/figbird.js'
+import { createSchema, service } from '../lib/core/schema.js'
+import { TestClock } from './clock.js'
+import { mockFeathers } from './helpers.js'
+
+function fixture() {
+  const clock = new TestClock()
+  const feathers = mockFeathers({ notes: { data: { 1: { id: 1, content: 'hello' } } } })
+  const schema = createSchema({
+    services: { notes: service<{ item: { id: number; content: string } }>() },
+  })
+  const figbird = new Figbird({ adapter: new FeathersAdapter(feathers), schema, clock })
+  return { clock, feathers, figbird }
+}
+
+test('prefetch releases its graph at the expiry deadline', async t => {
+  const { clock, figbird } = fixture()
+  t.teardown(() => figbird.dispose())
+  figbird.prefetch(figbird.q.notes, { staleTime: 100 })
+  await figbird.query(figbird.q.notes).suspensePromise()
+  await clock.advance(99)
+  t.is(figbird.inspectRelational()[0]?.prefetchCount, 1)
+  await clock.advance(1)
+  t.is(figbird.inspectRelational().length, 0)
+})
+
+test('prefetch deduplicates fresh work and replaces expired leases', async t => {
+  const { clock, feathers, figbird } = fixture()
+  t.teardown(() => figbird.dispose())
+  figbird.prefetch(figbird.q.notes, { staleTime: 100 })
+  await figbird.query(figbird.q.notes).suspensePromise()
+  await clock.advance(0)
+  const timerCount = clock.pendingTimers
+  await clock.advance(10)
+  figbird.prefetch(figbird.q.notes, { staleTime: 100 })
+  t.is(feathers.service('notes').counts.find, 1)
+  figbird.prefetch(figbird.q.notes, { staleTime: 2 })
+  await figbird.query(figbird.q.notes).suspensePromise()
+  t.is(feathers.service('notes').counts.find, 2)
+  t.is(figbird.inspectRelational()[0]?.prefetchCount, 1)
+  await clock.advance(0)
+  t.is(clock.pendingTimers, timerCount, 'replacement cancels the original expiry timer')
+  await clock.advance(1)
+  t.is(figbird.inspectRelational()[0]?.prefetchCount, 1)
+  await clock.advance(1)
+  t.is(figbird.inspectRelational().length, 0, 'the replacement owns the new deadline')
+})
+
+test('disposing an active prefetch cancels its timer and releases its graph', async t => {
+  const { clock, figbird } = fixture()
+  t.teardown(() => figbird.dispose())
+  figbird.prefetch(figbird.q.notes, { staleTime: 100 })
+  await figbird.query(figbird.q.notes).suspensePromise()
+  await clock.advance(0)
+  t.true(clock.pendingTimers > 0)
+  figbird.dispose()
+  t.is(clock.pendingTimers, 0)
+  t.is(figbird.inspectRelational().length, 0)
+  await clock.advance(200)
+  t.is(figbird.getState().size, 0)
+})
+
+test('state subscribers receive the original maintenance shape without internal policy', async t => {
+  const { figbird } = fixture()
+  t.teardown(() => figbird.dispose())
+  const maintenanceKeys: PropertyKey[][] = []
+  const releaseState = figbird.subscribeToStateChanges(state => {
+    for (const service of state.values()) {
+      for (const query of service.queries.values()) {
+        maintenanceKeys.push(Reflect.ownKeys(query.maintenance))
+      }
+    }
+  })
+  t.teardown(releaseState)
+  const ref = figbird.query(figbird.q.notes)
+  const releaseQuery = ref.subscribe(() => {})
+  t.teardown(releaseQuery)
+  await ref.suspensePromise()
+  await figbird.m.notes.patch(1, { content: 'changed' })
+  t.true(maintenanceKeys.length > 0)
+  for (const keys of maintenanceKeys) {
+    t.deepEqual(keys, [
+      'classification',
+      'matches',
+      'matchesLocal',
+      'compare',
+      'limit',
+      'skip',
+      'isProjection',
+    ])
+  }
+})
diff --git a/test/query-maintenance.test.ts b/test/query-maintenance.test.ts
new file mode 100644
index 0000000..bc34cf7
--- /dev/null
+++ b/test/query-maintenance.test.ts
@@ -0,0 +1,112 @@
+import test from 'ava'
+import { compileQueryMaintenance, queryPolicy } from '../lib/core/queryMaintenance.js'
+import type { QueryConfig, QueryDescriptor } from '../lib/core/queryTypes.js'
+import { compareValues } from '../lib/core/sort.js'
+
+// Preserve the rules that preceded the compiler as an independent equivalence oracle.
+test('compiled policies preserve the original rules in all 1,024 combinations', t => {
+  const methods = ['get', 'find'] as const
+  const realtimeModes: QueryConfig['realtime'][] = [undefined, 'merge', 'refetch', 'disabled']
+  const fetchPolicies: QueryConfig['fetchPolicy'][] = [
+    undefined,
+    'swr',
+    'cache-first',
+    'network-only',
+  ]
+  const queries: Record<string, unknown>[] = [
+    {},
+    { active: true },
+    { $limit: 5 },
+    { $skip: 5 },
+    { $sort: { id: 1 } },
+    { $select: ['id'] },
+    { $unknown: true },
+    { $or: [{ $select: ['id'] }] },
+  ]
+  let cases = 0
+  for (const method of methods) {
+    for (const realtime of realtimeModes) {
+      for (const fetchPolicy of fetchPolicies) {
+        for (const server of [false, true]) {
+          for (const allPages of [false, true]) {
+            for (const query of queries) {
+              const config: QueryConfig = {
+                ...(realtime === undefined ? {} : { realtime }),
+                ...(fetchPolicy === undefined ? {} : { fetchPolicy }),
+                server,
+                allPages,
+              }
+              const desc: QueryDescriptor =
+                method === 'get'
+                  ? { method, serviceName: 'notes', resourceId: 1, params: { query } }
+                  : { method, serviceName: 'notes', params: { query } }
+              const maintenance = compileQueryMaintenance({
+                desc,
+                config,
+                defaultSort: undefined,
+                compare: compareValues,
+                localOperators: new Set(),
+                matcher: () => () => true,
+              })
+              const { classification, isProjection } = maintenance
+              const serverMaintained =
+                classification === 'server-window' || classification === 'server-authoritative'
+              const fetchOwned =
+                realtime === 'refetch' ||
+                ((realtime === undefined || realtime === 'merge') &&
+                  classification === 'server-authoritative')
+              const responseMode =
+                realtime === 'disabled'
+                  ? 'snapshot'
+                  : isProjection
+                    ? 'projection'
+                    : fetchOwned
+                      ? 'fetch-owned'
+                      : 'entity'
+              const ownsValues =
+                realtime === 'disabled' ||
+                fetchOwned ||
+                fetchPolicy === 'network-only' ||
+                isProjection
+              const label = JSON.stringify({
+                method,
+                realtime,
+                fetchPolicy,
+                server,
+                allPages,
+                query,
+              })
+              t.deepEqual(
+                queryPolicy(maintenance),
+                {
+                  responseMode,
+                  rowSource: ownsValues ? 'values' : 'entities',
+                  mergeEvents:
+                    realtime === 'merge' && !(method === 'find' && fetchPolicy === 'network-only'),
+                  replayVisibleEvents:
+                    realtime === 'refetch' || fetchPolicy === 'network-only' || serverMaintained,
+                  rebaseMembership: realtime === 'merge' && !serverMaintained,
+                  refetchEvents: realtime === 'refetch',
+                  realtimeStrategy:
+                    realtime === 'disabled'
+                      ? 'manual'
+                      : realtime === 'refetch' || serverMaintained
+                        ? 'refetch'
+                        : 'merge',
+                },
+                label,
+              )
+              t.is(
+                maintenance.matches({}),
+                realtime === 'merge' && classification !== 'server-authoritative',
+                label,
+              )
+              cases++
+            }
+          }
+        }
+      }
+    }
+  }
+  t.is(cases, 1_024)
+})