diff --git a/packages/client/src/HybridQueryHandle.ts b/packages/client/src/HybridQueryHandle.ts
index af65c5de..c3a05215 100644
--- a/packages/client/src/HybridQueryHandle.ts
+++ b/packages/client/src/HybridQueryHandle.ts
@@ -89,6 +89,16 @@ export class HybridQueryHandle<T> {
   // Track server data reception
   private hasReceivedServerData: boolean = false;
 
+  // Cached sorted-results snapshot for synchronous external-store reads
+  // (useSyncExternalStore.getSnapshot). Rebuilt lazily only when the result
+  // set actually changes; a referentially-stable array between mutations is
+  // what keeps React's useSyncExternalStore from looping.
+  private resultsSnapshot: HybridResultItem<T>[] = [];
+  private resultsSnapshotDirty: boolean = true;
+  // True once any result emission (local or server) has occurred — lets a
+  // synchronous consumer derive `loading` without waiting for the first notify.
+  private hasEmitted: boolean = false;
+
   // Pagination info
   private _paginationInfo: PaginationInfo = { hasMore: false, cursorStatus: 'none' };
   private paginationListeners: Set<(info: PaginationInfo) => void> = new Set();
@@ -275,12 +285,37 @@ export class HybridQueryHandle<T> {
   }
 
   private notify(): void {
+    // Result set changed — invalidate the cached snapshot and record emission.
+    this.resultsSnapshotDirty = true;
+    this.hasEmitted = true;
     const results = this.getSortedResults();
     for (const listener of this.listeners) {
       listener(results);
     }
   }
 
+  /**
+   * Synchronous, referentially-stable snapshot of the current sorted result
+   * set for React 18 `useSyncExternalStore`. The array identity only changes
+   * when the result set changes (dirty flag flipped in notify()), preventing
+   * render loops.
+   */
+  public getSnapshot(): HybridResultItem<T>[] {
+    if (this.resultsSnapshotDirty) {
+      this.resultsSnapshot = this.getSortedResults();
+      this.resultsSnapshotDirty = false;
+    }
+    return this.resultsSnapshot;
+  }
+
+  /**
+   * Synchronous metadata for deriving `loading`. `hasEmitted` is true once any
+   * result (local or server) has been delivered.
+   */
+  public getSnapshotMeta(): { hasEmitted: boolean } {
+    return { hasEmitted: this.hasEmitted };
+  }
+
   /**
    * Get sorted results with _key and _score.
    */
diff --git a/packages/client/src/HybridSearchHandle.ts b/packages/client/src/HybridSearchHandle.ts
index 9f824fc5..d8b04c28 100644
--- a/packages/client/src/HybridSearchHandle.ts
+++ b/packages/client/src/HybridSearchHandle.ts
@@ -99,6 +99,13 @@ export class HybridSearchHandle<T = unknown> {
   /** Whether the handle has been disposed */
   private disposed = false;
 
+  // Cached sorted-results snapshot for synchronous external-store reads
+  // (useSyncExternalStore.getSnapshot). Rebuilt lazily only when results change
+  // (dirty flag flipped in notifyListeners). A referentially-stable array
+  // between mutations is what keeps React's useSyncExternalStore from looping.
+  private resultsSnapshot: HybridSearchHandleResult<T>[] = [];
+  private resultsSnapshotDirty = true;
+
   /** Reference to SyncEngine */
   private syncEngine: SyncEngine;
 
@@ -265,6 +272,20 @@ export class HybridSearchHandle<T = unknown> {
     return Array.from(this.results.values()).sort((a, b) => b.score - a.score);
   }
 
+  /**
+   * Synchronous, referentially-stable snapshot of the current sorted results
+   * for React 18 `useSyncExternalStore`. The array identity only changes when
+   * results change (dirty flag flipped in notifyListeners), preventing render
+   * loops.
+   */
+  getSnapshot(): HybridSearchHandleResult<T>[] {
+    if (this.resultsSnapshotDirty) {
+      this.resultsSnapshot = this.getResults();
+      this.resultsSnapshotDirty = false;
+    }
+    return this.resultsSnapshot;
+  }
+
   /**
    * Update the search query text.
    * Sends HYBRID_SEARCH_UNSUB for the current subscriptionId, clears local results,
@@ -393,6 +414,8 @@ export class HybridSearchHandle<T = unknown> {
    * misbehaving listener cannot silence the rest.
    */
   private notifyListeners(): void {
+    // Results changed — invalidate the cached snapshot for getSnapshot().
+    this.resultsSnapshotDirty = true;
     const results = this.getResults();
     for (const listener of this.listeners) {
       try {
diff --git a/packages/client/src/QueryHandle.ts b/packages/client/src/QueryHandle.ts
index 0f42a498..51a2696e 100644
--- a/packages/client/src/QueryHandle.ts
+++ b/packages/client/src/QueryHandle.ts
@@ -75,6 +75,19 @@ export class QueryHandle<T> {
   // otherwise accumulated pages beyond `limit` would be hidden.
   private _paginated: boolean = false;
 
+  // Cached sorted-results snapshot for synchronous external-store reads
+  // (useSyncExternalStore.getSnapshot). Rebuilt lazily only when the result
+  // set actually changes (the dirty flag flips on every notify()). Returning a
+  // referentially-stable array between mutations is what keeps React's
+  // useSyncExternalStore from looping; building a fresh array on every read
+  // would.
+  private resultsSnapshot: (T & { _key: string })[] = [];
+  private resultsSnapshotDirty: boolean = true;
+  // True once the handle has produced ANY result emission (local or server) —
+  // lets a synchronous consumer derive `loading` without waiting for the first
+  // async notify.
+  private hasEmitted: boolean = false;
+
   // Change tracking for delta notifications
   private changeTracker = new ChangeTracker<T>();
   private pendingChanges: ChangeEvent<T>[] = [];
@@ -121,14 +134,21 @@ export class QueryHandle<T> {
     // [FIX]: Attempt to load local results immediately if available
     // This ensures that if data is already in storage but sync hasn't happened,
     // we still show something.
-    this.loadInitialLocalData().then((data) => {
-      // If we haven't received server results yet (currentResults empty),
-      // and we have local data OR it's just the initial load, we should notify.
-      // Even if data is empty, we might want to tell the subscriber "nothing here yet".
-      if (this.currentResults.size === 0) {
-        this.onResult(data, 'local');
-      }
-    });
+    this.loadInitialLocalData()
+      .then((data) => {
+        // If we haven't received server results yet (currentResults empty),
+        // and we have local data OR it's just the initial load, we should notify.
+        // Even if data is empty, we might want to tell the subscriber "nothing here yet".
+        if (this.currentResults.size === 0) {
+          this.onResult(data, 'local');
+        }
+      })
+      .catch((err) => {
+        // Best-effort pre-load. Some clauses (groupBy/aggregations/cursor) cannot
+        // be evaluated offline and reject by design — skip the local snapshot and
+        // let the authoritative server result settle the query.
+        logger.debug({ err }, 'local query pre-load skipped (not available offline)');
+      });
 
     return () => {
       this.listeners.delete(callback);
@@ -368,6 +388,10 @@ export class QueryHandle<T> {
   }
 
   private notify() {
+    // The result set changed — invalidate the cached snapshot so the next
+    // getSnapshot() rebuilds it, and record that we have emitted at least once.
+    this.resultsSnapshotDirty = true;
+    this.hasEmitted = true;
     const results = this.getSortedResults();
     // Snapshot the latch once per emission so every subscriber in this pass
     // observes the same settled value.
@@ -423,6 +447,36 @@ export class QueryHandle<T> {
     return results;
   }
 
+  /**
+   * Synchronous, referentially-stable snapshot of the current sorted result
+   * set. Designed for React 18 `useSyncExternalStore(subscribe, getSnapshot)`:
+   * the returned array identity only changes when the result set actually
+   * changes (tracked via the dirty flag flipped in notify()), so a render that
+   * reads this between mutations gets the same array and React does not loop.
+   *
+   * Returns whatever the handle currently holds — including any local
+   * pre-loaded rows — so the FIRST render after subscribing reflects cached
+   * data instead of an empty array. (loadInitialLocalData runs on subscribe;
+   * if data is already cached it is reflected on the next getSnapshot.)
+   */
+  public getSnapshot(): (T & { _key: string })[] {
+    if (this.resultsSnapshotDirty) {
+      this.resultsSnapshot = this.getSortedResults();
+      this.resultsSnapshotDirty = false;
+    }
+    return this.resultsSnapshot;
+  }
+
+  /**
+   * Synchronous metadata for deriving `loading` without an async round-trip.
+   * `settled` is the server-answered latch; `hasEmitted` is true once any
+   * result (local or server) has been delivered. A consumer can show data
+   * immediately when `hasEmitted` is true even before the server settles.
+   */
+  public getSnapshotMeta(): { settled: boolean; hasEmitted: boolean } {
+    return { settled: this.settled, hasEmitted: this.hasEmitted };
+  }
+
   public getFilter(): QueryFilter {
     return this.filter;
   }
diff --git a/packages/client/src/SearchHandle.ts b/packages/client/src/SearchHandle.ts
index 1c2aa526..8229750c 100644
--- a/packages/client/src/SearchHandle.ts
+++ b/packages/client/src/SearchHandle.ts
@@ -68,6 +68,13 @@ export class SearchHandle<T = unknown> {
   /** Whether the handle has been disposed */
   private disposed = false;
 
+  // Cached sorted-results snapshot for synchronous external-store reads
+  // (useSyncExternalStore.getSnapshot). Rebuilt lazily only when results change
+  // (dirty flag flipped in notifyListeners). A referentially-stable array
+  // between mutations is what keeps React's useSyncExternalStore from looping.
+  private resultsSnapshot: SearchResult<T>[] = [];
+  private resultsSnapshotDirty = true;
+
   /** Reference to SyncEngine */
   private syncEngine: SyncEngine;
 
@@ -154,6 +161,21 @@ export class SearchHandle<T = unknown> {
     return sorted;
   }
 
+  /**
+   * Synchronous, referentially-stable snapshot of the current sorted results
+   * for React 18 `useSyncExternalStore`. The array identity only changes when
+   * results change (dirty flag flipped in notifyListeners), preventing render
+   * loops. Reflects whatever results are currently cached, so the first render
+   * after subscribing shows them synchronously.
+   */
+  getSnapshot(): SearchResult<T>[] {
+    if (this.resultsSnapshotDirty) {
+      this.resultsSnapshot = this.getResults();
+      this.resultsSnapshotDirty = false;
+    }
+    return this.resultsSnapshot;
+  }
+
   /**
    * Get result count.
    */
@@ -346,6 +368,8 @@ export class SearchHandle<T = unknown> {
    * Notify all listeners of result changes.
    */
   private notifyListeners(): void {
+    // Results changed — invalidate the cached snapshot for getSnapshot().
+    this.resultsSnapshotDirty = true;
     const results = this.getResults();
     for (const listener of this.listeners) {
       try {
diff --git a/packages/client/src/SyncEngine.ts b/packages/client/src/SyncEngine.ts
index 592cbbb5..743f1432 100644
--- a/packages/client/src/SyncEngine.ts
+++ b/packages/client/src/SyncEngine.ts
@@ -1174,8 +1174,21 @@ export class SyncEngine {
     const localMap = this.maps.get(mapName);
     if (localMap) {
       if (localMap instanceof LWWMap && record) {
-        localMap.merge(key, record);
-        await this.storageAdapter.put(`${mapName}:${key}`, record);
+        const accepted = localMap.merge(key, record);
+        if (accepted) {
+          // Server record won LWW: memory now holds it — persist so disk matches.
+          await this.storageAdapter.put(`${mapName}:${key}`, record);
+        } else if (localMap.adoptServerEcho(key, record)) {
+          // Echo of our own optimistic write that the server re-stamped with a
+          // timestamp our clock had briefly outrun. The value is unchanged, so
+          // adopt the server's authoritative timestamp to converge memory, the
+          // Merkle tree, and disk with the server instead of re-requesting
+          // buckets forever (the memory/disk HLC skew this used to create).
+          await this.storageAdapter.put(`${mapName}:${key}`, record);
+        }
+        // else: a strictly newer local write supersedes the echo. Keep memory
+        // and disk as-is (both carry the local write); do NOT persist the stale
+        // server record — that unconditional put was the F8 skew bug.
       } else if (localMap instanceof ORMap) {
         if (eventType === 'OR_ADD' && orRecord) {
           localMap.apply(key, orRecord);
diff --git a/packages/client/src/__tests__/QueryManager.test.ts b/packages/client/src/__tests__/QueryManager.test.ts
index cad213f5..d4d207e7 100644
--- a/packages/client/src/__tests__/QueryManager.test.ts
+++ b/packages/client/src/__tests__/QueryManager.test.ts
@@ -219,6 +219,61 @@ describe('QueryManager', () => {
       expect(results[0].value.price).toBe(150);
       expect(results[1].value.price).toBe(200);
     });
+
+    // F11 (TODO-501): offline parity — the local engine used to ignore
+    // sort/limit and silently dropped cursor/groupBy/aggregations, so offline
+    // results diverged from the server's.
+    describe('offline parity: sort / limit / unsupported clauses (F11)', () => {
+      beforeEach(() => {
+        mockStorage.getAllKeys.mockResolvedValue(['items:1', 'items:2', 'items:3']);
+        mockStorage.get.mockImplementation(async (key: string) => {
+          const data: Record<string, any> = {
+            'items:1': { value: { name: 'b', price: 20 } },
+            'items:2': { value: { name: 'a', price: 30 } },
+            'items:3': { value: { name: 'c', price: 10 } },
+          };
+          return data[key];
+        });
+      });
+
+      it('applies sort offline (ascending) instead of returning storage order', async () => {
+        const results = await queryManager.runLocalQuery('items', { sort: { price: 'asc' } });
+        expect(results.map((r) => r.value.price)).toEqual([10, 20, 30]);
+      });
+
+      it('applies sort offline (descending)', async () => {
+        const results = await queryManager.runLocalQuery('items', { sort: { name: 'desc' } });
+        expect(results.map((r) => r.value.name)).toEqual(['c', 'b', 'a']);
+      });
+
+      it('clamps to limit offline after sorting (server-shaped top-N)', async () => {
+        const results = await queryManager.runLocalQuery('items', {
+          sort: { price: 'asc' },
+          limit: 2,
+        });
+        expect(results.map((r) => r.value.price)).toEqual([10, 20]);
+      });
+
+      it('rejects groupBy offline instead of silently ignoring it', async () => {
+        await expect(queryManager.runLocalQuery('items', { groupBy: ['name'] })).rejects.toThrow(
+          /offline query does not support groupBy/i,
+        );
+      });
+
+      it('rejects aggregations offline', async () => {
+        await expect(
+          queryManager.runLocalQuery('items', {
+            aggregations: [{ func: 'sum', field: 'price' } as any],
+          }),
+        ).rejects.toThrow(/aggregations/i);
+      });
+
+      it('rejects cursor offline', async () => {
+        await expect(queryManager.runLocalQuery('items', { cursor: 'abc' })).rejects.toThrow(
+          /cursor/i,
+        );
+      });
+    });
   });
 
   describe('runLocalHybridQuery', () => {
diff --git a/packages/client/src/__tests__/SyncEngine.test.ts b/packages/client/src/__tests__/SyncEngine.test.ts
index f46f492f..4b22fe1d 100644
--- a/packages/client/src/__tests__/SyncEngine.test.ts
+++ b/packages/client/src/__tests__/SyncEngine.test.ts
@@ -772,6 +772,85 @@ describe('SyncEngine', () => {
       );
     });
 
+    test('adopts a rejected echo of our own write so disk matches memory (F8)', async () => {
+      // The client writes optimistically with a timestamp ahead of the server.
+      // The server re-stamps with a lower (arrival-order) timestamp and echoes
+      // the same value back. The echo loses LWW, but the old code persisted it
+      // unconditionally — disk got the server ts while memory kept the client ts
+      // (Merkle skew). The fix adopts the server record so memory AND disk land
+      // on the server timestamp.
+      syncEngine = new SyncEngine(config);
+      const hlc = syncEngine.getHLC();
+      const lwwMap = new LWWMap<string, any>(hlc);
+      syncEngine.registerMap('users', lwwMap);
+      await jest.runAllTimersAsync();
+
+      const value = { name: 'Alice' };
+      // Optimistic local write with a timestamp far ahead of the server.
+      lwwMap.merge('user1', {
+        value,
+        timestamp: { millis: 9_000_000, counter: 0, nodeId: 'client' },
+      });
+      mockStorage.put.mockClear();
+
+      const serverStamp = { millis: 1_000_000, counter: 0, nodeId: 'server' };
+      const ws = MockWebSocket.getLastInstance()!;
+      ws.simulateMessage({
+        type: 'SERVER_EVENT',
+        payload: {
+          mapName: 'users',
+          eventType: 'PUT',
+          key: 'user1',
+          record: { value, timestamp: serverStamp },
+        },
+      });
+      await jest.runAllTimersAsync();
+
+      // Memory adopted the server's authoritative timestamp (converged)...
+      expect(lwwMap.getRecord('user1')?.timestamp).toEqual(serverStamp);
+      // ...and disk was written with the same server record (no skew).
+      expect(mockStorage.put).toHaveBeenCalledWith(
+        'users:user1',
+        expect.objectContaining({ timestamp: serverStamp }),
+      );
+    });
+
+    test('does NOT persist a stale echo superseded by a newer local write (F8 no data loss)', async () => {
+      syncEngine = new SyncEngine(config);
+      const hlc = syncEngine.getHLC();
+      const lwwMap = new LWWMap<string, any>(hlc);
+      syncEngine.registerMap('users', lwwMap);
+      await jest.runAllTimersAsync();
+
+      // A newer local write the server hasn't seen yet.
+      const newer = { name: 'Bob' };
+      lwwMap.merge('user1', {
+        value: newer,
+        timestamp: { millis: 9_000_000, counter: 0, nodeId: 'client' },
+      });
+      mockStorage.put.mockClear();
+
+      // A stale echo of an OLDER write (different value, lower ts) arrives.
+      const ws = MockWebSocket.getLastInstance()!;
+      ws.simulateMessage({
+        type: 'SERVER_EVENT',
+        payload: {
+          mapName: 'users',
+          eventType: 'PUT',
+          key: 'user1',
+          record: {
+            value: { name: 'Alice' },
+            timestamp: { millis: 1_000_000, counter: 0, nodeId: 'server' },
+          },
+        },
+      });
+      await jest.runAllTimersAsync();
+
+      // The newer local write is preserved, and the stale echo was NOT persisted.
+      expect(lwwMap.get('user1')).toEqual(newer);
+      expect(mockStorage.put).not.toHaveBeenCalled();
+    });
+
     test('should handle SERVER_EVENT for ORMap add', async () => {
       syncEngine = new SyncEngine(config);
       const hlc = syncEngine.getHLC();
diff --git a/packages/client/src/sync/QueryManager.ts b/packages/client/src/sync/QueryManager.ts
index 5bc9c5d6..c0c44352 100644
--- a/packages/client/src/sync/QueryManager.ts
+++ b/packages/client/src/sync/QueryManager.ts
@@ -193,6 +193,10 @@ export class QueryManager implements IQueryManager {
     filter: QueryFilter,
     // eslint-disable-next-line @typescript-eslint/no-explicit-any -- local query returns raw record values whose type is unknown at the query manager level; callers cast to T
   ): Promise<{ key: string; value: any }[]> {
+    // Fail loudly on clauses we cannot evaluate offline instead of silently
+    // returning results that differ from the server.
+    this.assertOfflineSupported(filter);
+
     // Retrieve all keys for the map
     const keys = await this.config.storageAdapter.getAllKeys();
     const mapKeys = keys.filter((k) => k.startsWith(mapName + ':'));
@@ -229,9 +233,73 @@ export class QueryManager implements IQueryManager {
         }
       }
     }
+
+    // Apply the same ordering + windowing the server would, so an offline
+    // `query`/`queryOnce({ allowLocal: true })` returns results shaped like the
+    // online answer instead of an unsorted, unlimited set.
+    if (filter.sort) {
+      const sort = filter.sort;
+      results.sort((a, b) => this.compareBySort(a, b, sort));
+    }
+    if (filter.limit) {
+      return results.slice(0, filter.limit);
+    }
     return results;
   }
 
+  /**
+   * Clauses the offline local query engine cannot faithfully evaluate. Rather
+   * than silently drop them — returning results that diverge from the server's —
+   * reject explicitly so the caller knows the query needs a server connection.
+   * (`sort` and `limit` ARE applied offline; only these remain server-only.)
+   */
+  private assertOfflineSupported(filter: QueryFilter): void {
+    const unsupported: string[] = [];
+    if (filter.cursor) unsupported.push('cursor');
+    if (filter.groupBy && filter.groupBy.length > 0) unsupported.push('groupBy');
+    if (filter.aggregations && filter.aggregations.length > 0) unsupported.push('aggregations');
+    if (unsupported.length > 0) {
+      throw new Error(
+        `Offline query does not support ${unsupported.join(', ')}: ` +
+          'these clauses require a server connection. Run the query online, or ' +
+          'remove them for offline use.',
+      );
+    }
+  }
+
+  /**
+   * Sort comparator shared by the offline standard and hybrid local query
+   * engines so the two cannot diverge. Honors the `_score` and `_key`
+   * pseudo-fields (score defaults to 0 for the non-hybrid path).
+   */
+  private compareBySort<R extends { key: string; value: unknown; score?: number }>(
+    a: R,
+    b: R,
+    sort: Record<string, 'asc' | 'desc'>,
+  ): number {
+    for (const [field, direction] of Object.entries(sort)) {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any -- dynamic field access by runtime sort field name
+      let valA: any;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any -- dynamic field access by runtime sort field name
+      let valB: any;
+      if (field === '_score') {
+        valA = a.score ?? 0;
+        valB = b.score ?? 0;
+      } else if (field === '_key') {
+        valA = a.key;
+        valB = b.key;
+      } else {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any -- dynamic field access on typed value
+        valA = (a.value as any)[field];
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any -- dynamic field access on typed value
+        valB = (b.value as any)[field];
+      }
+      if (valA < valB) return direction === 'asc' ? -1 : 1;
+      if (valA > valB) return direction === 'asc' ? 1 : -1;
+    }
+    return 0;
+  }
+
   /**
    * Run a local hybrid query (FTS + filter combination).
    * For FTS predicates, returns results with score = 0 (local-only mode).
@@ -291,33 +359,10 @@ export class QueryManager implements IQueryManager {
       });
     }
 
-    // Sort results
+    // Sort results (shared comparator with the standard offline path)
     if (filter.sort) {
-      results.sort((a, b) => {
-        for (const [field, direction] of Object.entries(filter.sort!)) {
-          // eslint-disable-next-line @typescript-eslint/no-explicit-any -- sort comparator accesses dynamic field keys on typed results; any is used to index by runtime field name
-          let valA: any;
-          // eslint-disable-next-line @typescript-eslint/no-explicit-any -- sort comparator accesses dynamic field keys on typed results
-          let valB: any;
-
-          if (field === '_score') {
-            valA = a.score ?? 0;
-            valB = b.score ?? 0;
-          } else if (field === '_key') {
-            valA = a.key;
-            valB = b.key;
-          } else {
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any -- dynamic field access on typed value; sort field name is a runtime string
-            valA = (a.value as any)[field];
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any -- dynamic field access on typed value; sort field name is a runtime string
-            valB = (b.value as any)[field];
-          }
-
-          if (valA < valB) return direction === 'asc' ? -1 : 1;
-          if (valA > valB) return direction === 'asc' ? 1 : -1;
-        }
-        return 0;
-      });
+      const sort = filter.sort;
+      results.sort((a, b) => this.compareBySort(a, b, sort));
     }
 
     // Apply limit (cursor filtering is done server-side)
diff --git a/packages/core/src/LWWMap.ts b/packages/core/src/LWWMap.ts
index d8834bd1..ce0d338d 100644
--- a/packages/core/src/LWWMap.ts
+++ b/packages/core/src/LWWMap.ts
@@ -11,6 +11,41 @@ export interface LWWRecord<V> {
   ttlMs?: number;
 }
 
+/**
+ * Structural equality for record values, used to decide whether a rejected
+ * server echo is a pure re-stamp of the same data (safe to adopt) or a
+ * genuinely different (newer local) write that must be preserved. Order-stable
+ * over object keys so two structurally-equal values that survived a serialize/
+ * deserialize round-trip compare equal regardless of key insertion order.
+ */
+function valuesEqual(a: unknown, b: unknown, depth = 0): boolean {
+  if (a === b) return true;
+  if (a === null || b === null || a === undefined || b === undefined) {
+    return a === b;
+  }
+  if (typeof a !== 'object' || typeof b !== 'object') return false;
+
+  // Depth guard: record values reach this only via the server-echo reconcile
+  // path and normally round-trip through msgpack (which rejects cycles), but a
+  // pathological or cyclic in-memory value must not stack-overflow the sync
+  // handler. Treat over-deep structures as "not equal" → conservatively skip
+  // adoption rather than crash.
+  if (depth > 64) return false;
+
+  if (Array.isArray(a) || Array.isArray(b)) {
+    if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length) return false;
+    return a.every((item, i) => valuesEqual(item, b[i], depth + 1));
+  }
+
+  const aObj = a as Record<string, unknown>;
+  const bObj = b as Record<string, unknown>;
+  const aKeys = Object.keys(aObj);
+  const bKeys = Object.keys(bObj);
+  if (aKeys.length !== bKeys.length) return false;
+  const hasOwn = Object.prototype.hasOwnProperty;
+  return aKeys.every((k) => hasOwn.call(bObj, k) && valuesEqual(aObj[k], bObj[k], depth + 1));
+}
+
 /**
  * Last-Write-Wins Map Implementation.
  * This structure guarantees convergence by always keeping the entry with the highest timestamp.
@@ -145,6 +180,49 @@ export class LWWMap<K, V> {
     return false;
   }
 
+  /**
+   * Reconciles the server's authoritative re-stamp of one of this client's own
+   * optimistic writes.
+   *
+   * Optimistic writes are stamped with the client's HLC; the server re-stamps
+   * with its own arrival-order HLC and echoes the record back. When the client
+   * clock briefly outran the server, the echo's timestamp is ≤ the local one,
+   * so {@link merge} rejects it and memory keeps the client timestamp. Because
+   * the Merkle tree hashes only `key:timestamp` (not the value), that leaves the
+   * client's Merkle root permanently diverged from the server's — the same value
+   * under two different timestamps — defeating the "root-equal ⇒ in-sync" fast
+   * path and forcing endless bucket re-requests until a reload re-hydrates from
+   * disk.
+   *
+   * Adopting the server's record aligns memory, the Merkle tree, and (via the
+   * caller's persistence) disk with the server. This is only safe when the local
+   * value is unchanged: if it differs, the client holds a strictly newer write
+   * the server has not seen yet, which must be preserved (no data loss).
+   *
+   * Call this only after {@link merge} has returned `false` for a server echo.
+   *
+   * @returns `true` if the server record was adopted (the caller should persist
+   *   it so disk matches memory); `false` if a newer local write supersedes the
+   *   echo (the caller must NOT persist the stale server record).
+   */
+  public adoptServerEcho(key: K, serverRecord: LWWRecord<V>): boolean {
+    const localRecord = this.data.get(key);
+    if (!localRecord || !valuesEqual(localRecord.value, serverRecord.value)) {
+      return false;
+    }
+    // Defensive precondition: this path only reconciles an echo that LWW
+    // rejected because the server timestamp is ≤ ours. If the server record is
+    // strictly newer, plain `merge` already wins it — never downgrade a newer
+    // record through here (guards against misuse if merge semantics change).
+    if (HLC.compare(serverRecord.timestamp, localRecord.timestamp) > 0) {
+      return false;
+    }
+    this.data.set(key, serverRecord);
+    this.merkleTree.update(String(key), serverRecord);
+    this.notify();
+    return true;
+  }
+
   /**
    * Garbage Collection: Prunes tombstones older than the specified timestamp.
    * Only removes records that are tombstones (deleted) AND older than the threshold.
diff --git a/packages/core/src/__tests__/LWWMap.test.ts b/packages/core/src/__tests__/LWWMap.test.ts
index c6f563c9..0af46c88 100644
--- a/packages/core/src/__tests__/LWWMap.test.ts
+++ b/packages/core/src/__tests__/LWWMap.test.ts
@@ -150,3 +150,83 @@ describe('LWWMap (Last-Write-Wins CRDT)', () => {
     jest.restoreAllMocks();
   });
 });
+
+describe('LWWMap.adoptServerEcho (optimistic re-stamp reconciliation, F8)', () => {
+  // Same key + value under two timestamps: the client's optimistic HLC (which
+  // briefly outran the server) and the server's authoritative arrival-order HLC.
+  const key = 'doc';
+  const value = { title: 'hello', n: 1 };
+  const clientStamp = { millis: 2000, counter: 0, nodeId: 'client' };
+  const serverStamp = { millis: 1000, counter: 0, nodeId: 'server' };
+
+  function clientMapWithOptimisticWrite() {
+    const map = new LWWMap<string, typeof value>(new HLC('client'));
+    // Optimistic local write carrying the (ahead-of-server) client timestamp.
+    map.merge(key, { value, timestamp: clientStamp });
+    return map;
+  }
+
+  function serverMerkleRoot() {
+    const server = new LWWMap<string, typeof value>(new HLC('server'));
+    server.merge(key, { value, timestamp: serverStamp });
+    return server.getMerkleTree().getRootHash();
+  }
+
+  test('rejected echo of our own write diverges the Merkle root (the F8 bug)', () => {
+    // Negative control: without adoption, the old unconditional-merge behavior
+    // leaves the client on the client timestamp while the server is on the
+    // server timestamp — the Merkle roots disagree even though the value is
+    // identical, which is exactly the perpetual bucket-re-request bug.
+    const client = clientMapWithOptimisticWrite();
+    const accepted = client.merge(key, { value, timestamp: serverStamp });
+    expect(accepted).toBe(false); // server ts < client ts → rejected
+    expect(client.getMerkleTree().getRootHash()).not.toBe(serverMerkleRoot());
+  });
+
+  test('adopting the rejected echo converges the Merkle root with the server', () => {
+    const client = clientMapWithOptimisticWrite();
+    const accepted = client.merge(key, { value, timestamp: serverStamp });
+    expect(accepted).toBe(false);
+
+    const adopted = client.adoptServerEcho(key, { value, timestamp: serverStamp });
+    expect(adopted).toBe(true);
+
+    // Memory now carries the server's authoritative timestamp...
+    expect(client.getRecord(key)?.timestamp).toEqual(serverStamp);
+    // ...and the value is unchanged...
+    expect(client.get(key)).toEqual(value);
+    // ...so the Merkle root matches the server's after a single round-trip.
+    expect(client.getMerkleTree().getRootHash()).toBe(serverMerkleRoot());
+  });
+
+  test('does NOT adopt when a newer local write supersedes the echo (no data loss)', () => {
+    const client = clientMapWithOptimisticWrite();
+    // A second, newer local write changes the value before the first echo lands.
+    const newerValue = { title: 'hello', n: 2 };
+    const newerStamp = { millis: 3000, counter: 0, nodeId: 'client' };
+    client.merge(key, { value: newerValue, timestamp: newerStamp });
+
+    // The stale echo of the FIRST write arrives and is rejected by LWW...
+    expect(client.merge(key, { value, timestamp: serverStamp })).toBe(false);
+    // ...and must not be adopted, because its value differs from the live one.
+    expect(client.adoptServerEcho(key, { value, timestamp: serverStamp })).toBe(false);
+
+    // The newer local write is preserved intact.
+    expect(client.get(key)).toEqual(newerValue);
+    expect(client.getRecord(key)?.timestamp).toEqual(newerStamp);
+  });
+
+  test('returns false when there is no local record for the key', () => {
+    const client = new LWWMap<string, typeof value>(new HLC('client'));
+    expect(client.adoptServerEcho(key, { value, timestamp: serverStamp })).toBe(false);
+  });
+
+  test('reconciles a rejected tombstone echo (REMOVE) by adopting the server ts', () => {
+    const client = new LWWMap<string, typeof value>(new HLC('client'));
+    // Local optimistic delete at an ahead-of-server timestamp.
+    client.merge(key, { value: null, timestamp: clientStamp });
+    expect(client.merge(key, { value: null, timestamp: serverStamp })).toBe(false);
+    expect(client.adoptServerEcho(key, { value: null, timestamp: serverStamp })).toBe(true);
+    expect(client.getRecord(key)?.timestamp).toEqual(serverStamp);
+  });
+});
diff --git a/packages/react/src/__tests__/useSyncExternalStore.migration.test.tsx b/packages/react/src/__tests__/useSyncExternalStore.migration.test.tsx
new file mode 100644
index 00000000..bf1c5dcf
--- /dev/null
+++ b/packages/react/src/__tests__/useSyncExternalStore.migration.test.tsx
@@ -0,0 +1,224 @@
+/**
+ * Behavioral tests for the useSyncExternalStore migration (TODO-515, folds 516).
+ *
+ * These are the negative-control / invariant tests the audit (DEPTH_REACT F1/F6)
+ * called for:
+ *   - useQuery returns cached data synchronously on first render (no
+ *     {loading:true, data:[]} flash) when the handle already has a snapshot.
+ *   - StrictMode double-invoke leaves a net-zero subscribe/unsubscribe balance.
+ *   - A rerender that does not change identity does NOT re-subscribe.
+ *   - getSnapshot is referentially stable across renders without a mutation
+ *     (the property that prevents tearing / infinite render loops).
+ */
+import React, { StrictMode } from 'react';
+import { renderHook, act, render } from '@testing-library/react';
+import { TopGunProvider } from '../TopGunProvider';
+import { useQuery } from '../hooks/useQuery';
+import { useMap } from '../hooks/useMap';
+import { TopGunClient } from '@topgunbuild/client';
+
+// A minimal LWWMap-shaped mock (mirrors useMap.test.tsx) — importing the real
+// LWWMap from @topgunbuild/core pulls in msgpackr ESM, which the package's jest
+// transform does not handle. We only need subscribe + a mutation that notifies.
+function createMockLWWMap() {
+  const data = new Map<string, any>();
+  const listeners = new Set<() => void>();
+  return {
+    get: (k: string) => data.get(k),
+    set: (k: string, v: any) => {
+      data.set(k, v);
+      listeners.forEach((cb) => cb());
+    },
+    entries: () => Array.from(data.entries()),
+    subscribe: (cb: () => void) => {
+      listeners.add(cb);
+      return () => listeners.delete(cb);
+    },
+  };
+}
+
+// ---- A QueryHandle mock that exposes the new synchronous snapshot accessors.
+function createSnapshotQueryHandle(initial: any[]) {
+  const dataListeners = new Set<(results: any[], meta?: any) => void>();
+  let snapshot = initial;
+  let emitted = initial.length > 0; // already has cached data
+  let subscribeCount = 0;
+  let unsubscribeCount = 0;
+  return {
+    subscribe: jest.fn((cb: (results: any[], meta?: any) => void) => {
+      subscribeCount++;
+      dataListeners.add(cb);
+      return () => {
+        unsubscribeCount++;
+        dataListeners.delete(cb);
+      };
+    }),
+    onDelta: jest.fn(() => () => {}),
+    onPaginationChange: jest.fn(() => () => {}),
+    onSyncStateChange: jest.fn(() => () => {}),
+    loadMore: jest.fn().mockResolvedValue(undefined),
+    // Synchronous cached snapshot accessors (the migration target).
+    getSnapshot: () => snapshot,
+    getSnapshotMeta: () => ({ settled: emitted, hasEmitted: emitted }),
+    // Test helpers
+    _emit: (next: any[]) => {
+      snapshot = next;
+      emitted = true;
+      for (const cb of dataListeners) cb(next, { settled: true });
+    },
+    _counts: () => ({ subscribeCount, unsubscribeCount }),
+  };
+}
+
+describe('useSyncExternalStore migration — useQuery synchronous snapshot', () => {
+  it('returns cached results synchronously on first render with loading:false (no loading flash)', () => {
+    const handle = createSnapshotQueryHandle([{ _key: 'a', title: 'cached' }]);
+    const client = {
+      query: jest.fn().mockReturnValue(handle),
+    } as unknown as TopGunClient;
+
+    const wrapper = ({ children }: { children: React.ReactNode }) => (
+      <TopGunProvider client={client}>{children}</TopGunProvider>
+    );
+
+    // Capture every committed (loading, data) pair.
+    const commits: Array<{ loading: boolean; len: number }> = [];
+    renderHook(
+      () => {
+        const q = useQuery('todos');
+        commits.push({ loading: q.loading, len: q.data.length });
+        return q;
+      },
+      { wrapper },
+    );
+
+    // The FIRST commit must already carry the cached row with loading:false.
+    // Negative control: before the migration this first commit was
+    // {loading:true, len:0} (a flash), because subscribe did not deliver the
+    // first listener synchronously and the snapshot was read async.
+    expect(commits[0]).toEqual({ loading: false, len: 1 });
+    // And there must be NO intermediate {loading:true, len:0} commit at all.
+    expect(commits.some((c) => c.loading === true && c.len === 0)).toBe(false);
+  });
+
+  it('still falls back to loading:true when the handle has no cached data', () => {
+    const handle = createSnapshotQueryHandle([]); // empty → not emitted
+    const client = {
+      query: jest.fn().mockReturnValue(handle),
+    } as unknown as TopGunClient;
+    const wrapper = ({ children }: { children: React.ReactNode }) => (
+      <TopGunProvider client={client}>{children}</TopGunProvider>
+    );
+
+    const { result } = renderHook(() => useQuery('todos'), { wrapper });
+    expect(result.current.loading).toBe(true);
+    expect(result.current.data).toEqual([]);
+
+    act(() => handle._emit([{ _key: 'a', title: 'arrived' }]));
+    expect(result.current.loading).toBe(false);
+    expect(result.current.data).toEqual([{ _key: 'a', title: 'arrived' }]);
+  });
+});
+
+describe('useSyncExternalStore migration — subscribe-count invariants', () => {
+  it('does NOT re-subscribe on a rerender that keeps the query identity', () => {
+    const handle = createSnapshotQueryHandle([]);
+    const client = {
+      query: jest.fn().mockReturnValue(handle),
+    } as unknown as TopGunClient;
+    const wrapper = ({ children }: { children: React.ReactNode }) => (
+      <TopGunProvider client={client}>{children}</TopGunProvider>
+    );
+
+    const { rerender } = renderHook(() => useQuery('todos'), { wrapper });
+    const before = handle._counts().subscribeCount;
+
+    rerender();
+    rerender();
+    rerender();
+
+    expect(handle._counts().subscribeCount).toBe(before);
+  });
+
+  it('StrictMode double-invoke leaves a net-zero subscribe/unsubscribe balance (no leak)', () => {
+    const handle = createSnapshotQueryHandle([]);
+    const client = {
+      query: jest.fn().mockReturnValue(handle),
+    } as unknown as TopGunClient;
+
+    function Probe() {
+      useQuery('todos');
+      return null;
+    }
+
+    const { unmount } = render(
+      <StrictMode>
+        <TopGunProvider client={client}>
+          <Probe />
+        </TopGunProvider>
+      </StrictMode>,
+    );
+    unmount();
+
+    const { subscribeCount, unsubscribeCount } = handle._counts();
+    // Every subscribe must be matched by an unsubscribe after unmount.
+    expect(subscribeCount).toBe(unsubscribeCount);
+    expect(subscribeCount).toBeGreaterThan(0);
+  });
+});
+
+describe('useSyncExternalStore migration — getSnapshot referential stability', () => {
+  it('useQuery.data keeps the same array identity across rerenders without a mutation', () => {
+    const handle = createSnapshotQueryHandle([{ _key: 'a' }]);
+    const client = {
+      query: jest.fn().mockReturnValue(handle),
+    } as unknown as TopGunClient;
+    const wrapper = ({ children }: { children: React.ReactNode }) => (
+      <TopGunProvider client={client}>{children}</TopGunProvider>
+    );
+
+    const { result, rerender } = renderHook(() => useQuery('todos'), { wrapper });
+    const first = result.current.data;
+    rerender();
+    rerender();
+    // Same identity — this is the property that prevents useSyncExternalStore
+    // from looping / tearing. A getSnapshot that built a fresh [] each call
+    // would fail this and (under native USES) throw a max-depth loop.
+    expect(result.current.data).toBe(first);
+  });
+
+  it('useMap returns the same stable LWWMap object and re-renders only on mutation', () => {
+    const map = createMockLWWMap();
+    const client = {
+      getMap: jest.fn().mockReturnValue(map),
+      getRecordSyncStateTracker: jest.fn(),
+    } as unknown as TopGunClient;
+    const wrapper = ({ children }: { children: React.ReactNode }) => (
+      <TopGunProvider client={client}>{children}</TopGunProvider>
+    );
+
+    let renderCount = 0;
+    const { result, rerender } = renderHook(
+      () => {
+        renderCount++;
+        return useMap('cart');
+      },
+      { wrapper },
+    );
+
+    const mapRef = result.current;
+    expect(mapRef).toBe(map);
+
+    rerender();
+    expect(result.current).toBe(map); // identity stable across rerenders
+
+    const beforeMutation = renderCount;
+    act(() => {
+      map.set('item-1', { qty: 1 });
+    });
+    // A real mutation drives a re-render via the version-counter snapshot.
+    expect(renderCount).toBeGreaterThan(beforeMutation);
+    // The returned object is still the same mutable map.
+    expect(result.current).toBe(map);
+  });
+});
diff --git a/packages/react/src/__tests__/useTopic.test.tsx b/packages/react/src/__tests__/useTopic.test.tsx
index 6fee4913..a2eb95cb 100644
--- a/packages/react/src/__tests__/useTopic.test.tsx
+++ b/packages/react/src/__tests__/useTopic.test.tsx
@@ -154,4 +154,67 @@ describe('useTopic', () => {
     });
     expect(callback2).toHaveBeenCalledWith({ msg: 'second' }, { timestamp: 2 });
   });
+
+  // TODO-516 regression: previously the subscription effect depended on
+  // [topic, callback], so a fresh inline callback every render tore down and
+  // re-created the subscription on EVERY render — the callbackRef indirection
+  // was dead code. The migration keys the subscription only on topic identity
+  // (callback presence, not identity), so a changing inline callback must NOT
+  // re-subscribe, and the latest callback must still receive messages.
+  it('does NOT re-subscribe when only the inline callback identity changes (TODO-516)', () => {
+    // Render with a NEW inline callback every render — the common
+    // useTopic('chat', d => …) usage. We capture the latest closure so we can
+    // assert the most recent one still fires.
+    const received: any[] = [];
+    const { rerender } = renderHook(
+      // Intentionally a fresh inline callback per render to exercise the churn fix.
+      ({ tag }: { tag: number }) =>
+        useTopic('test-topic', (data: any) => {
+          received.push({ tag, data });
+        }),
+      { wrapper, initialProps: { tag: 0 } },
+    );
+
+    // Exactly one subscribe across the initial render(s).
+    expect(mockTopicHandle.subscribe).toHaveBeenCalledTimes(1);
+    expect(mockTopicHandle._getListenerCount()).toBe(1);
+
+    // Force several rerenders, each with a brand-new inline callback identity.
+    rerender({ tag: 1 });
+    rerender({ tag: 2 });
+    rerender({ tag: 3 });
+
+    // STILL exactly one subscribe — no churn.
+    expect(mockTopicHandle.subscribe).toHaveBeenCalledTimes(1);
+    expect(mockTopicHandle._getListenerCount()).toBe(1);
+
+    // The LATEST callback (tag 3) must receive the message.
+    act(() => {
+      mockTopicHandle._triggerMessage({ msg: 'hello' }, { timestamp: 9 });
+    });
+    expect(received).toHaveLength(1);
+    expect(received[0]).toEqual({ tag: 3, data: { msg: 'hello' } });
+  });
+
+  it('StrictMode double-invoke leaves no leaked topic listener', () => {
+    const callback = jest.fn();
+
+    const StrictWrapper = ({ children }: { children: React.ReactNode }) => (
+      <React.StrictMode>
+        <TopGunProvider client={mockClient}>{children}</TopGunProvider>
+      </React.StrictMode>
+    );
+
+    const { unmount } = renderHook(() => useTopic('test-topic', callback), {
+      wrapper: StrictWrapper,
+    });
+
+    // After mount (even with StrictMode's mount→unmount→mount), exactly one
+    // live listener remains.
+    expect(mockTopicHandle._getListenerCount()).toBe(1);
+
+    unmount();
+    // Net-zero: no leaked listener.
+    expect(mockTopicHandle._getListenerCount()).toBe(0);
+  });
 });
diff --git a/packages/react/src/hooks/internal/useExternalStore.ts b/packages/react/src/hooks/internal/useExternalStore.ts
new file mode 100644
index 00000000..a0c2af12
--- /dev/null
+++ b/packages/react/src/hooks/internal/useExternalStore.ts
@@ -0,0 +1,137 @@
+import * as React from 'react';
+import { useCallback, useEffect, useRef, useState } from 'react';
+
+/**
+ * `useSyncExternalStore` accessor that degrades gracefully on React < 18.
+ *
+ * React 18 added `useSyncExternalStore` as a named export. The package's
+ * peer range is `react >= 16.8`, so we cannot assume the export exists.
+ * When it does, we use it directly — it returns the cached snapshot
+ * synchronously during render (no first-paint loading flash) and re-checks
+ * the snapshot immediately after subscribing (closing the
+ * subscribe-in-passive-effect gap), and is tear-free under concurrent
+ * rendering. When it does not (React 16.8 / 17), we fall back to a
+ * `useState` + `useEffect(subscribe)` shim that preserves the same call
+ * signature; the synchronous-snapshot and tearing guarantees simply reduce
+ * to the legacy behaviour on those older runtimes.
+ *
+ * The fallback intentionally mirrors React's own
+ * `use-sync-external-store/shim` semantics closely enough for our hooks: it
+ * seeds state from `getSnapshot()` (so the first render still reads the
+ * cached value), re-reads after subscribing, and re-reads on every store
+ * notification.
+ */
+
+type Subscribe = (onStoreChange: () => void) => () => void;
+
+// React 18 exposes useSyncExternalStore as a named export; older versions do not.
+const nativeUSES = (React as { useSyncExternalStore?: <T>(s: Subscribe, g: () => T) => T })
+  .useSyncExternalStore;
+
+function fallbackUseSyncExternalStore<T>(subscribe: Subscribe, getSnapshot: () => T): T {
+  // Seed from the snapshot so the first render reflects the cached value,
+  // matching useSyncExternalStore's synchronous-read contract as closely as
+  // a passive-effect shim can.
+  const [snapshot, setSnapshot] = useState<T>(getSnapshot);
+
+  // Keep the latest getSnapshot in a ref so the subscribe effect does not
+  // need it as a dependency (subscribe identity drives re-subscription).
+  const getSnapshotRef = useRef(getSnapshot);
+  getSnapshotRef.current = getSnapshot;
+
+  useEffect(() => {
+    let mounted = true;
+    const checkForUpdates = () => {
+      if (!mounted) return;
+      const next = getSnapshotRef.current();
+      // Functional update so we compare against the freshest committed value
+      // and avoid scheduling a render when the snapshot is referentially equal.
+      setSnapshot((prev) => (Object.is(prev, next) ? prev : next));
+    };
+    const unsubscribe = subscribe(checkForUpdates);
+    // Re-check immediately after subscribing to catch a mutation that landed
+    // between the initial getSnapshot() seed and the subscription taking effect.
+    checkForUpdates();
+    return () => {
+      mounted = false;
+      unsubscribe();
+    };
+  }, [subscribe]);
+
+  return snapshot;
+}
+
+/**
+ * Subscribe to an external store and read a synchronous, referentially-stable
+ * snapshot. `subscribe` MUST be a stable function (memoize with `useCallback`
+ * keyed by the store identity) and `getSnapshot` MUST return a cached value
+ * that only changes identity when the store actually changes — otherwise React
+ * re-subscribes / re-renders on every pass.
+ */
+export function useExternalStore<T>(subscribe: Subscribe, getSnapshot: () => T): T {
+  const impl = nativeUSES ?? fallbackUseSyncExternalStore;
+  return impl(subscribe, getSnapshot);
+}
+
+/**
+ * A hook-local external store for hooks whose value is produced imperatively
+ * (e.g. handle lifecycle managed across effects, debounce timers) rather than
+ * read directly from a long-lived client store. The hook keeps the current
+ * snapshot in `ref.current` and calls `notify()` after mutating it; React reads
+ * the snapshot through `useSyncExternalStore`, giving tearing-safety and unmount
+ * safety without an `isMounted` ref.
+ *
+ * Returns `[value, notify]` where `value` is the React-tracked snapshot and
+ * `notify` schedules a re-render after the caller mutates `ref.current`.
+ * `getSnapshot` MUST return a referentially-stable value between notifies.
+ */
+export function useLocalStore<T>(getSnapshot: () => T): [T, () => void] {
+  const listenersRef = useRef(new Set<() => void>());
+
+  const subscribe = useCallback<Subscribe>((onStoreChange) => {
+    const listeners = listenersRef.current;
+    listeners.add(onStoreChange);
+    return () => {
+      listeners.delete(onStoreChange);
+    };
+  }, []);
+
+  const notify = useCallback(() => {
+    for (const l of listenersRef.current) l();
+  }, []);
+
+  const value = useExternalStore(subscribe, getSnapshot);
+  return [value, notify];
+}
+
+/**
+ * Subscribe to a store that notifies a bare `() => void` callback (the common
+ * shape of TopGun's CRDT maps, counters, and trackers) and drive re-renders via
+ * a monotonically increasing **version counter**. The returned number is a
+ * stable primitive that only changes when the store notifies, so it satisfies
+ * the `getSnapshot` referential-stability contract while the caller continues
+ * to return the live store object from the hook.
+ *
+ * `subscribeToStore(onChange)` must register `onChange` with the store and
+ * return an unsubscribe function. It must be stable (memoize by store identity).
+ */
+export function useStoreVersion(subscribeToStore: Subscribe): number {
+  // A mutable cell holding the current version. Bumped inside the store's
+  // notify callback (before React's onStoreChange) so getSnapshot reads the
+  // new value synchronously. Never recreated, so its identity is stable.
+  const versionRef = useRef(0);
+
+  const subscribe = useCallback<Subscribe>(
+    (onStoreChange) => {
+      return subscribeToStore(() => {
+        versionRef.current += 1;
+        onStoreChange();
+      });
+    },
+    [subscribeToStore],
+  );
+
+  const getSnapshot = useCallback(() => versionRef.current, []);
+
+  return useExternalStore(subscribe, getSnapshot);
+}
diff --git a/packages/react/src/hooks/internal/useTrackerMapSnapshot.ts b/packages/react/src/hooks/internal/useTrackerMapSnapshot.ts
new file mode 100644
index 00000000..ed0cfe54
--- /dev/null
+++ b/packages/react/src/hooks/internal/useTrackerMapSnapshot.ts
@@ -0,0 +1,32 @@
+import { useCallback } from 'react';
+import type { RecordSyncState } from '@topgunbuild/client';
+import { useClient } from '../useClient';
+import { useExternalStore } from './useExternalStore';
+
+/**
+ * Read the per-map `RecordSyncState` snapshot from the client's
+ * `RecordSyncStateTracker` as a React 18 external store.
+ *
+ * The tracker already exposes the two pieces `useSyncExternalStore` needs:
+ * `getMapSnapshot(mapName)` returns a cached, referentially-stable
+ * `ReadonlyMap` (identity changes only when a key's projection changes), and
+ * `onChange(mapName, cb)` notifies on those changes. Subscribing through
+ * `useExternalStore` gives a synchronous first-render snapshot (no empty-map
+ * flash) and tearing-freedom, replacing the prior `useState` + `useEffect`
+ * seed-then-subscribe dance.
+ */
+export function useTrackerMapSnapshot(mapName: string): ReadonlyMap<string, RecordSyncState> {
+  const client = useClient();
+
+  const subscribe = useCallback(
+    (onChange: () => void) => client.getRecordSyncStateTracker().onChange(mapName, onChange),
+    [client, mapName],
+  );
+
+  const getSnapshot = useCallback(
+    () => client.getRecordSyncStateTracker().getMapSnapshot(mapName),
+    [client, mapName],
+  );
+
+  return useExternalStore(subscribe, getSnapshot);
+}
diff --git a/packages/react/src/hooks/useEventJournal.ts b/packages/react/src/hooks/useEventJournal.ts
index e6199d90..0b94a04f 100644
--- a/packages/react/src/hooks/useEventJournal.ts
+++ b/packages/react/src/hooks/useEventJournal.ts
@@ -2,6 +2,9 @@ import { useState, useEffect, useRef, useCallback, useMemo } from 'react';
 import type { JournalEvent, JournalEventType } from '@topgunbuild/core';
 import type { EventJournalReader, JournalSubscribeOptions } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useLocalStore } from './internal/useExternalStore';
+
+const EMPTY_EVENTS: JournalEvent[] = [];
 
 /**
  * Options for useEventJournal hook.
@@ -100,11 +103,16 @@ export interface UseEventJournalResult {
  */
 export function useEventJournal(options: UseEventJournalOptions = {}): UseEventJournalResult {
   const client = useClient();
-  const [events, setEvents] = useState<JournalEvent[]>([]);
+
+  // Events accumulate in a ref read through useSyncExternalStore (no isMounted
+  // ref). The ref holds a referentially-stable array between notifies.
+  const eventsRef = useRef<JournalEvent[]>(EMPTY_EVENTS);
+  const getEventsSnapshot = useCallback(() => eventsRef.current, []);
+  const [events, notifyEvents] = useLocalStore(getEventsSnapshot);
+
   const [lastEvent, setLastEvent] = useState<JournalEvent | null>(null);
   const [isSubscribed, setIsSubscribed] = useState(false);
 
-  const isMounted = useRef(true);
   const journalRef = useRef<EventJournalReader | null>(null);
 
   const maxEvents = options.maxEvents ?? 100;
@@ -115,9 +123,10 @@ export function useEventJournal(options: UseEventJournalOptions = {}): UseEventJ
 
   // Clear events callback
   const clearEvents = useCallback(() => {
-    setEvents([]);
+    eventsRef.current = EMPTY_EVENTS;
+    notifyEvents();
     setLastEvent(null);
-  }, []);
+  }, [notifyEvents]);
 
   // Read historical events
   const readFrom = useCallback(
@@ -151,8 +160,6 @@ export function useEventJournal(options: UseEventJournalOptions = {}): UseEventJ
   );
 
   useEffect(() => {
-    isMounted.current = true;
-
     // Don't subscribe if paused
     if (options.paused) {
       setIsSubscribed(false);
@@ -169,16 +176,12 @@ export function useEventJournal(options: UseEventJournalOptions = {}): UseEventJ
     };
 
     const unsubscribe = journal.subscribe((event) => {
-      if (!isMounted.current) return;
-
-      // Add to events with rotation
-      setEvents((prev) => {
-        const newEvents = [...prev, event];
-        if (newEvents.length > maxEvents) {
-          return newEvents.slice(-maxEvents);
-        }
-        return newEvents;
-      });
+      // Append with rotation into the ref, then notify React. No isMounted
+      // guard needed — the effect cleanup unsubscribes before unmount.
+      const prev = eventsRef.current;
+      const newEvents = [...prev, event];
+      eventsRef.current = newEvents.length > maxEvents ? newEvents.slice(-maxEvents) : newEvents;
+      notifyEvents();
 
       setLastEvent(event);
 
@@ -189,11 +192,12 @@ export function useEventJournal(options: UseEventJournalOptions = {}): UseEventJ
     setIsSubscribed(true);
 
     return () => {
-      isMounted.current = false;
       setIsSubscribed(false);
       unsubscribe();
     };
-  }, [client, filterKey, maxEvents]);
+    // filterKey is the serialized form of the filter options (mapName/types/
+    // fromSequence/paused); the raw option fields are intentionally not listed.
+  }, [client, filterKey, maxEvents, notifyEvents]);
 
   return useMemo(
     () => ({
diff --git a/packages/react/src/hooks/useHybridQuery.ts b/packages/react/src/hooks/useHybridQuery.ts
index c38ad770..dfc59cad 100644
--- a/packages/react/src/hooks/useHybridQuery.ts
+++ b/packages/react/src/hooks/useHybridQuery.ts
@@ -1,4 +1,4 @@
-import { useState, useEffect, useRef, useMemo } from 'react';
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
 import type {
   HybridQueryHandle,
   HybridResultItem,
@@ -7,6 +7,9 @@ import type {
   PaginationInfo,
 } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useExternalStore } from './internal/useExternalStore';
+
+const EMPTY_HYBRID_RESULTS: HybridResultItem<unknown>[] = [];
 
 /**
  * Extended options for useHybridQuery hook.
@@ -129,120 +132,108 @@ export function useHybridQuery<T = unknown>(
   options?: UseHybridQueryOptions,
 ): UseHybridQueryResult<T> {
   const client = useClient();
-  const [results, setResults] = useState<HybridResultItem<T>[]>([]);
-  const [loading, setLoading] = useState(true);
-  const [error, setError] = useState<Error | null>(null);
-
-  const [paginationInfo, setPaginationInfo] = useState<PaginationInfo>({
-    hasMore: false,
-    cursorStatus: 'none',
-  });
-
-  // Track if component is mounted
-  const isMounted = useRef(true);
-
-  // Store handle ref
-  const handleRef = useRef<HybridQueryHandle<T> | null>(null);
-
-  // Store unsubscribe function
-  const unsubscribeRef = useRef<(() => void) | null>(null);
-
-  // Memoize filter to avoid unnecessary re-renders
-  const memoizedFilter = useMemo(
-    () => filter,
-    [
-      JSON.stringify(filter.predicate),
-      JSON.stringify(filter.where),
-      JSON.stringify(filter.sort),
-      filter.limit,
-      filter.cursor,
-    ],
-  );
 
   // Skip option
   const skip = options?.skip ?? false;
 
-  // Effect for creating/disposing handle
-  useEffect(() => {
-    isMounted.current = true;
-
-    // Don't subscribe if skip is true
-    if (skip) {
-      setResults([]);
-      setLoading(false);
-      setError(null);
-      return;
-    }
-
-    setLoading(true);
-    setError(null);
+  // Stable filter dependency key.
+  const filterKey = JSON.stringify({
+    predicate: filter.predicate,
+    where: filter.where,
+    sort: filter.sort,
+    limit: filter.limit,
+    cursor: filter.cursor,
+  });
 
-    try {
-      // Cleanup old handle if exists
-      if (handleRef.current) {
-        if (unsubscribeRef.current) {
-          unsubscribeRef.current();
-          unsubscribeRef.current = null;
-        }
+  // Construct the handle in render, memoized by [client, mapName, filterKey,
+  // skip]. client.hybridQuery() is a pure constructor — subscription activates
+  // on handle.subscribe() inside useExternalStore. When skipped, no handle is
+  // created. Construction throws are captured and surfaced as `error`.
+  const { handle, constructError } = useMemo<{
+    handle: HybridQueryHandle<T> | null;
+    constructError: Error | null;
+  }>(
+    () => {
+      if (skip) return { handle: null, constructError: null };
+      try {
+        return { handle: client.hybridQuery<T>(mapName, filter), constructError: null };
+      } catch (err) {
+        return {
+          handle: null,
+          constructError: err instanceof Error ? err : new Error(String(err)),
+        };
       }
+    },
+    // filterKey is the serialized form of `filter`; depending on the `filter`
+    // object identity would re-create the handle every render.
+    [client, mapName, filterKey, skip],
+  );
 
-      // Create new handle
-      const handle = client.hybridQuery<T>(mapName, memoizedFilter);
-      handleRef.current = handle;
-
-      // Flag to track if we've received initial results
-      let hasReceivedResults = false;
-
-      // Subscribe to result updates
-      unsubscribeRef.current = handle.subscribe((newResults) => {
-        if (isMounted.current) {
-          setResults(newResults);
-          if (!hasReceivedResults) {
-            hasReceivedResults = true;
-            setLoading(false);
-          }
-        }
+  // ---- results via useSyncExternalStore -----------------------------------
+  const cacheRef = useRef<{ value: HybridResultItem<T>[]; emitted: boolean }>({
+    value: EMPTY_HYBRID_RESULTS as HybridResultItem<T>[],
+    emitted: false,
+  });
+  const lastHandleRef = useRef<HybridQueryHandle<T> | null>(null);
+  if (lastHandleRef.current !== handle) {
+    lastHandleRef.current = handle;
+    cacheRef.current = { value: EMPTY_HYBRID_RESULTS as HybridResultItem<T>[], emitted: false };
+  }
+
+  const subscribeResults = useCallback(
+    (onChange: () => void) => {
+      if (!handle) return () => {};
+      return handle.subscribe((newResults) => {
+        cacheRef.current = { value: newResults, emitted: true };
+        onChange();
       });
+    },
+    [handle],
+  );
 
-      const unsubscribePagination = handle.onPaginationChange((info) => {
-        if (isMounted.current) {
-          setPaginationInfo(info);
-        }
-      });
+  const getResults = useCallback((): HybridResultItem<T>[] => {
+    if (handle && typeof handle.getSnapshot === 'function') {
+      return handle.getSnapshot() as HybridResultItem<T>[];
+    }
+    return cacheRef.current.value;
+  }, [handle]);
 
-      // Store combined unsubscribe function
-      const originalUnsubscribe = unsubscribeRef.current;
-      unsubscribeRef.current = () => {
-        originalUnsubscribe?.();
-        unsubscribePagination();
-      };
-    } catch (err) {
-      if (isMounted.current) {
-        setError(err instanceof Error ? err : new Error(String(err)));
-        setLoading(false);
-      }
+  const results = useExternalStore(subscribeResults, getResults);
+
+  const loading = useMemo(() => {
+    if (skip || constructError) return false;
+    if (handle && typeof handle.getSnapshotMeta === 'function') {
+      return !handle.getSnapshotMeta().hasEmitted;
     }
+    return !cacheRef.current.emitted;
+    // `results` is included so loading re-derives after each emission.
+  }, [handle, skip, constructError, results]);
 
-    // Cleanup on unmount or dependency change
-    return () => {
-      isMounted.current = false;
-      if (unsubscribeRef.current) {
-        unsubscribeRef.current();
-        unsubscribeRef.current = null;
-      }
-      handleRef.current = null;
-    };
-  }, [client, mapName, memoizedFilter, skip]);
+  // ---- pagination (effect-driven) -----------------------------------------
+  const [paginationInfo, setPaginationInfo] = useState<PaginationInfo>({
+    hasMore: false,
+    cursorStatus: 'none',
+  });
+
+  useEffect(() => {
+    if (!handle) {
+      setPaginationInfo({ hasMore: false, cursorStatus: 'none' });
+      return;
+    }
+    return handle.onPaginationChange((info) => {
+      setPaginationInfo(info);
+    });
+  }, [handle]);
 
   return useMemo(
     () => ({
       results,
       loading,
-      error,
+      error: constructError,
       nextCursor: paginationInfo.nextCursor,
       hasMore: paginationInfo.hasMore,
       cursorStatus: paginationInfo.cursorStatus,
     }),
-    [results, loading, error, paginationInfo],
+    [results, loading, constructError, paginationInfo],
   );
 }
diff --git a/packages/react/src/hooks/useHybridSearchSubscribe.ts b/packages/react/src/hooks/useHybridSearchSubscribe.ts
index d87704f7..412ed322 100644
--- a/packages/react/src/hooks/useHybridSearchSubscribe.ts
+++ b/packages/react/src/hooks/useHybridSearchSubscribe.ts
@@ -1,10 +1,13 @@
-import { useState, useEffect, useRef, useMemo } from 'react';
+import { useState, useEffect, useRef, useMemo, useCallback } from 'react';
 import type {
   HybridSearchHandle,
   HybridSearchHandleResult,
   HybridSearchSubscribeOptions,
 } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useLocalStore } from './internal/useExternalStore';
+
+const EMPTY_HYBRID_SEARCH_RESULTS: HybridSearchHandleResult<unknown>[] = [];
 
 /**
  * Options for the useHybridSearchSubscribe hook.
@@ -82,13 +85,17 @@ export function useHybridSearchSubscribe<T = unknown>(
   options?: UseHybridSearchSubscribeOptions,
 ): UseHybridSearchSubscribeResult<T> {
   const client = useClient();
-  const [results, setResults] = useState<HybridSearchHandleResult<T>[]>([]);
+
+  // Results live in a ref read through useSyncExternalStore (no isMounted ref).
+  const resultsRef = useRef<HybridSearchHandleResult<T>[]>(
+    EMPTY_HYBRID_SEARCH_RESULTS as HybridSearchHandleResult<T>[],
+  );
+  const getResultsSnapshot = useCallback(() => resultsRef.current, []);
+  const [results, notifyResults] = useLocalStore(getResultsSnapshot);
+
   const [loading, setLoading] = useState(true);
   const [error, setError] = useState<Error | null>(null);
 
-  // Track mount state to suppress state updates after unmount
-  const isMounted = useRef(true);
-
   // Store HybridSearchHandle for reuse across query/options changes
   const handleRef = useRef<HybridSearchHandle<T> | null>(null);
 
@@ -140,11 +147,11 @@ export function useHybridSearchSubscribe<T = unknown>(
         clearTimeout(debounceTimeoutRef.current);
       }
       debounceTimeoutRef.current = setTimeout(() => {
-        if (isMounted.current) {
-          setDebouncedQueryText(queryText);
-        }
+        setDebouncedQueryText(queryText);
       }, debounceMs);
       return () => {
+        // Clearing the pending timer on unmount/dep-change prevents a setState
+        // after unmount — no isMounted ref needed.
         if (debounceTimeoutRef.current) {
           clearTimeout(debounceTimeoutRef.current);
         }
@@ -155,12 +162,10 @@ export function useHybridSearchSubscribe<T = unknown>(
     }
   }, [queryText, debounceMs]);
 
-  // Effect 1 — handle lifecycle: dispose and reset on mapName/client change or unmount
+  // Effect 1 — handle lifecycle: dispose and reset on mapName/client change or
+  // unmount. useSyncExternalStore handles unmount safety for the results read.
   useEffect(() => {
-    isMounted.current = true;
-
     return () => {
-      isMounted.current = false;
       if (unsubscribeRef.current) {
         unsubscribeRef.current();
         unsubscribeRef.current = null;
@@ -186,7 +191,8 @@ export function useHybridSearchSubscribe<T = unknown>(
         handleRef.current.dispose();
         handleRef.current = null;
       }
-      setResults([]);
+      resultsRef.current = EMPTY_HYBRID_SEARCH_RESULTS as HybridSearchHandleResult<T>[];
+      notifyResults();
       setLoading(false);
       setError(null);
       return;
@@ -208,22 +214,20 @@ export function useHybridSearchSubscribe<T = unknown>(
         let hasReceivedFirstData = false;
 
         unsubscribeRef.current = handle.subscribe((newResults) => {
-          if (isMounted.current) {
-            setResults(newResults);
-            if (!hasReceivedFirstData) {
-              hasReceivedFirstData = true;
-              setLoading(false);
-            }
+          resultsRef.current =
+            typeof handle.getSnapshot === 'function' ? handle.getSnapshot() : newResults;
+          notifyResults();
+          if (!hasReceivedFirstData) {
+            hasReceivedFirstData = true;
+            setLoading(false);
           }
         });
       }
     } catch (err) {
-      if (isMounted.current) {
-        setError(err instanceof Error ? err : new Error(String(err)));
-        setLoading(false);
-      }
+      setError(err instanceof Error ? err : new Error(String(err)));
+      setLoading(false);
     }
-  }, [client, mapName, debouncedQueryText, searchOptions, enabled]);
+  }, [client, mapName, debouncedQueryText, searchOptions, enabled, notifyResults]);
 
   return useMemo(() => ({ results, loading, error }), [results, loading, error]);
 }
diff --git a/packages/react/src/hooks/useMap.ts b/packages/react/src/hooks/useMap.ts
index 421845de..dc251e19 100644
--- a/packages/react/src/hooks/useMap.ts
+++ b/packages/react/src/hooks/useMap.ts
@@ -1,39 +1,26 @@
-import { useState, useEffect, useRef } from 'react';
+import { useCallback, useMemo } from 'react';
 import { LWWMap } from '@topgunbuild/core';
 import type { RecordSyncState } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useStoreVersion } from './internal/useExternalStore';
+import { useTrackerMapSnapshot } from './internal/useTrackerMapSnapshot';
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any -- V defaults to any so callers without a schema type still get a usable map; narrowed via generic at call site
 export function useMap<K = string, V = any>(mapName: string): LWWMap<K, V> {
   const client = useClient();
-  // Get the map instance. This is stable for the same mapName.
-  const map = client.getMap<K, V>(mapName);
+  // Memoize the lookup by [client, name] so we do not create/register a fresh
+  // handle during every render (render-purity: the first getMap call news up
+  // the CRDT + kicks an async restore).
+  const map = useMemo(() => client.getMap<K, V>(mapName), [client, mapName]);
 
-  // We use a dummy state to trigger re-renders when the map changes
-  const [, setTick] = useState(0);
-  const isMounted = useRef(true);
-
-  useEffect(() => {
-    isMounted.current = true;
-
-    // Subscribe to map changes
-    const unsubscribe = map.subscribe(() => {
-      if (isMounted.current) {
-        setTick((t) => t + 1);
-      }
-    });
-
-    return () => {
-      isMounted.current = false;
-      unsubscribe();
-    };
-  }, [map]);
+  // Stable subscribe keyed by the map identity. Drives re-renders via a version
+  // counter snapshot — the returned LWWMap stays the same mutable object.
+  const subscribe = useCallback((onChange: () => void) => map.subscribe(onChange), [map]);
+  useStoreVersion(subscribe);
 
   return map;
 }
 
-const EMPTY_SYNC_STATE: ReadonlyMap<string, RecordSyncState> = new Map();
-
 /**
  * Companion to `useMap` — returns the underlying LWWMap alongside a
  * `syncState` snapshot tracking each key's per-record sync state. The
@@ -55,38 +42,12 @@ export function useMapWithSyncState<K = string, V = any>(
   mapName: string,
 ): { map: LWWMap<K, V>; syncState: ReadonlyMap<string, RecordSyncState> } {
   const client = useClient();
-  const map = client.getMap<K, V>(mapName);
-
-  const [, setTick] = useState(0);
-  const [syncState, setSyncState] =
-    useState<ReadonlyMap<string, RecordSyncState>>(EMPTY_SYNC_STATE);
-  const isMounted = useRef(true);
-
-  useEffect(() => {
-    isMounted.current = true;
-
-    const unsubscribeMap = map.subscribe(() => {
-      if (isMounted.current) {
-        setTick((t) => t + 1);
-      }
-    });
+  const map = useMemo(() => client.getMap<K, V>(mapName), [client, mapName]);
 
-    const tracker = client.getRecordSyncStateTracker();
-    // Seed with current snapshot so the first render reflects any
-    // pre-existing tracker state for this map name.
-    setSyncState(tracker.getMapSnapshot(mapName));
-    const unsubscribeSync = tracker.onChange(mapName, (snapshot) => {
-      if (isMounted.current) {
-        setSyncState(snapshot);
-      }
-    });
+  const subscribe = useCallback((onChange: () => void) => map.subscribe(onChange), [map]);
+  useStoreVersion(subscribe);
 
-    return () => {
-      isMounted.current = false;
-      unsubscribeMap();
-      unsubscribeSync();
-    };
-  }, [client, map, mapName]);
+  const syncState = useTrackerMapSnapshot(mapName);
 
   return { map, syncState };
 }
diff --git a/packages/react/src/hooks/useMergeRejections.ts b/packages/react/src/hooks/useMergeRejections.ts
index d6c96a2f..26cf3e98 100644
--- a/packages/react/src/hooks/useMergeRejections.ts
+++ b/packages/react/src/hooks/useMergeRejections.ts
@@ -1,7 +1,10 @@
-import { useState, useEffect, useCallback } from 'react';
+import { useState, useEffect, useCallback, useRef } from 'react';
 import { useClient } from './useClient';
+import { useLocalStore } from './internal/useExternalStore';
 import type { MergeRejection } from '@topgunbuild/core';
 
+const EMPTY_REJECTIONS: MergeRejection[] = [];
+
 /**
  * Options for useMergeRejections hook.
  */
@@ -80,7 +83,12 @@ export function useMergeRejections(
   const client = useClient();
   const { mapName, maxHistory = 100 } = options;
 
-  const [rejections, setRejections] = useState<MergeRejection[]>([]);
+  // Rejections accumulate in a ref read through useSyncExternalStore (tearing-/
+  // unmount-safe). The ref holds a referentially-stable array between notifies.
+  const rejectionsRef = useRef<MergeRejection[]>(EMPTY_REJECTIONS);
+  const getRejectionsSnapshot = useCallback(() => rejectionsRef.current, []);
+  const [rejections, notifyRejections] = useLocalStore(getRejectionsSnapshot);
+
   const [lastRejection, setLastRejection] = useState<MergeRejection | null>(null);
 
   useEffect(() => {
@@ -93,23 +101,19 @@ export function useMergeRejections(
       }
 
       setLastRejection(rejection);
-      setRejections((prev) => {
-        const next = [...prev, rejection];
-        // Limit history size
-        if (next.length > maxHistory) {
-          return next.slice(-maxHistory);
-        }
-        return next;
-      });
+      const next = [...rejectionsRef.current, rejection];
+      rejectionsRef.current = next.length > maxHistory ? next.slice(-maxHistory) : next;
+      notifyRejections();
     });
 
     return unsubscribe;
-  }, [client, mapName, maxHistory]);
+  }, [client, mapName, maxHistory, notifyRejections]);
 
   const clear = useCallback(() => {
-    setRejections([]);
+    rejectionsRef.current = EMPTY_REJECTIONS;
+    notifyRejections();
     setLastRejection(null);
-  }, []);
+  }, [notifyRejections]);
 
   return {
     rejections,
diff --git a/packages/react/src/hooks/useORMap.ts b/packages/react/src/hooks/useORMap.ts
index cd290921..75a43ab4 100644
--- a/packages/react/src/hooks/useORMap.ts
+++ b/packages/react/src/hooks/useORMap.ts
@@ -1,36 +1,25 @@
-import { useState, useEffect, useRef } from 'react';
+import { useCallback, useMemo } from 'react';
 import { ORMap } from '@topgunbuild/core';
 import type { RecordSyncState } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useStoreVersion } from './internal/useExternalStore';
+import { useTrackerMapSnapshot } from './internal/useTrackerMapSnapshot';
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any -- V defaults to any so callers without a schema type still get a usable ORMap; narrowed via generic at call site
 export function useORMap<K = string, V = any>(mapName: string): ORMap<K, V> {
   const client = useClient();
-  const map = client.getORMap<K, V>(mapName);
+  // Memoize the lookup by [client, name] so we do not create/register a fresh
+  // handle during every render (render-purity).
+  const map = useMemo(() => client.getORMap<K, V>(mapName), [client, mapName]);
 
-  const [, setTick] = useState(0);
-  const isMounted = useRef(true);
-
-  useEffect(() => {
-    isMounted.current = true;
-
-    const unsubscribe = map.subscribe(() => {
-      if (isMounted.current) {
-        setTick((t) => t + 1);
-      }
-    });
-
-    return () => {
-      isMounted.current = false;
-      unsubscribe();
-    };
-  }, [map]);
+  // Stable subscribe keyed by the map identity; re-renders via version counter.
+  // The returned ORMap stays the same mutable object.
+  const subscribe = useCallback((onChange: () => void) => map.subscribe(onChange), [map]);
+  useStoreVersion(subscribe);
 
   return map;
 }
 
-const EMPTY_OR_SYNC_STATE: ReadonlyMap<string, RecordSyncState> = new Map();
-
 /**
  * Companion to `useORMap` — returns the underlying ORMap alongside a
  * `syncState` snapshot tracking each key's per-record sync state. The
@@ -42,36 +31,12 @@ export function useORMapWithSyncState<K = string, V = any>(
   mapName: string,
 ): { map: ORMap<K, V>; syncState: ReadonlyMap<string, RecordSyncState> } {
   const client = useClient();
-  const map = client.getORMap<K, V>(mapName);
-
-  const [, setTick] = useState(0);
-  const [syncState, setSyncState] =
-    useState<ReadonlyMap<string, RecordSyncState>>(EMPTY_OR_SYNC_STATE);
-  const isMounted = useRef(true);
-
-  useEffect(() => {
-    isMounted.current = true;
-
-    const unsubscribeMap = map.subscribe(() => {
-      if (isMounted.current) {
-        setTick((t) => t + 1);
-      }
-    });
+  const map = useMemo(() => client.getORMap<K, V>(mapName), [client, mapName]);
 
-    const tracker = client.getRecordSyncStateTracker();
-    setSyncState(tracker.getMapSnapshot(mapName));
-    const unsubscribeSync = tracker.onChange(mapName, (snapshot) => {
-      if (isMounted.current) {
-        setSyncState(snapshot);
-      }
-    });
+  const subscribe = useCallback((onChange: () => void) => map.subscribe(onChange), [map]);
+  useStoreVersion(subscribe);
 
-    return () => {
-      isMounted.current = false;
-      unsubscribeMap();
-      unsubscribeSync();
-    };
-  }, [client, map, mapName]);
+  const syncState = useTrackerMapSnapshot(mapName);
 
   return { map, syncState };
 }
diff --git a/packages/react/src/hooks/usePNCounter.ts b/packages/react/src/hooks/usePNCounter.ts
index ec16ed13..7431ad89 100644
--- a/packages/react/src/hooks/usePNCounter.ts
+++ b/packages/react/src/hooks/usePNCounter.ts
@@ -1,5 +1,6 @@
-import { useState, useEffect, useCallback, useMemo } from 'react';
+import { useCallback, useEffect, useMemo, useState } from 'react';
 import { useClient } from './useClient';
+import { useExternalStore } from './internal/useExternalStore';
 
 /**
  * Result type for usePNCounter hook.
@@ -78,24 +79,33 @@ export interface UsePNCounterResult {
  */
 export function usePNCounter(name: string): UsePNCounterResult {
   const client = useClient();
-  const [value, setValue] = useState(0);
-  const [loading, setLoading] = useState(true);
 
-  // Get or create counter handle - memoized by name
-  const counter = useMemo(() => {
-    return client.getPNCounter(name);
-  }, [client, name]);
+  // Get or create counter handle - memoized by name (render purity).
+  const counter = useMemo(() => client.getPNCounter(name), [client, name]);
+
+  // Value is read from the handle's external store. `get()` returns a number
+  // primitive, so the snapshot is referentially stable by value.
+  const subscribe = useCallback(
+    (onChange: () => void) => counter.subscribe(() => onChange()),
+    [counter],
+  );
+  const getSnapshot = useCallback(() => counter.get(), [counter]);
+  const value = useExternalStore(subscribe, getSnapshot);
 
+  // `loading` is "true until the first value is observed". Track it with a tiny
+  // effect that flips false once the counter notifies (mirrors the previous
+  // behaviour without an isMounted ref — effect cleanup handles unmount).
+  const [loading, setLoading] = useState(true);
   useEffect(() => {
-    // Reset state when counter changes
     setLoading(true);
-
-    const unsubscribe = counter.subscribe((newValue) => {
-      setValue(newValue);
-      setLoading(false);
+    let active = true;
+    const unsubscribe = counter.subscribe(() => {
+      if (active) setLoading(false);
     });
-
-    return unsubscribe;
+    return () => {
+      active = false;
+      unsubscribe();
+    };
   }, [counter]);
 
   const increment = useCallback(() => {
diff --git a/packages/react/src/hooks/useQuery.ts b/packages/react/src/hooks/useQuery.ts
index c2982133..4dc1e460 100644
--- a/packages/react/src/hooks/useQuery.ts
+++ b/packages/react/src/hooks/useQuery.ts
@@ -1,4 +1,4 @@
-import { useState, useEffect, useRef, useCallback, useMemo } from 'react';
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
 import {
   QueryFilter,
   QueryResultItem,
@@ -9,8 +9,10 @@ import {
 } from '@topgunbuild/client';
 import type { RecordSyncState } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useExternalStore } from './internal/useExternalStore';
 
 const EMPTY_SYNC_STATE: ReadonlyMap<string, RecordSyncState> = new Map();
+const EMPTY_DATA: QueryResultItem<unknown>[] = [];
 
 /**
  * Options for useQuery change callbacks.
@@ -126,148 +128,215 @@ export function useQuery<T = any>(
   options?: UseQueryOptions<T>,
 ): UseQueryResult<T> {
   const client = useClient();
-  const [data, setData] = useState<QueryResultItem<T>[]>([]);
-  const [loading, setLoading] = useState(true);
-  const [error, setError] = useState<Error | null>(null);
 
+  // We serialize the query to use it as a stable dependency.
+  const queryJson = JSON.stringify(query);
+
+  // Construct the handle in render, memoized by [client, mapName, queryJson].
+  // client.query() is a pure constructor — it only news up a QueryHandle; the
+  // server subscription is activated by handle.subscribe(), which runs inside
+  // useExternalStore's subscribe. Memoizing keeps the handle identity stable so
+  // the store subscription is not torn down on unrelated re-renders. A throw
+  // during construction is captured (not propagated) and surfaced as `error`,
+  // matching the prior effect-based error contract.
+  const { handle, constructError } = useMemo<{
+    handle: QueryHandle<T> | null;
+    constructError: Error | null;
+  }>(
+    () => {
+      try {
+        return { handle: client.query<T>(mapName, query), constructError: null };
+      } catch (err) {
+        return {
+          handle: null,
+          constructError: err instanceof Error ? err : new Error(String(err)),
+        };
+      }
+    },
+    // queryJson is the serialized form of `query`; depending on the `query`
+    // object identity would re-create the handle every render.
+    [client, mapName, queryJson],
+  );
+
+  // ---- data + loading via useSyncExternalStore ----------------------------
+  // The handle exposes synchronous, referentially-stable accessors
+  // (getSnapshot / getSnapshotMeta). When present, the first render returns the
+  // currently-cached results with loading derived synchronously — no
+  // {loading:true, data:[]} flash when local data already exists. A
+  // hook-local cache mirrors the handle for the rare path where a handle does
+  // not expose getSnapshot (kept referentially stable to satisfy the
+  // useSyncExternalStore contract).
+  const dataCacheRef = useRef<{ value: QueryResultItem<T>[]; settled: boolean; emitted: boolean }>({
+    value: EMPTY_DATA as QueryResultItem<T>[],
+    settled: false,
+    emitted: false,
+  });
+  // Reset the local cache when the handle identity changes (new query).
+  const lastHandleRef = useRef<QueryHandle<T> | null>(null);
+  if (lastHandleRef.current !== handle) {
+    lastHandleRef.current = handle;
+    dataCacheRef.current = {
+      value: EMPTY_DATA as QueryResultItem<T>[],
+      settled: false,
+      emitted: false,
+    };
+  }
+
+  const subscribeData = useCallback(
+    (onChange: () => void) => {
+      // handle.subscribe both activates the server subscription and delivers
+      // the latest sorted results. We mirror them into the local cache (for
+      // handles without getSnapshot) and notify React.
+      if (!handle) return () => {};
+      return handle.subscribe((results, meta) => {
+        dataCacheRef.current = {
+          value: results as QueryResultItem<T>[],
+          settled: meta?.settled ?? dataCacheRef.current.settled,
+          emitted: true,
+        };
+        onChange();
+      });
+    },
+    [handle],
+  );
+
+  const getData = useCallback((): QueryResultItem<T>[] => {
+    if (handle && typeof handle.getSnapshot === 'function') {
+      return handle.getSnapshot() as QueryResultItem<T>[];
+    }
+    return dataCacheRef.current.value;
+  }, [handle]);
+
+  const data = useExternalStore(subscribeData, getData);
+
+  const loading = useMemo(() => {
+    if (constructError) return false;
+    if (handle && typeof handle.getSnapshotMeta === 'function') {
+      return !handle.getSnapshotMeta().hasEmitted;
+    }
+    return !dataCacheRef.current.emitted;
+    // `data` is included so loading re-derives after each emission for the
+    // fallback path; the getSnapshotMeta path is also re-read per render.
+  }, [handle, constructError, data]);
+
+  // ---- error: construction errors surface synchronously; runtime errors
+  // (if any) could be added to a state slot. We track a state slot for parity
+  // and OR it with the synchronous construction error.
+  const [runtimeError, setRuntimeError] = useState<Error | null>(null);
+  const error = constructError ?? runtimeError;
+
+  // ---- change accumulation (genuine reducer over a push stream) -----------
   const [changes, setChanges] = useState<ChangeEvent<T>[]>([]);
   const [lastChange, setLastChange] = useState<ChangeEvent<T> | null>(null);
 
+  // ---- pagination ----------------------------------------------------------
   const [paginationInfo, setPaginationInfo] = useState<PaginationInfo>({
     hasMore: false,
     cursorStatus: 'none',
   });
 
+  // ---- per-record sync state ----------------------------------------------
   const [syncState, setSyncState] =
     useState<ReadonlyMap<string, RecordSyncState>>(EMPTY_SYNC_STATE);
 
-  // Use a ref to track if the component is mounted to avoid state updates on unmounted components
-  const isMounted = useRef(true);
-
-  // Store handle ref for cleanup
-  const handleRef = useRef<QueryHandle<T> | null>(null);
-
-  // We serialize the query to use it as a stable dependency for the effect
-  const queryJson = JSON.stringify(query);
-
   const clearChanges = useCallback(() => {
     setChanges([]);
     setLastChange(null);
   }, []);
 
-  // Stable reference: the underlying handle can change across query-filter changes,
-  // but this ref-based callback always routes to the current handle without
-  // creating a new function identity on every render.
+  // Stable loadMore that always routes to the current handle.
   const loadMore = useCallback((): Promise<void> => {
-    const handle = handleRef.current;
-    if (!handle || !paginationInfo.hasMore) {
+    if (!handle || !paginationInfo.hasMore || typeof handle.loadMore !== 'function') {
       return Promise.resolve();
     }
     return handle.loadMore();
-  }, [paginationInfo.hasMore]);
+  }, [handle, paginationInfo.hasMore]);
 
-  // Memoize options callbacks to avoid unnecessary effect runs
+  // Keep latest options in a ref so the side-effect wiring does not re-run when
+  // only callback identities change.
   const optionsRef = useRef(options);
   optionsRef.current = options;
 
+  // Side-effecting extras: change accumulation, pagination, options callbacks,
+  // and per-record sync state. These are genuine side effects (accumulation /
+  // external callbacks) so they remain effect-based; useExternalStore handles
+  // unmount safety, so there is no isMounted ref. Re-runs only on handle change.
   useEffect(() => {
-    isMounted.current = true;
-    setLoading(true);
-
-    // Reset changes when query changes
+    // Reset accumulation when the query (handle) changes.
     setChanges([]);
     setLastChange(null);
+    setRuntimeError(null);
 
-    try {
-      const handle = client.query<T>(mapName, query);
-      handleRef.current = handle;
+    if (!handle) return;
 
-      // Subscribe to data updates
-      const unsubscribeData = handle.subscribe((results) => {
-        if (isMounted.current) {
-          setData(results);
-          setLoading(false);
+    const noop = () => {};
+    const unsubscribeChanges = handle.onDelta((newChanges) => {
+      const maxChanges = optionsRef.current?.maxChanges ?? 1000;
+
+      setChanges((prev) => {
+        const combined = [...prev, ...newChanges];
+        if (combined.length > maxChanges) {
+          return combined.slice(-maxChanges);
         }
+        return combined;
       });
 
-      const unsubscribeChanges = handle.onDelta((newChanges) => {
-        if (!isMounted.current) return;
-
-        const maxChanges = optionsRef.current?.maxChanges ?? 1000;
-
-        // Accumulate changes with rotation to prevent memory leaks
-        setChanges((prev) => {
-          const combined = [...prev, ...newChanges];
-          // Rotate oldest changes if exceeding limit
-          if (combined.length > maxChanges) {
-            return combined.slice(-maxChanges);
-          }
-          return combined;
-        });
-
-        // Track last change
-        if (newChanges.length > 0) {
-          setLastChange(newChanges[newChanges.length - 1]);
-        }
+      if (newChanges.length > 0) {
+        setLastChange(newChanges[newChanges.length - 1]);
+      }
 
-        // Invoke callbacks from options
-        const opts = optionsRef.current;
-        if (opts) {
-          for (const change of newChanges) {
-            opts.onChange?.(change);
+      const opts = optionsRef.current;
+      if (opts) {
+        for (const change of newChanges) {
+          opts.onChange?.(change);
 
-            switch (change.type) {
-              case 'add':
-                if (change.value !== undefined) {
-                  opts.onAdd?.(change.key, change.value);
-                }
-                break;
-              case 'update':
-                if (change.value !== undefined && change.previousValue !== undefined) {
-                  opts.onUpdate?.(change.key, change.value, change.previousValue);
-                }
-                break;
-              case 'remove':
-                if (change.previousValue !== undefined) {
-                  opts.onRemove?.(change.key, change.previousValue);
-                }
-                break;
-            }
+          switch (change.type) {
+            case 'add':
+              if (change.value !== undefined) {
+                opts.onAdd?.(change.key, change.value);
+              }
+              break;
+            case 'update':
+              if (change.value !== undefined && change.previousValue !== undefined) {
+                opts.onUpdate?.(change.key, change.value, change.previousValue);
+              }
+              break;
+            case 'remove':
+              if (change.previousValue !== undefined) {
+                opts.onRemove?.(change.key, change.previousValue);
+              }
+              break;
           }
         }
-      });
+      }
+    });
 
-      const unsubscribePagination = handle.onPaginationChange((info) => {
-        if (isMounted.current) {
-          setPaginationInfo(info);
-        }
-      });
+    // onPaginationChange / onSyncStateChange always exist on the real
+    // QueryHandle; guard for partial test handles that omit them.
+    const unsubscribePagination =
+      typeof handle.onPaginationChange === 'function'
+        ? handle.onPaginationChange((info) => {
+            setPaginationInfo(info);
+          })
+        : noop;
 
-      const unsubscribeSyncState = handle.onSyncStateChange((snapshot) => {
-        if (isMounted.current) {
-          setSyncState(snapshot);
-        }
-      });
+    const unsubscribeSyncState =
+      typeof handle.onSyncStateChange === 'function'
+        ? handle.onSyncStateChange((snapshot) => {
+            setSyncState(snapshot);
+          })
+        : noop;
 
-      return () => {
-        isMounted.current = false;
-        unsubscribeData();
-        unsubscribeChanges();
-        unsubscribePagination();
-        unsubscribeSyncState();
-        handleRef.current = null;
-      };
-    } catch (err) {
-      if (isMounted.current) {
-        setError(err instanceof Error ? err : new Error(String(err)));
-        setLoading(false);
-      }
-      return () => {
-        isMounted.current = false;
-        handleRef.current = null;
-      };
-    }
-  }, [client, mapName, queryJson]);
+    return () => {
+      unsubscribeChanges();
+      unsubscribePagination();
+      unsubscribeSyncState();
+    };
+    // mapName + queryJson are included so the change-accumulation reset fires
+    // whenever the query identity changes — even if a (test) client reuses the
+    // same handle object across queries. A real client returns a fresh handle,
+    // so `handle` alone would also suffice there.
+  }, [handle, mapName, queryJson]);
 
   return useMemo(
     () => ({
diff --git a/packages/react/src/hooks/useSearch.ts b/packages/react/src/hooks/useSearch.ts
index 41bf4c72..54cd73bc 100644
--- a/packages/react/src/hooks/useSearch.ts
+++ b/packages/react/src/hooks/useSearch.ts
@@ -1,7 +1,10 @@
-import { useState, useEffect, useRef, useMemo } from 'react';
+import { useState, useEffect, useRef, useMemo, useCallback } from 'react';
 import type { SearchHandle, SearchResult } from '@topgunbuild/client';
 import type { SearchOptions } from '@topgunbuild/core';
 import { useClient } from './useClient';
+import { useLocalStore } from './internal/useExternalStore';
+
+const EMPTY_SEARCH_RESULTS: SearchResult<unknown>[] = [];
 
 /**
  * Extended search options for useSearch hook.
@@ -96,13 +99,17 @@ export function useSearch<T = unknown>(
   options?: UseSearchOptions,
 ): UseSearchResult<T> {
   const client = useClient();
-  const [results, setResults] = useState<SearchResult<T>[]>([]);
+
+  // Results live in a ref and are read through useSyncExternalStore for
+  // tearing-/unmount-safety (no isMounted ref). The ref holds a
+  // referentially-stable array between notifies; notify() schedules re-renders.
+  const resultsRef = useRef<SearchResult<T>[]>(EMPTY_SEARCH_RESULTS as SearchResult<T>[]);
+  const getResultsSnapshot = useCallback(() => resultsRef.current, []);
+  const [results, notifyResults] = useLocalStore(getResultsSnapshot);
+
   const [loading, setLoading] = useState(true);
   const [error, setError] = useState<Error | null>(null);
 
-  // Track if component is mounted
-  const isMounted = useRef(true);
-
   // Store handle ref for reuse
   const handleRef = useRef<SearchHandle<T> | null>(null);
 
@@ -138,12 +145,12 @@ export function useSearch<T = unknown>(
       }
 
       debounceTimeoutRef.current = setTimeout(() => {
-        if (isMounted.current) {
-          setDebouncedQuery(query);
-        }
+        setDebouncedQuery(query);
       }, debounceMs);
 
       return () => {
+        // Clearing the pending timeout on unmount/dep-change prevents a
+        // setState after unmount — no isMounted ref needed.
         if (debounceTimeoutRef.current) {
           clearTimeout(debounceTimeoutRef.current);
         }
@@ -156,12 +163,11 @@ export function useSearch<T = unknown>(
 
   // Effect for creating/disposing handle when mapName changes
   useEffect(() => {
-    isMounted.current = true;
     isFirstQuery.current = true;
 
-    // Cleanup on mapName change or unmount
+    // Cleanup on mapName change or unmount. useSyncExternalStore handles
+    // unmount safety for the results read, so no isMounted ref is needed.
     return () => {
-      isMounted.current = false;
       if (unsubscribeRef.current) {
         unsubscribeRef.current();
         unsubscribeRef.current = null;
@@ -177,7 +183,8 @@ export function useSearch<T = unknown>(
   useEffect(() => {
     // Don't subscribe for empty queries
     if (!debouncedQuery.trim()) {
-      setResults([]);
+      resultsRef.current = EMPTY_SEARCH_RESULTS as SearchResult<T>[];
+      notifyResults();
       setLoading(false);
       setError(null);
       return;
@@ -207,24 +214,24 @@ export function useSearch<T = unknown>(
         // Flag to track if we've received initial results
         let hasReceivedResults = false;
 
-        // Subscribe to result updates
+        // Subscribe to result updates. The handle delivers a sorted array;
+        // prefer its cached getSnapshot() for referential stability when
+        // available, falling back to the delivered array.
         unsubscribeRef.current = handle.subscribe((newResults) => {
-          if (isMounted.current) {
-            setResults(newResults);
-            if (!hasReceivedResults) {
-              hasReceivedResults = true;
-              setLoading(false);
-            }
+          resultsRef.current =
+            typeof handle.getSnapshot === 'function' ? handle.getSnapshot() : newResults;
+          notifyResults();
+          if (!hasReceivedResults) {
+            hasReceivedResults = true;
+            setLoading(false);
           }
         });
       }
     } catch (err) {
-      if (isMounted.current) {
-        setError(err instanceof Error ? err : new Error(String(err)));
-        setLoading(false);
-      }
+      setError(err instanceof Error ? err : new Error(String(err)));
+      setLoading(false);
     }
-  }, [client, mapName, debouncedQuery, searchOptions]);
+  }, [client, mapName, debouncedQuery, searchOptions, notifyResults]);
 
   // Effect for handling options changes (use setOptions)
   useEffect(() => {
diff --git a/packages/react/src/hooks/useSyncState.ts b/packages/react/src/hooks/useSyncState.ts
index 012a7b8f..ec34dacd 100644
--- a/packages/react/src/hooks/useSyncState.ts
+++ b/packages/react/src/hooks/useSyncState.ts
@@ -1,6 +1,7 @@
-import { useState, useEffect, useRef } from 'react';
+import { useCallback } from 'react';
 import type { RecordSyncState } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useExternalStore } from './internal/useExternalStore';
 
 /**
  * Read the per-record sync state for a single `(mapName, key)` outside of a
@@ -22,29 +23,20 @@ import { useClient } from './useClient';
  */
 export function useSyncState(mapName: string, key: string): RecordSyncState {
   const client = useClient();
-  const [state, setState] = useState<RecordSyncState>(() =>
-    client.getRecordSyncStateTracker().get(mapName, key),
-  );
-  const isMounted = useRef(true);
 
-  useEffect(() => {
-    isMounted.current = true;
-    const tracker = client.getRecordSyncStateTracker();
-    // Re-seed on prop change in case mapName or key changed between renders.
-    setState(tracker.get(mapName, key));
-    const unsubscribe = tracker.onChange(mapName, (snapshot) => {
-      if (!isMounted.current) return;
-      // Filter internally to the specific key — only re-render when this
-      // key's projection changes, not when sibling keys do.
-      const next = snapshot.get(key) ?? 'synced';
-      setState((prev) => (prev === next ? prev : next));
-    });
+  // Subscribe to the map's tracker channel. The returned RecordSyncState is a
+  // string primitive, so getSnapshot is referentially stable by value — React
+  // only re-renders when THIS key's projection changes, even though the tracker
+  // fires for any key in the map.
+  const subscribe = useCallback(
+    (onChange: () => void) => client.getRecordSyncStateTracker().onChange(mapName, onChange),
+    [client, mapName],
+  );
 
-    return () => {
-      isMounted.current = false;
-      unsubscribe();
-    };
-  }, [client, mapName, key]);
+  const getSnapshot = useCallback(
+    () => client.getRecordSyncStateTracker().get(mapName, key),
+    [client, mapName, key],
+  );
 
-  return state;
+  return useExternalStore(subscribe, getSnapshot);
 }
diff --git a/packages/react/src/hooks/useTopic.ts b/packages/react/src/hooks/useTopic.ts
index b563964b..2d8093bc 100644
--- a/packages/react/src/hooks/useTopic.ts
+++ b/packages/react/src/hooks/useTopic.ts
@@ -1,34 +1,56 @@
-import { useEffect, useRef } from 'react';
+import { useCallback, useMemo, useRef } from 'react';
 import { TopicCallback } from '@topgunbuild/client';
 import { useClient } from './useClient';
+import { useExternalStore } from './internal/useExternalStore';
 
+/**
+ * Subscribe to a pub/sub topic. Returns the (stable, cached) topic handle so
+ * callers can `topic.publish(...)`.
+ *
+ * The subscription is keyed ONLY by the topic identity — NOT by the callback
+ * identity. The latest callback is kept in a ref and invoked on each message,
+ * so passing a fresh inline callback every render (the common
+ * `useTopic('chat', d => …)` usage) does NOT tear down and re-create the
+ * subscription. This is the TODO-516 churn fix: previously the effect depended
+ * on `[topic, callback]`, so every render re-subscribed and the `callbackRef`
+ * indirection was dead code.
+ */
 export function useTopic(topicName: string, callback?: TopicCallback) {
   const client = useClient();
-  const topic = client.topic(topicName);
-  const isMounted = useRef(true);
+  // Memoize the handle lookup by name (render purity); topic handles are cached
+  // by the client, so this is the same object across renders anyway.
+  const topic = useMemo(() => client.topic(topicName), [client, topicName]);
 
-  // Keep callback ref stable to avoid re-subscribing if callback function identity changes
+  // Keep the latest callback in a ref so the subscription does not depend on
+  // its identity. Updated synchronously during render so the freshest callback
+  // is live before the next message dispatch.
   const callbackRef = useRef(callback);
-  useEffect(() => {
-    callbackRef.current = callback;
-  }, [callback]);
+  callbackRef.current = callback;
 
-  useEffect(() => {
-    isMounted.current = true;
-
-    if (!callback) return;
-
-    const unsubscribe = topic.subscribe((data, context) => {
-      if (isMounted.current && callbackRef.current) {
-        callbackRef.current(data, context);
+  // Subscribe is stable per-topic. We only register a real listener when a
+  // callback is present; otherwise we register a noop unsubscribe so
+  // useExternalStore has a valid (stable) subscribe with no churn. Presence
+  // (not identity) of the callback is the only thing that toggles subscription.
+  const hasCallback = !!callback;
+  const subscribe = useCallback(
+    (_onStoreChange: () => void) => {
+      if (!hasCallback) {
+        return () => {};
       }
-    });
+      return topic.subscribe((data, context) => {
+        callbackRef.current?.(data, context);
+      });
+    },
+    [topic, hasCallback],
+  );
 
-    return () => {
-      isMounted.current = false;
-      unsubscribe();
-    };
-  }, [topic, callback]); // Re-subscribe if topic handle changes (rare) or if callback presence toggles
+  // This hook does not surface a value to render — the topic handle is the
+  // return value and is stable. We still route through useExternalStore so the
+  // subscription lifecycle (mount/unmount, tearing-safety) is managed by React
+  // rather than a hand-rolled effect + isMounted ref. The snapshot is a
+  // constant so it never triggers a re-render on its own.
+  const getSnapshot = useCallback(() => topic, [topic]);
+  useExternalStore(subscribe, getSnapshot);
 
   return topic;
 }
diff --git a/packages/server-rust/src/network/config.rs b/packages/server-rust/src/network/config.rs
index c8ee19d2..00ad7068 100644
--- a/packages/server-rust/src/network/config.rs
+++ b/packages/server-rust/src/network/config.rs
@@ -140,6 +140,26 @@ pub struct ConnectionConfig {
     /// Maximum size in bytes of a single inbound WebSocket frame (a message may
     /// span multiple frames). Defaults to 2 MB.
     pub ws_max_frame_size: usize,
+    /// Consecutive live-event broadcasts dropped (outbound channel full) before
+    /// a slow client connection is disconnected to force reconnect + Merkle
+    /// resync, instead of letting it diverge silently. Reset on any successful
+    /// broadcast, so a connection that drains at all is never disconnected by
+    /// this — only one stuck for `threshold` events in a row. 0 disables the
+    /// disconnect (pure best-effort drop). Defaults to 256 (one full outbound
+    /// buffer's worth of consecutive misses).
+    pub slow_consumer_drop_threshold: u64,
+    /// Sustained inbound op-rate ceiling, in ops/second, applied per WebSocket
+    /// connection on the data plane (Phase-2 ops). Layered on top of aggregate
+    /// load shedding so a single abusive peer cannot saturate the in-flight
+    /// pipeline for everyone; over-rate ops get a 429 back-off without tearing
+    /// the connection down. 0 disables per-connection limiting. The default
+    /// (20 000) leaves large headroom over realistic client and benchmark
+    /// per-connection rates while still bounding a flood.
+    pub data_plane_max_ops_per_sec: u32,
+    /// Burst capacity for the per-connection inbound op-rate limiter (the most
+    /// ops acceptable instantaneously after an idle period). Defaults to 40 000
+    /// (2× the sustained rate).
+    pub data_plane_ops_burst: u32,
 }
 
 impl Default for ConnectionConfig {
@@ -154,6 +174,9 @@ impl Default for ConnectionConfig {
             ws_max_write_buffer_size: 524_288,    // 512 KB
             ws_max_message_size: 2 * 1024 * 1024, // 2 MB — matches HTTP /sync body limit
             ws_max_frame_size: 2 * 1024 * 1024,   // 2 MB
+            slow_consumer_drop_threshold: 256,    // == default outbound capacity
+            data_plane_max_ops_per_sec: 20_000,   // generous; bounds a flooding peer
+            data_plane_ops_burst: 40_000,         // 2× sustained
         }
     }
 }
@@ -194,6 +217,9 @@ mod tests {
         assert_eq!(config.ws_max_write_buffer_size, 524_288);
         assert_eq!(config.ws_max_message_size, 2 * 1024 * 1024);
         assert_eq!(config.ws_max_frame_size, 2 * 1024 * 1024);
+        assert_eq!(config.slow_consumer_drop_threshold, 256);
+        assert_eq!(config.data_plane_max_ops_per_sec, 20_000);
+        assert_eq!(config.data_plane_ops_burst, 40_000);
     }
 
     #[test]
diff --git a/packages/server-rust/src/network/connection.rs b/packages/server-rust/src/network/connection.rs
index 9abe0595..cede6d27 100644
--- a/packages/server-rust/src/network/connection.rs
+++ b/packages/server-rust/src/network/connection.rs
@@ -10,9 +10,11 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use dashmap::DashMap;
+use tokio::sync::mpsc::error::TrySendError;
 use tokio::sync::{mpsc, RwLock};
 use tokio_util::sync::CancellationToken;
 use topgun_core::{Principal, Timestamp};
+use tracing::warn;
 
 use super::config::ConnectionConfig;
 
@@ -49,6 +51,27 @@ pub enum SendError {
     Full,
 }
 
+/// Outcome of a best-effort broadcast send to a (possibly slow) consumer.
+///
+/// Distinguishes the three states a live-event push can land in so the caller
+/// (and metrics) can tell "kept up" from "fell behind" from "gave up", instead
+/// of the old binary `is_ok()` that hid silent divergence.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BroadcastOutcome {
+    /// The event was enqueued; the consumer is keeping up.
+    Sent,
+    /// The channel was full and the event was dropped, but the consecutive-drop
+    /// count is still under the disconnect threshold. The connection survives;
+    /// the per-connection dropped-event counter was incremented.
+    Dropped,
+    /// The consecutive-drop count crossed the threshold: the connection was
+    /// cancelled to force a reconnect + Merkle resync rather than let it diverge
+    /// silently. Only client connections are disconnected this way.
+    Disconnected,
+    /// The receiver is gone (connection already closed).
+    Closed,
+}
+
 /// Handle to a single connection, providing send capabilities and metadata access.
 ///
 /// Each connection gets a bounded mpsc channel for backpressure. The receiver
@@ -72,6 +95,18 @@ pub struct ConnectionHandle {
     /// run its normal cleanup path. The reaper cancels this to evict stale or
     /// never-authenticated connections.
     pub cancel: CancellationToken,
+    /// Number of consecutive live-event broadcasts dropped because the outbound
+    /// channel was full. Reset to 0 on the next successful broadcast. When this
+    /// crosses `slow_consumer_drop_threshold` the connection is cancelled
+    /// (forcing reconnect + resync) instead of diverging silently.
+    consecutive_drops: AtomicU64,
+    /// Cumulative count of live-event broadcasts dropped over the connection's
+    /// lifetime. Surfaced as a slow-consumer metric; never reset.
+    dropped_broadcasts: AtomicU64,
+    /// Consecutive-drop count at which a slow client connection is disconnected
+    /// to force resync. 0 disables the disconnect behavior (pure best-effort
+    /// drop, the legacy semantics).
+    slow_consumer_drop_threshold: u64,
 }
 
 impl ConnectionHandle {
@@ -84,6 +119,66 @@ impl ConnectionHandle {
         self.tx.try_send(msg).is_ok()
     }
 
+    /// Best-effort send of a live-event broadcast to a possibly-slow consumer.
+    ///
+    /// Unlike [`try_send`](Self::try_send), this tracks slow-consumer state so a
+    /// subscriber that persistently fails to drain its channel does not diverge
+    /// silently. On a full channel the event is dropped (the channel stays
+    /// memory-bounded), the cumulative drop metric is incremented, and the
+    /// consecutive-drop counter advances; once it reaches
+    /// `slow_consumer_drop_threshold` the connection is cancelled so the client
+    /// reconnects and re-syncs via the Merkle tree. A successful send resets the
+    /// consecutive-drop counter.
+    ///
+    /// Only `Client` connections are disconnected on threshold; cluster peers
+    /// fall back to plain best-effort drop because their liveness is governed by
+    /// the cluster failure detector, not this idle/slow heuristic.
+    #[must_use]
+    pub fn try_send_broadcast(&self, msg: OutboundMessage) -> BroadcastOutcome {
+        match self.tx.try_send(msg) {
+            Ok(()) => {
+                self.consecutive_drops.store(0, Ordering::Relaxed);
+                BroadcastOutcome::Sent
+            }
+            Err(TrySendError::Closed(_)) => BroadcastOutcome::Closed,
+            Err(TrySendError::Full(_)) => {
+                // Relaxed ordering is sufficient: concurrent broadcasts to one
+                // connection can in principle race the success-reset against a
+                // drop-increment at the exact threshold boundary, disconnecting a
+                // connection one event early. That is a SAFE failure mode — the
+                // disconnect just forces the reconnect + Merkle resync this whole
+                // path exists to trigger, with no data loss — so it does not
+                // warrant a compare-exchange loop on the hot broadcast path.
+                self.dropped_broadcasts.fetch_add(1, Ordering::Relaxed);
+                let consecutive = self.consecutive_drops.fetch_add(1, Ordering::Relaxed) + 1;
+                let should_disconnect = self.kind == ConnectionKind::Client
+                    && self.slow_consumer_drop_threshold != 0
+                    && consecutive >= self.slow_consumer_drop_threshold
+                    && !self.cancel.is_cancelled();
+                if should_disconnect {
+                    warn!(
+                        conn_id = ?self.id,
+                        consecutive_drops = consecutive,
+                        dropped_total = self.dropped_broadcasts.load(Ordering::Relaxed),
+                        "slow consumer exceeded broadcast drop threshold; \
+                         disconnecting to force reconnect + resync"
+                    );
+                    self.cancel();
+                    BroadcastOutcome::Disconnected
+                } else {
+                    BroadcastOutcome::Dropped
+                }
+            }
+        }
+    }
+
+    /// Cumulative number of live-event broadcasts dropped to this connection
+    /// because its outbound channel was full (slow-consumer metric).
+    #[must_use]
+    pub fn dropped_broadcasts(&self) -> u64 {
+        self.dropped_broadcasts.load(Ordering::Relaxed)
+    }
+
     /// Sends a message with a timeout.
     ///
     /// # Errors
@@ -221,6 +316,9 @@ impl ConnectionRegistry {
             connected_at: Instant::now(),
             kind,
             cancel: CancellationToken::new(),
+            consecutive_drops: AtomicU64::new(0),
+            dropped_broadcasts: AtomicU64::new(0),
+            slow_consumer_drop_threshold: config.slow_consumer_drop_threshold,
         });
 
         self.connections.insert(id, Arc::clone(&handle));
@@ -264,29 +362,31 @@ impl ConnectionRegistry {
             .collect()
     }
 
-    /// Sends a binary message to all connections of a given kind.
+    /// Sends a binary live-event message to all connections of a given kind.
     ///
-    /// Uses non-blocking `try_send` so a single slow connection cannot
-    /// block the broadcast. Full channels are silently skipped.
+    /// Uses non-blocking [`try_send_broadcast`](ConnectionHandle::try_send_broadcast)
+    /// so a single slow connection cannot block the broadcast. A full channel
+    /// drops the event (bounded memory) and advances that connection's
+    /// slow-consumer state; a persistently-slow client is disconnected to force
+    /// reconnect + resync rather than diverging silently.
     pub fn broadcast(&self, msg_bytes: &[u8], kind: ConnectionKind) {
         for entry in &self.connections {
             let handle = entry.value();
             if handle.kind == kind {
-                // Intentionally ignore the result: broadcast skips full channels
-                let _ = handle.try_send(OutboundMessage::Binary(msg_bytes.to_vec()));
+                let _ = handle.try_send_broadcast(OutboundMessage::Binary(msg_bytes.to_vec()));
             }
         }
     }
 
-    /// Sends a binary message to a specific set of connection IDs.
+    /// Sends a binary live-event message to a specific set of connection IDs.
     ///
-    /// Uses non-blocking `try_send` so a single slow connection cannot
-    /// block the caller. Missing connections and full channels are silently
-    /// skipped (same semantics as `broadcast`).
+    /// Same slow-consumer semantics as [`broadcast`](Self::broadcast): missing
+    /// connections are skipped, a full channel drops the event and advances the
+    /// target's slow-consumer state (disconnecting a persistently-slow client).
     pub fn send_to_connections(&self, ids: &HashSet<ConnectionId>, msg_bytes: &[u8]) {
         for id in ids {
             if let Some(handle) = self.get(*id) {
-                let _ = handle.try_send(OutboundMessage::Binary(msg_bytes.to_vec()));
+                let _ = handle.try_send_broadcast(OutboundMessage::Binary(msg_bytes.to_vec()));
             }
         }
     }
@@ -628,4 +728,195 @@ mod tests {
         // Should not block or panic when channel is full
         registry.send_to_connections(&ids, &[3]);
     }
+
+    // ---------------------------------------------------------------------------
+    // Slow-consumer broadcast backpressure (F4 / TODO-509)
+    //
+    // A persistently-slow subscriber used to be dropped silently (no error, no
+    // Close, no counter) and diverge until it independently reconnected. The
+    // fix counts consecutive drops and, past a threshold, disconnects the
+    // connection to force reconnect + Merkle resync, while exposing a drop
+    // metric. These tests assert the new contract *and* carry the pre-fix
+    // negative controls (draining consumer / cluster peer never disconnected).
+    // ---------------------------------------------------------------------------
+
+    fn slow_consumer_config(capacity: usize, threshold: u64) -> ConnectionConfig {
+        ConnectionConfig {
+            outbound_channel_capacity: capacity,
+            slow_consumer_drop_threshold: threshold,
+            ..ConnectionConfig::default()
+        }
+    }
+
+    #[test]
+    fn slow_client_disconnected_after_consecutive_drop_threshold() {
+        // Reproduces the F4 scenario: a client whose channel stays full. Before
+        // the fix it silently dropped forever and stayed connected; now it is
+        // disconnected once consecutive drops reach the threshold.
+        let config = slow_consumer_config(2, 3);
+        let registry = ConnectionRegistry::new();
+        let (handle, _rx) = registry.register(ConnectionKind::Client, &config);
+
+        // Fill the channel (capacity 2) so every subsequent broadcast drops.
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![1])),
+            BroadcastOutcome::Sent
+        );
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![2])),
+            BroadcastOutcome::Sent
+        );
+
+        // Drops 1 and 2 are under threshold (3): event lost but connection lives.
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![3])),
+            BroadcastOutcome::Dropped
+        );
+        assert!(!handle.is_cancelled());
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![4])),
+            BroadcastOutcome::Dropped
+        );
+        assert!(!handle.is_cancelled());
+
+        // Drop 3 hits the threshold: disconnect to force resync.
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![5])),
+            BroadcastOutcome::Disconnected
+        );
+        assert!(
+            handle.is_cancelled(),
+            "slow client must be cancelled past the drop threshold, not diverge silently"
+        );
+        assert_eq!(
+            handle.dropped_broadcasts(),
+            3,
+            "drop metric must count losses"
+        );
+    }
+
+    #[test]
+    fn draining_consumer_is_never_disconnected() {
+        // Negative control: a consumer that drains its channel resets the
+        // consecutive-drop counter on each successful send and is never
+        // disconnected, even across far more than `threshold` broadcasts.
+        let config = slow_consumer_config(2, 3);
+        let registry = ConnectionRegistry::new();
+        let (handle, mut rx) = registry.register(ConnectionKind::Client, &config);
+
+        for i in 0..100u8 {
+            let outcome = handle.try_send_broadcast(OutboundMessage::Binary(vec![i]));
+            assert_eq!(outcome, BroadcastOutcome::Sent);
+            // Drain immediately so the channel never fills.
+            assert!(rx.try_recv().is_ok());
+        }
+
+        assert!(!handle.is_cancelled(), "a draining consumer must survive");
+        assert_eq!(handle.dropped_broadcasts(), 0);
+    }
+
+    #[test]
+    fn successful_broadcast_resets_consecutive_drops() {
+        // A few drops followed by a successful send must reset the streak, so a
+        // connection that occasionally falls behind but recovers is not reaped.
+        let config = slow_consumer_config(1, 3);
+        let registry = ConnectionRegistry::new();
+        let (handle, mut rx) = registry.register(ConnectionKind::Client, &config);
+
+        // Fill (cap 1), then two drops (under threshold 3).
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![1])),
+            BroadcastOutcome::Sent
+        );
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![2])),
+            BroadcastOutcome::Dropped
+        );
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![3])),
+            BroadcastOutcome::Dropped
+        );
+
+        // Drain, then a successful send resets the streak.
+        assert!(rx.try_recv().is_ok());
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![4])),
+            BroadcastOutcome::Sent
+        );
+
+        // Now the channel is full again; it takes another full `threshold` drops
+        // (not just one) to disconnect — proving the reset took effect.
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![5])),
+            BroadcastOutcome::Dropped
+        );
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![6])),
+            BroadcastOutcome::Dropped
+        );
+        assert!(!handle.is_cancelled());
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![7])),
+            BroadcastOutcome::Disconnected
+        );
+        assert!(handle.is_cancelled());
+    }
+
+    #[test]
+    fn cluster_peer_is_not_disconnected_on_drops() {
+        // Negative control: cluster peers fall back to best-effort drop; their
+        // liveness is the cluster failure detector's job, not this heuristic.
+        let config = slow_consumer_config(1, 1);
+        let registry = ConnectionRegistry::new();
+        let (handle, _rx) = registry.register(ConnectionKind::ClusterPeer, &config);
+
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![1])),
+            BroadcastOutcome::Sent
+        );
+        for _ in 0..10 {
+            assert_eq!(
+                handle.try_send_broadcast(OutboundMessage::Binary(vec![2])),
+                BroadcastOutcome::Dropped
+            );
+        }
+        assert!(
+            !handle.is_cancelled(),
+            "cluster peer must never be disconnected by the slow-consumer heuristic"
+        );
+        assert_eq!(handle.dropped_broadcasts(), 10);
+    }
+
+    #[test]
+    fn broadcast_outcome_closed_when_receiver_gone() {
+        let config = slow_consumer_config(2, 3);
+        let registry = ConnectionRegistry::new();
+        let (handle, rx) = registry.register(ConnectionKind::Client, &config);
+        drop(rx);
+
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![1])),
+            BroadcastOutcome::Closed
+        );
+    }
+
+    #[test]
+    fn threshold_zero_disables_disconnect() {
+        // threshold = 0 preserves the legacy pure best-effort drop semantics.
+        let config = slow_consumer_config(1, 0);
+        let registry = ConnectionRegistry::new();
+        let (handle, _rx) = registry.register(ConnectionKind::Client, &config);
+
+        assert_eq!(
+            handle.try_send_broadcast(OutboundMessage::Binary(vec![1])),
+            BroadcastOutcome::Sent
+        );
+        for _ in 0..50 {
+            assert_eq!(
+                handle.try_send_broadcast(OutboundMessage::Binary(vec![2])),
+                BroadcastOutcome::Dropped
+            );
+        }
+        assert!(!handle.is_cancelled());
+    }
 }
diff --git a/packages/server-rust/src/network/handlers/websocket.rs b/packages/server-rust/src/network/handlers/websocket.rs
index 7113dab1..9eb2c714 100644
--- a/packages/server-rust/src/network/handlers/websocket.rs
+++ b/packages/server-rust/src/network/handlers/websocket.rs
@@ -283,6 +283,17 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) {
         meta.principal.clone()
     };
 
+    // Per-connection inbound op-rate limiter (data-plane fairness). Aggregate
+    // load shedding (MAX_IN_FLIGHT + worker-inbox Overloaded) bounds total work
+    // but not a single abusive peer; this token bucket caps one connection's op
+    // rate so a flood is throttled (429 back-off) without starving others or
+    // tearing the connection down. Owned solely by this read loop — no locking.
+    let mut rate_limiter = crate::network::rate_limit::TokenBucket::new(
+        state.config.connection.data_plane_max_ops_per_sec,
+        state.config.connection.data_plane_ops_burst,
+        std::time::Instant::now(),
+    );
+
     // Phase 2: pipeline mode — each binary frame is dispatched concurrently.
     // The reader continues immediately after spawning, so multiple frames
     // can be in-flight simultaneously up to MAX_IN_FLIGHT.
@@ -308,6 +319,36 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) {
                     }
                 };
 
+                // Per-connection inbound op-rate limit. Cost = number of ops the
+                // frame carries (a batch counts as its op count) so one peer's
+                // flood is throttled fairly. On exceed we send a 429 back-off and
+                // drop the frame — the connection stays up and recovers as tokens
+                // refill (отбой, не падение).
+                let op_cost = inbound_op_cost(&tg_msg);
+                if !rate_limiter.try_consume(op_cost) {
+                    debug!(
+                        "rate limit exceeded for {:?} (op_cost={}); backing off",
+                        conn_id, op_cost
+                    );
+                    let err_msg = TopGunMessage::Error {
+                        payload: ErrorPayload {
+                            code: 429,
+                            message: "rate limit exceeded, slow down".to_string(),
+                            details: None,
+                        },
+                    };
+                    if let Ok(bytes) = rmp_serde::to_vec_named(&err_msg) {
+                        // Best-effort, non-blocking: a flooding client is by
+                        // definition behind on its outbound channel, so awaiting
+                        // here would stall this read loop on the very connection
+                        // we are throttling (and would gate its own recovery
+                        // tokens). If the channel is full, dropping the 429 is
+                        // fine — the client is already getting backpressure.
+                        let _ = handle.try_send(OutboundMessage::Binary(bytes));
+                    }
+                    continue;
+                }
+
                 // Acquire a permit before spawning; if the semaphore is closed
                 // (shutdown signal), exit the reader loop.
                 let Ok(permit) = semaphore.clone().acquire_owned().await else {
@@ -414,6 +455,21 @@ fn release_session_state(state: &AppState, conn_id: ConnectionId) {
 /// individually. Non-BATCH messages are classified, have `connection_id`
 /// and `principal` set, and are routed through the pipeline. Each
 /// `OperationResponse` variant is mapped to the appropriate outbound message(s).
+/// Cost, in op-rate-limiter tokens, of an inbound frame.
+///
+/// A frame's cost is the number of individual ops it carries so a single peer
+/// cannot evade the per-connection rate limit by packing many ops into one
+/// `OpBatch`/`Batch` frame. Non-batch messages cost one token. Always at least
+/// 1 so an empty batch still consumes a token (and cannot be used to spin).
+fn inbound_op_cost(msg: &TopGunMessage) -> u32 {
+    let count = match msg {
+        TopGunMessage::OpBatch(b) => b.payload.ops.len(),
+        TopGunMessage::Batch(b) => b.count as usize,
+        _ => 1,
+    };
+    u32::try_from(count.max(1)).unwrap_or(u32::MAX)
+}
+
 async fn dispatch_message(
     tg_msg: TopGunMessage,
     conn_id: ConnectionId,
@@ -878,6 +934,43 @@ mod tests {
     use dashmap::DashSet;
     use topgun_core::messages::base::Query;
     use topgun_core::messages::search::SearchOptions;
+    use topgun_core::messages::{BatchMessage, ClientOp, OpBatchMessage, OpBatchPayload};
+
+    /// The op-rate limiter charges a frame by the number of ops it carries, so a
+    /// peer cannot evade the per-connection cap by packing ops into one batch.
+    #[test]
+    fn inbound_op_cost_counts_batch_ops() {
+        // A non-batch message costs one token.
+        let ping = TopGunMessage::Ping(topgun_core::messages::PingData { timestamp: 0 });
+        assert_eq!(inbound_op_cost(&ping), 1);
+
+        // An OpBatch costs its op count.
+        let three_ops = TopGunMessage::OpBatch(OpBatchMessage {
+            payload: OpBatchPayload {
+                ops: vec![
+                    ClientOp::default(),
+                    ClientOp::default(),
+                    ClientOp::default(),
+                ],
+                write_concern: None,
+                timeout: None,
+            },
+        });
+        assert_eq!(inbound_op_cost(&three_ops), 3);
+
+        // An empty batch still costs one token (cannot be used to spin for free).
+        let empty = TopGunMessage::OpBatch(OpBatchMessage {
+            payload: OpBatchPayload::default(),
+        });
+        assert_eq!(inbound_op_cost(&empty), 1);
+
+        // A transport Batch costs its declared message count.
+        let batch = TopGunMessage::Batch(BatchMessage {
+            count: 5,
+            data: vec![],
+        });
+        assert_eq!(inbound_op_cost(&batch), 5);
+    }
 
     /// Disconnect cleanup must drain a connection's standing subscriptions from
     /// every registry wired into `AppState`, not just lock/topic/counter. This
diff --git a/packages/server-rust/src/network/mod.rs b/packages/server-rust/src/network/mod.rs
index 9bee568f..de72e1cf 100644
--- a/packages/server-rust/src/network/mod.rs
+++ b/packages/server-rust/src/network/mod.rs
@@ -6,6 +6,7 @@ pub mod handlers;
 pub mod middleware;
 pub mod module;
 pub mod openapi;
+pub mod rate_limit;
 pub mod reaper;
 pub mod shutdown;
 
@@ -13,5 +14,6 @@ pub use config::*;
 pub use connection::*;
 pub use handlers::AppState;
 pub use module::NetworkModule;
+pub use rate_limit::TokenBucket;
 pub use reaper::{spawn_connection_reaper, ReaperConfig};
 pub use shutdown::*;
diff --git a/packages/server-rust/src/network/rate_limit.rs b/packages/server-rust/src/network/rate_limit.rs
new file mode 100644
index 00000000..487f2404
--- /dev/null
+++ b/packages/server-rust/src/network/rate_limit.rs
@@ -0,0 +1,179 @@
+//! Per-connection inbound op-rate limiting for the WebSocket data plane.
+//!
+//! The `tower_governor` rate limiter only guards the admin/login HTTP routes;
+//! the data plane (`/ws`, `/sync`) relies on aggregate load shedding
+//! (`MAX_IN_FLIGHT` + worker-inbox `Overloaded`). Load shedding bounds total
+//! in-flight work but does **not** bound a single abusive peer: one connection
+//! can continuously occupy the in-flight slots and saturate worker inboxes,
+//! degrading every other client.
+//!
+//! [`TokenBucket`] adds a per-connection cap on inbound op rate so one peer's
+//! flood is throttled (the client is told to back off with a 429) without
+//! starving others and without tearing the connection down.
+//!
+//! Deliberately **not** thread-safe: each connection's Phase-2 read loop is the
+//! sole owner of its bucket, so no locking is required.
+
+use std::time::Instant;
+
+/// A monotonic token bucket.
+///
+/// Tokens refill continuously at `refill_per_sec` up to `capacity` (the burst
+/// ceiling). Each accepted op consumes tokens equal to its cost; when the
+/// bucket cannot cover the cost the op is rejected (the caller backs the client
+/// off) but the connection survives and recovers as tokens refill — "отбой, не
+/// падение".
+///
+/// A `refill_per_sec` of 0 disables limiting entirely (every op is allowed),
+/// preserving the legacy unlimited behavior when an operator opts out.
+#[derive(Debug)]
+pub struct TokenBucket {
+    capacity: f64,
+    tokens: f64,
+    refill_per_sec: f64,
+    last_refill: Instant,
+}
+
+impl TokenBucket {
+    /// Creates a full bucket as of `now`.
+    ///
+    /// `refill_per_sec` is the sustained ops/second; `burst` is the bucket
+    /// capacity (the most ops that can be accepted instantaneously after an idle
+    /// period). `burst` is floored to 1 when non-zero refill is configured so a
+    /// misconfigured `burst = 0` cannot wedge the connection.
+    #[must_use]
+    pub fn new(refill_per_sec: u32, burst: u32, now: Instant) -> Self {
+        let capacity = if refill_per_sec > 0 {
+            f64::from(burst.max(1))
+        } else {
+            f64::from(burst)
+        };
+        Self {
+            capacity,
+            tokens: capacity,
+            refill_per_sec: f64::from(refill_per_sec),
+            last_refill: now,
+        }
+    }
+
+    /// Whether limiting is disabled (refill rate 0).
+    #[must_use]
+    pub fn is_disabled(&self) -> bool {
+        self.refill_per_sec <= 0.0
+    }
+
+    /// Attempts to consume `cost` tokens as of `now`. Returns `true` if the op
+    /// is allowed.
+    ///
+    /// `cost` is clamped to `capacity` so a single legitimate op (or batch)
+    /// larger than the burst ceiling is never permanently rejected — it simply
+    /// drains the bucket and the next ops wait for refill.
+    pub fn try_consume_at(&mut self, cost: u32, now: Instant) -> bool {
+        if self.is_disabled() {
+            return true;
+        }
+        let elapsed = now
+            .saturating_duration_since(self.last_refill)
+            .as_secs_f64();
+        self.last_refill = now;
+        self.tokens = (self.tokens + elapsed * self.refill_per_sec).min(self.capacity);
+
+        let cost = f64::from(cost).min(self.capacity);
+        if self.tokens >= cost {
+            self.tokens -= cost;
+            true
+        } else {
+            false
+        }
+    }
+
+    /// Convenience wrapper over [`try_consume_at`](Self::try_consume_at) using
+    /// the current instant.
+    pub fn try_consume(&mut self, cost: u32) -> bool {
+        self.try_consume_at(cost, Instant::now())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use super::*;
+
+    #[test]
+    fn disabled_bucket_always_allows() {
+        let mut b = TokenBucket::new(0, 0, Instant::now());
+        assert!(b.is_disabled());
+        for _ in 0..10_000 {
+            assert!(b.try_consume(1_000));
+        }
+    }
+
+    #[test]
+    fn allows_up_to_burst_then_throttles() {
+        // Flood within a single instant: burst 10 allows 10 single-op consumes,
+        // the 11th is throttled (rejected, not a panic) — отбой, не падение.
+        let t0 = Instant::now();
+        let mut b = TokenBucket::new(100, 10, t0);
+        for i in 0..10 {
+            assert!(b.try_consume_at(1, t0), "op {i} within burst must pass");
+        }
+        assert!(
+            !b.try_consume_at(1, t0),
+            "op past burst must be throttled, not crash"
+        );
+    }
+
+    #[test]
+    fn refills_over_time_and_recovers() {
+        // After exhausting the burst, the bucket recovers as time passes — the
+        // peer is slowed, never permanently shut out.
+        let t0 = Instant::now();
+        let mut b = TokenBucket::new(100, 10, t0); // 100 ops/sec
+        for _ in 0..10 {
+            assert!(b.try_consume_at(1, t0));
+        }
+        assert!(!b.try_consume_at(1, t0));
+
+        // 100ms later → 10 tokens refilled (100/sec * 0.1s).
+        let t1 = t0 + Duration::from_millis(100);
+        for _ in 0..10 {
+            assert!(b.try_consume_at(1, t1), "bucket must recover after refill");
+        }
+        assert!(!b.try_consume_at(1, t1));
+    }
+
+    #[test]
+    fn batch_cost_drains_proportionally() {
+        let t0 = Instant::now();
+        let mut b = TokenBucket::new(1_000, 100, t0);
+        // One 60-op batch then a 40-op batch exactly drains the burst of 100.
+        assert!(b.try_consume_at(60, t0));
+        assert!(b.try_consume_at(40, t0));
+        // Bucket empty: a further op is throttled.
+        assert!(!b.try_consume_at(1, t0));
+    }
+
+    #[test]
+    fn oversized_op_clamped_not_permanently_rejected() {
+        // A single op whose cost exceeds the burst ceiling still passes against a
+        // full bucket (cost clamped to capacity) instead of wedging forever.
+        let t0 = Instant::now();
+        let mut b = TokenBucket::new(1_000, 100, t0);
+        assert!(
+            b.try_consume_at(10_000, t0),
+            "oversized op must pass against a full bucket"
+        );
+        // It drained the bucket, so the next op waits for refill.
+        assert!(!b.try_consume_at(1, t0));
+    }
+
+    #[test]
+    fn zero_burst_with_refill_is_floored_to_one() {
+        let t0 = Instant::now();
+        let mut b = TokenBucket::new(100, 0, t0);
+        // Floored to capacity 1 so a misconfig cannot wedge the connection.
+        assert!(b.try_consume_at(1, t0));
+        assert!(!b.try_consume_at(1, t0));
+    }
+}
diff --git a/packages/server-rust/src/service/domain/crdt.rs b/packages/server-rust/src/service/domain/crdt.rs
index f7b0ce8b..7709b9ce 100644
--- a/packages/server-rust/src/service/domain/crdt.rs
+++ b/packages/server-rust/src/service/domain/crdt.rs
@@ -901,7 +901,7 @@ impl CrdtService {
                 if let Ok(bytes) = rmp_serde::to_vec_named(&msg) {
                     use crate::network::connection::OutboundMessage;
                     if let Some(handle) = self.connection_registry.get(sub.connection_id) {
-                        let _ = handle.try_send(OutboundMessage::Binary(bytes));
+                        let _ = handle.try_send_broadcast(OutboundMessage::Binary(bytes));
                     }
                 }
             }
diff --git a/packages/server-rust/src/service/domain/persistence.rs b/packages/server-rust/src/service/domain/persistence.rs
index 366bbd99..3469f91a 100644
--- a/packages/server-rust/src/service/domain/persistence.rs
+++ b/packages/server-rust/src/service/domain/persistence.rs
@@ -288,8 +288,10 @@ impl PersistenceService {
                 continue;
             }
             if let Some(handle) = self.connection_registry.get(sub_conn_id) {
-                // Best-effort delivery: skip full channels.
-                let _ = handle.try_send(OutboundMessage::Binary(bytes.clone()));
+                // Best-effort live delivery: a full channel drops the event and
+                // advances the target's slow-consumer state (force-resync on a
+                // persistently-slow client) rather than diverging silently.
+                let _ = handle.try_send_broadcast(OutboundMessage::Binary(bytes.clone()));
             }
         }
 
diff --git a/packages/server-rust/src/service/domain/query.rs b/packages/server-rust/src/service/domain/query.rs
index 5b3aeb4d..f73b3164 100644
--- a/packages/server-rust/src/service/domain/query.rs
+++ b/packages/server-rust/src/service/domain/query.rs
@@ -250,7 +250,7 @@ impl QueryMutationObserver {
         let msg = Message::QueryUpdate { payload };
         if let Ok(bytes) = rmp_serde::to_vec_named(&msg) {
             if let Some(handle) = self.connection_registry.get(sub.connection_id) {
-                let _ = handle.try_send(OutboundMessage::Binary(bytes));
+                let _ = handle.try_send_broadcast(OutboundMessage::Binary(bytes));
             }
         }
     }