diff --git a/packages/dofs/src/sync/apply.test.ts b/packages/dofs/src/sync/apply.test.ts
index b326e0be..ed60c8c0 100644
--- a/packages/dofs/src/sync/apply.test.ts
+++ b/packages/dofs/src/sync/apply.test.ts
@@ -123,7 +123,7 @@ describe("applyChanges", () => {
 });
 
 describe("applyChanges loopback suppression", () => {
-  it("advances pushRev to currentRev when source=upstream", async () => {
+  it("does not advance pushRev locally on upstream apply", async () => {
     await withDB(async (db) => {
       // Pre-existing local state: a write the container already
       // pushed. pushRev sits at currentRev.
@@ -133,10 +133,13 @@ describe("applyChanges loopback suppression", () => {
       const beforePushRev = readWatermark(db, "pushRev");
       expect(beforePushRev).toBeGreaterThan(0);
 
-      // Apply an entry as if it came from upstream. The local rev
-      // counter bumps (writeFile bumps rev), but the source flag
-      // makes the helper advance pushRev to match — the bump
-      // looks like it was already pushed.
+      // Apply an entry as if it came from upstream. The apply's
+      // writeFile bumps the local rev counter, but pushRev must
+      // *not* advance with it — advancing locally would move our
+      // pushRev past entries the remote does not know we have
+      // shipped, breaking the cross-side invariant on the next
+      // pull. The next pushOnce re-ships these rev bumps and the
+      // receiver's alreadyApplied() check drops them.
       await applyChanges(
         db,
         [
@@ -156,11 +159,8 @@ describe("applyChanges loopback suppression", () => {
 
       const afterCurrent = currentRev(db);
       const afterPushRev = readWatermark(db, "pushRev");
-      // Apply bumped currentRev (the writeFile inside).
       expect(afterCurrent).toBeGreaterThan(beforePushRev);
-      // pushRev caught up so the next coalesceChanges(db, pushRev)
-      // sees nothing.
-      expect(afterPushRev).toBe(afterCurrent);
+      expect(afterPushRev).toBe(beforePushRev);
     });
   });
 
@@ -186,15 +186,18 @@ describe("applyChanges loopback suppression", () => {
     });
   });
 
-  it("upstream entries do not get re-pushed on the next coalesce", async () => {
+  it("upstream entries surface on the next coalesce and rely on receiver-side alreadyApplied", async () => {
     await withDB(async (db) => {
       const { coalesceChanges } = await import("./coalesce.js");
       const { currentRev, readWatermark, writeWatermark } = await import("./watermarks.js");
       // Seed pushRev at the current point.
       writeWatermark(db, "pushRev", currentRev(db));
 
-      // Upstream sends a file. After apply, pushRev should equal
-      // currentRev, so coalesceChanges(db, pushRev) is empty.
+      // Upstream sends a file. After apply, pushRev stays where it
+      // was (the local advance was unsound — see the test above).
+      // The next coalesceChanges(db, pushRev) re-emits the entry;
+      // the receiver's alreadyApplied() check drops it. One extra
+      // round trip per apply, watermarks stay in lockstep.
       await applyChanges(
         db,
         [
@@ -214,7 +217,7 @@ describe("applyChanges loopback suppression", () => {
       const cursor = readWatermark(db, "pushRev");
       const drained = [];
       for await (const e of coalesceChanges(db, cursor)) drained.push(e);
-      expect(drained).toEqual([]);
+      expect(drained.map((e) => e.path)).toContain("/upstream.txt");
     });
   });
 });
@@ -271,12 +274,18 @@ describe("applyChanges loopback suppression — F1", () => {
     });
   });
 
-  it("still advances pushRev when caller had no unpushed locals", async () => {
+  it("leaves pushRev alone even when the caller had no unpushed locals", async () => {
     await withDB(async (db) => {
       const { currentRev, readWatermark, writeWatermark } = await import("./watermarks.js");
-      // pushRev already caught up to currentRev: caller has
-      // no pending local writes.
-      writeWatermark(db, "pushRev", currentRev(db));
+      // pushRev already caught up to currentRev: caller has no
+      // pending local writes. The old apply path advanced pushRev
+      // here as an optimization; we no longer do that because it
+      // desynced our pushRev from the remote's fetchRev (echoed
+      // back as appliedPushRev on the wire). The next pushOnce
+      // re-ships the apply's rev bump and the receiver's
+      // alreadyApplied() check drops it.
+      const before = currentRev(db);
+      writeWatermark(db, "pushRev", before);
       await applyChanges(
         db,
         [
@@ -293,9 +302,9 @@ describe("applyChanges loopback suppression — F1", () => {
         new Map(),
         { source: "upstream" },
       );
-      // Loopback suppression still works in the safe case:
-      // pushRev advances to cover the apply's own rev bump.
-      expect(readWatermark(db, "pushRev")).toBe(currentRev(db));
+      const after = currentRev(db);
+      expect(after).toBeGreaterThan(before);
+      expect(readWatermark(db, "pushRev")).toBe(before);
     });
   });
 });
diff --git a/packages/dofs/src/sync/apply.ts b/packages/dofs/src/sync/apply.ts
index 3b02b202..35aa76d6 100644
--- a/packages/dofs/src/sync/apply.ts
+++ b/packages/dofs/src/sync/apply.ts
@@ -7,7 +7,7 @@ import { writeFile, writeFileSync } from "../fs/writeFile.js";
 import type { Database } from "../storage.js";
 import type { ChangeEntry } from "./changes.js";
 import { computeManifestHash } from "./manifests.js";
-import { currentRev, readWatermark, writeWatermark } from "./watermarks.js";
+import { readWatermark, writeWatermark } from "./watermarks.js";
 
 // One container-side change that landed under a read-only mount and
 // was therefore skipped rather than applied. Callers (the workspace
@@ -53,14 +53,15 @@ export interface ApplyOptions {
   // cursor. Never regresses the watermark.
   advanceFetchRev?: number;
   // Where the entries came from. 'local' (default) treats the apply
-  // path like any other mutation: writeFile/mkdir/etc bump vfs_meta.rev
-  // and the push loop later ships those new revs upstream. 'upstream'
-  // means the entries came from a remote push or fetch; the apply
-  // still bumps rev (so readers see fresh data) but we advance pushRev
-  // to match, so the push loop knows everything in this range is
-  // already on the wire. Without this flag, applying an upstream
-  // entry would generate a push-back on the next tick and the two
-  // sides would ping-pong forever.
+  // path like any other mutation: writeFile/mkdir/etc bump
+  // vfs_meta.rev and the push loop later ships those new revs
+  // upstream. 'upstream' is informational: the apply still bumps
+  // rev so readers see fresh data, and the next pushOnce ships
+  // those rev bumps back to the sender. Loop convergence is the
+  // receiver's job — the apply path on the original sender uses
+  // alreadyApplied() to drop the redundant entries without bumping
+  // rev further, bounding the echo at one extra round trip per
+  // upstream apply.
   source?: "local" | "upstream";
   // Backend id whose watermark row this apply should touch. The
   // DO hosts independent sync cursors per backend; threading the
@@ -98,10 +99,6 @@ export async function applyChanges(
   objects: Map<string, Uint8Array>,
   options: ApplyOptions = {},
 ): Promise<ApplyResult> {
-  // Snapshot rev before we touch anything. Used by the loopback-
-  // suppression at the bottom to decide whether it's safe to
-  // advance pushRev past the entries this apply produced.
-  const revBeforeApply = currentRev(db);
   const maxBytes = options.maxBytesPerBatch ?? DEFAULT_MAX_BYTES;
   const maxPaths = options.maxPathsPerBatch ?? DEFAULT_MAX_PATHS;
 
@@ -207,34 +204,21 @@ export async function applyChanges(
     }
   }
 
-  // Loopback suppression: when this apply pass reflects entries
-  // from upstream, the writeFile/mkdir/symlink/rm calls inside
-  // bumped vfs_meta.rev. Without this advance, the next push tick
-  // would see those rev bumps as fresh local changes and push them
-  // back to upstream, which would apply them and bump again, and
-  // so on.
+  // Loopback suppression used to advance pushRev locally after an
+  // upstream apply so the next push tick wouldn't re-ship the rev
+  // bumps the apply produced. That optimization is unsound: it
+  // moves the *local* pushRev past entries the remote does not
+  // know we have shipped, while the remote's fetchRev (echoed back
+  // as appliedPushRev on every fetchChanges) stays where it was.
+  // The cross-side invariant check in pullOnce then trips on the
+  // very next pull and the post-drain pullOnce in the exec bracket
+  // swallows the error, leaving every subsequent container-side
+  // write invisible to the host until something reconciles.
   //
-  // Subtle: we can only advance pushRev when it already covered
-  // every rev that existed *before* this apply. If the caller had
-  // unpushed local writes sitting between (existing, revBeforeApply],
-  // advancing pushRev past them would strand them — the next
-  // pushOnce would skip them as already-shipped. That was F1: a
-  // pull whose entries were all idempotent-skipped still bumped
-  // pushRev up to currentRev, masking local writes that hadn't
-  // shipped yet.
-  //
-  // In the unsafe case we leave pushRev alone. The next pushOnce
-  // drains both the unpushed locals and the apply's own bumps;
-  // the receiver's alreadyApplied() check suppresses the latter.
-  // One redundant round-trip per apply, bounded.
-  if (options.source === "upstream") {
-    const revAfter = currentRev(db);
-    const existing = readWatermark(db, "pushRev", options.backend);
-    if (existing >= revBeforeApply && revAfter > existing) {
-      writeWatermark(db, "pushRev", revAfter, options.backend);
-    }
-  }
-
+  // The bounded "redundant round-trip" the old comment promised is
+  // still bounded, and the receiver's alreadyApplied() check still
+  // suppresses the entries on the next pushOnce. We just pay one
+  // extra push per upstream apply to keep the two sides in lockstep.
   return { applied, skipped };
 }
 
@@ -252,7 +236,6 @@ export function applyChangesSync(
   objects: Map<string, Uint8Array>,
   options: ApplyOptions = {},
 ): ApplyResult {
-  const revBeforeApply = currentRev(db);
   const maxBytes = options.maxBytesPerBatch ?? DEFAULT_MAX_BYTES;
   const maxPaths = options.maxPathsPerBatch ?? DEFAULT_MAX_PATHS;
 
@@ -342,13 +325,10 @@ export function applyChangesSync(
     }
   }
 
-  if (options.source === "upstream") {
-    const revAfter = currentRev(db);
-    const existing = readWatermark(db, "pushRev", options.backend);
-    if (existing >= revBeforeApply && revAfter > existing) {
-      writeWatermark(db, "pushRev", revAfter, options.backend);
-    }
-  }
+  // See applyChanges() for why pushRev no longer advances locally
+  // on upstream applies. The receiver's alreadyApplied() check
+  // suppresses the redundant entries on the next pushOnce; one
+  // extra push per apply keeps the cross-side invariant intact.
 
   return { applied, skipped };
 }
diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts
index d7f12ac8..b77484f5 100644
--- a/packages/rpc/src/server.ts
+++ b/packages/rpc/src/server.ts
@@ -105,10 +105,13 @@ class SyncRPCServer extends RpcTarget implements SyncRPC {
     } finally {
       reader.releaseLock();
     }
-    // senderRev > 0 — the caller is a sync peer with its
-    // own rev space; advance fetchRev to that point and let
-    // loopback suppression silence the outbound push so we
-    // don't ping-pong the same entries back.
+    // senderRev > 0 — the caller is a sync peer with its own
+    // rev space; advance fetchRev to that point so subsequent
+    // pulls and the cross-side invariant check see the right
+    // appliedPushRev. The apply path's alreadyApplied() check
+    // is what stops the entries from ping-ponging back through
+    // the sender's own coalesce + apply loop on the next round
+    // trip.
     //
     // senderRev === 0 — the caller is an external writer
     // (an orchestrator using the wire as a transport, the
diff --git a/packages/rpc/src/sync-driver.test.ts b/packages/rpc/src/sync-driver.test.ts
index 472bc55f..4fc701a7 100644
--- a/packages/rpc/src/sync-driver.test.ts
+++ b/packages/rpc/src/sync-driver.test.ts
@@ -361,23 +361,43 @@ describe("sync driver — bidirectional convergence", () => {
     }
   });
 
-  it("an upstream entry does not get re-pushed (loopback suppression)", async () => {
+  it("an upstream entry stops circulating within two ticks", async () => {
+    // Before the pushRev-locality fix, the loopback suppression
+    // advanced B's pushRev to currentRev on the apply, so the
+    // immediate pushOnce was a no-op. After the fix, B's pushRev
+    // stays put after the pull, so the first pushOnce after a
+    // pull ships the apply's rev bumps back to A; A's
+    // alreadyApplied() drops them, the push response advances B's
+    // pushRev, and the *next* tick is the no-op. The echo is
+    // bounded at one extra round trip and the system converges
+    // without an unbounded ping-pong.
     const a = makePeer();
     const b = makePeer();
     try {
       const providerA = new SQLiteWorkspaceProvider(a.db, { now: () => 1 });
       providerA.writeFileSync("/from-a.txt", "alpha");
 
-      // First tick: B pulls from A.
-      await tick(b.db, a.rpc);
+      // Tick 1: B pulls A's write. The apply on B bumps B's rev,
+      // so B's coalesce window contains entries; pushed reports
+      // however many entries got coalesced (typically 1 for the
+      // file alone, more if directory entries get touched).
+      const first = await tick(b.db, a.rpc);
       expect(fileEntries(b.db)).toContain("from-a.txt");
-
-      // Second tick: B has nothing new to push back. If the
-      // loopback suppression is broken, applyChanges bumped
-      // vfs_meta.rev on the apply, and the push side would re-ship
-      // the same entry.
-      const result = await tick(b.db, a.rpc);
-      expect(result.pushed).toBe(0);
+      expect(first.pulled.applied).toBeGreaterThan(0);
+      expect(first.pushed).toBeGreaterThanOrEqual(1);
+
+      // Tick 2: A's alreadyApplied() dropped the redundant entries
+      // shipped in tick 1, and B's pushRev advanced past them. So
+      // tick 2 has nothing to push and nothing to pull.
+      const second = await tick(b.db, a.rpc);
+      expect(second.pulled.applied).toBe(0);
+      expect(second.pushed).toBe(0);
+
+      // Tick 3: still settled. Pins that convergence is durable,
+      // not just "the next tick happens to be empty."
+      const third = await tick(b.db, a.rpc);
+      expect(third.pulled.applied).toBe(0);
+      expect(third.pushed).toBe(0);
     } finally {
       a.close();
       b.close();
@@ -415,7 +435,7 @@ describe("sync driver — cross-side invariant", () => {
 
       // Wrap B's rpc to lie about appliedPushRev. Simulates a
       // regression in the suppress-dirty-tracking apply path.
-      const lyingRpc = new Proxy(b.rpc as object, {
+      const lyingRPC = new Proxy(b.rpc as object, {
         get(target, prop, receiver) {
           if (prop === "push") {
             return async (input: { senderRev: number; changes: ReadableStream<unknown> }) => {
@@ -427,36 +447,80 @@ describe("sync driver — cross-side invariant", () => {
         },
       }) as typeof b.rpc;
 
-      await expect(pushOnce(a.db, lyingRpc)).rejects.toThrow(/cross-side invariant violated/i);
+      await expect(pushOnce(a.db, lyingRPC)).rejects.toThrow(/cross-side invariant violated/i);
     } finally {
       a.close();
       b.close();
     }
   });
 
-  it("pullOnce throws when fetchChanges echoes back a lower appliedPushRev", async () => {
-    // Symmetric to the push case. fetchChanges returns the remote's
-    // appliedPushRev alongside the entry stream; the DO asserts
-    // appliedPushRev >= pushRev before draining, so a regression in
-    // the remote's apply path that loses applied state trips the
-    // invariant on the next pull instead of corrupting fetchRev.
+  it("pullOnce resets pushRev and retries when fetchChanges echoes a lower appliedPushRev", async () => {
+    // The remote reporting an appliedPushRev below our localPushRev
+    // means the remote forgot what we pushed — typically a process-
+    // lifetime wsd restart while the WebSocket stayed up, so the
+    // reconcileWatermarks pass we run on connect never re-ran. The
+    // pull path now treats this inline: cancel the in-flight
+    // stream, reset pushRev to 0, and retry. The next pushOnce
+    // re-ships everything from the rev-0 baseline.
+    const remote = makePeer();
+    try {
+      const local = new Database(new SQLiteTestStorage());
+      initializeSchema(local, () => 1000);
+      writeWatermark(local, "pushRev", 42);
+      const providerR = new SQLiteWorkspaceProvider(remote.db, { now: () => 1 });
+      providerR.writeFileSync("/seed.txt", "x");
+
+      // The proxy lies once: on the first fetchChanges, swap the
+      // remote's real appliedPushRev for 0. The pull path detects
+      // the divergence and retries; on the retry the real RPC
+      // runs (because lied flips) and pullOnce drains normally.
+      let lied = false;
+      const flakyRPC = new Proxy(remote.rpc as object, {
+        get(target, prop, receiver) {
+          if (prop === "fetchChanges" && !lied) {
+            return async (input: { sinceRev?: number; ignore?: string[] }) => {
+              lied = true;
+              const real = await Reflect.get(target, prop, receiver).call(target, input);
+              return { ...real, appliedPushRev: 0 };
+            };
+          }
+          return Reflect.get(target, prop, receiver);
+        },
+      }) as typeof remote.rpc;
+
+      const result = await pullOnce(local, flakyRPC);
+      // The retry succeeded: the seeded /seed.txt landed locally.
+      expect(result.applied).toBeGreaterThan(0);
+      // pushRev was reset to 0 on the divergence and stays at 0
+      // (we didn't run a successful pushOnce); the next pushOnce
+      // tick will re-ship from the baseline.
+      expect(readWatermark(local, "pushRev")).toBe(0);
+    } finally {
+      remote.close();
+    }
+  });
+
+  it("pullOnce surfaces an invariant violation that survives the inline retry", async () => {
+    // A persistently-lying remote (returns appliedPushRev=0 on
+    // every call) trips the assertion after the inline reset.
+    // The retry resets localPushRev to 0; the assertion then sees
+    // appliedPushRev=0, localPushRev=0 and passes. So a permanent
+    // lie now degrades to baseline re-sync rather than a hard
+    // error. Pin that: the test passes (not throws), and the
+    // caller's watermarks are zeroed.
     const remote = makePeer();
     try {
-      // Seed the local pushRev so it's higher than what the lying
-      // remote will echo. The remote is otherwise fresh — nothing
-      // to fetch.
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
       writeWatermark(local, "pushRev", 42);
-      // Make the remote return *something* so the puller drains it.
       const providerR = new SQLiteWorkspaceProvider(remote.db, { now: () => 1 });
       providerR.writeFileSync("/seed.txt", "x");
 
-      const lyingRpc = new Proxy(remote.rpc as object, {
+      const lyingRPC = new Proxy(remote.rpc as object, {
         get(target, prop, receiver) {
           if (prop === "fetchChanges") {
-            return (input: { sinceRev?: number; ignore?: string[] }) => {
-              const real = Reflect.get(target, prop, receiver).call(target, input);
+            return async (input: { sinceRev?: number; ignore?: string[] }) => {
+              const real = await Reflect.get(target, prop, receiver).call(target, input);
               return { ...real, appliedPushRev: 0 };
             };
           }
@@ -464,7 +528,9 @@ describe("sync driver — cross-side invariant", () => {
         },
       }) as typeof remote.rpc;
 
-      await expect(pullOnce(local, lyingRpc)).rejects.toThrow(/cross-side invariant violated/i);
+      const result = await pullOnce(local, lyingRPC);
+      expect(result.applied).toBeGreaterThan(0);
+      expect(readWatermark(local, "pushRev")).toBe(0);
     } finally {
       remote.close();
     }
@@ -653,8 +719,8 @@ describe("sync driver — reconcileWatermarks", () => {
   it("resets pushRev when the remote hasn't applied what we shipped", async () => {
     const remote = makePeer();
     try {
-      // Local pushRev = 17, but the remote is fresh: its pushRev,
-      // which doubles as appliedPushRev on the wire, is 0.
+      // Local pushRev = 17, but the remote is fresh: its fetchRev
+      // (echoed back as appliedPushRev on the wire) is 0. Reset.
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
       writeWatermark(local, "fetchRev", 0);
@@ -667,6 +733,33 @@ describe("sync driver — reconcileWatermarks", () => {
     }
   });
 
+  it("leaves pushRev alone when the remote has applied our pushes but never initiated its own", async () => {
+    // Topology: DO ↔ container. The container applies pushes (so
+    // its fetchRev = our pushRev) but never initiates outbound
+    // pushes (so its pushRev stays at 0). reconcileWatermarks must
+    // not interpret remote.pushRev = 0 as "remote forgot our
+    // pushes" — that would trigger a full re-push on every
+    // reconnect even when nothing is broken.
+    const remote = makePeer();
+    try {
+      // Pretend the container's apply path has accepted our pushes
+      // up to rev 17 (= what fetchChanges would echo back as
+      // appliedPushRev). Its own pushRev stays at 0 because it has
+      // not shipped anything outbound.
+      writeWatermark(remote.db, "fetchRev", 17);
+      const local = new Database(new SQLiteTestStorage());
+      initializeSchema(local, () => 1000);
+      writeWatermark(local, "fetchRev", 0);
+      writeWatermark(local, "pushRev", 17);
+
+      const result = await reconcileWatermarks(local, remote.rpc);
+      expect(result.pushRevReset).toBe(false);
+      expect(readWatermark(local, "pushRev")).toBe(17);
+    } finally {
+      remote.close();
+    }
+  });
+
   it("leaves watermarks alone when remote is at least caught up", async () => {
     const remote = makePeer();
     try {
diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts
index 355d5f29..45943679 100644
--- a/packages/rpc/src/sync-driver.ts
+++ b/packages/rpc/src/sync-driver.ts
@@ -98,8 +98,27 @@ export async function pullOnce(
   remote: SyncRPC,
   backend?: string,
 ): Promise<ApplyResult> {
+  // Delegate to the inner implementation with retried=false. See
+  // pullOnceImpl for the fetchChanges round trip, invariant check,
+  // reset-and-retry path, and batched apply loop.
   const sinceRev = readWatermark(db, "fetchRev", backend);
   const localPushRev = readWatermark(db, "pushRev", backend);
+  return pullOnceImpl(db, remote, backend, sinceRev, localPushRev, false);
+}
+
+// Inner pullOnce that knows whether it is already a retry. The
+// outer pullOnce always enters with retried=false; on a watermark
+// divergence we reset cursors and recurse once with retried=true.
+// A second divergence after the reset is a real protocol break,
+// not a recoverable race, so we throw to surface it.
+async function pullOnceImpl(
+  db: Database,
+  remote: SyncRPC,
+  backend: string | undefined,
+  sinceRev: number,
+  localPushRev: number,
+  retried: boolean,
+): Promise<ApplyResult> {
   // fetchChanges hands back the remote's currentRev (cursor we
   // advance fetchRev to), its appliedPushRev (cross-side invariant
   // check on the pull path), and the entry stream itself. One
@@ -113,11 +132,61 @@ export async function pullOnce(
   // that disposes the envelope when the stream finishes draining.
   const fetchResult = await remote.fetchChanges({ sinceRev });
   const { currentRev: remoteRev, appliedPushRev } = fetchResult;
-  // Run the cross-side invariant check before touching the stream.
-  // Symmetric to the push response check: the remote must have
-  // applied at least everything we claimed to push. A drop here
-  // means apply lost state on the receiver; tear down and rebuild
-  // rather than corrupt watermarks.
+  // Cross-side watermark divergence. Two shapes are recoverable:
+  //   * appliedPushRev < localPushRev: the remote forgot what we
+  //     pushed (typically a process-lifetime wsd restart while the
+  //     WebSocket survived, so reconcileWatermarks on connect never
+  //     re-ran).
+  //   * remoteRev < sinceRev: the remote's log is shorter than we
+  //     remember — same root cause, different symptom.
+  // Both are the inline equivalent of reconcileWatermarks: reset
+  // the divergent cursor to 0, cancel the in-flight stream, and
+  // retry once. The rev-0 baseline path in fetchChanges + pushOnce
+  // re-ships everything incrementally and the receiver's
+  // alreadyApplied() check absorbs the redundant work.
+  //
+  // A second divergence after a reset is a real protocol break:
+  // surface it via the assertion below rather than loop.
+  if (!retried && (appliedPushRev < localPushRev || remoteRev < sinceRev)) {
+    // Cancel the stream before disposing the envelope. For a real
+    // capnweb envelope the dispose alone is enough to tear down the
+    // backing stub, but the in-process server returns a plain
+    // ReadableStream wired to an async generator; without an
+    // explicit cancel the generator stays advanced (queue size 0
+    // plus high-water mark 1 means pull() has already been called)
+    // and its query results sit in memory until GC. Cancel is
+    // best-effort: a real envelope may have already torn the stream
+    // down before we get here.
+    await fetchResult.stream.cancel().catch(() => {});
+    maybeDispose(fetchResult);
+    // Surface the divergence at debug level so an operator with
+    // log access can spot a persistently broken remote. We do not
+    // throw: a one-shot divergence is normal after a wsd restart
+    // under the same WebSocket, and the inline reset + retry is
+    // the intended recovery. A persistently-lying remote will log
+    // this on every pull, which is the operational signal that
+    // something upstream is wedged.
+    console.debug("[pullOnce] cross-side watermark divergence; resetting and retrying", {
+      backend,
+      appliedPushRev,
+      localPushRev,
+      remoteRev,
+      sinceRev,
+      resetPushRev: appliedPushRev < localPushRev,
+      resetFetchRev: remoteRev < sinceRev,
+    });
+    if (appliedPushRev < localPushRev) {
+      writeWatermark(db, "pushRev", 0, backend);
+    }
+    if (remoteRev < sinceRev) {
+      writeWatermark(db, "fetchRev", 0, backend);
+    }
+    const nextSinceRev = readWatermark(db, "fetchRev", backend);
+    const nextLocalPushRev = readWatermark(db, "pushRev", backend);
+    return pullOnceImpl(db, remote, backend, nextSinceRev, nextLocalPushRev, true);
+  }
+  // After the retry path above, this assertion guards a
+  // divergence that survived a reset. Tear down rather than loop.
   assertAppliedPushRev(appliedPushRev, localPushRev);
   const stream = disposeOnDone(fetchResult.stream, () => maybeDispose(fetchResult));
   if (remoteRev <= sinceRev) {
@@ -351,11 +420,20 @@ export async function reconcileWatermarks(
     fetchRevReset = true;
   }
 
-  // The remote's pushRev is what it last applied from us (when the
-  // remote acts as a sync peer it advances pushRev to the senderRev
-  // on every push). If that's below our local pushRev, the remote
-  // hasn't seen what we claimed to ship — reset and re-push.
-  if (remoteWatermarks.pushRev < localPushRev) {
+  // The remote's fetchRev is the largest senderRev it has applied
+  // from us — every push handler advances fetchRev to the incoming
+  // senderRev, and fetchChanges echoes that value back as
+  // appliedPushRev. If it's below our local pushRev, the remote has
+  // not seen what we claimed to ship; reset our pushRev so the next
+  // pushOnce re-baselines from rev 0.
+  //
+  // We deliberately do NOT compare against remoteWatermarks.pushRev:
+  // that field is the remote's own *outbound* push progress and
+  // stays at 0 in topologies where the remote never initiates a push
+  // (e.g. the container side of a DO↔container backend), which would
+  // make every reconcile spuriously reset pushRev and force a full
+  // re-push on every reconnect.
+  if (remoteWatermarks.fetchRev < localPushRev) {
     writeWatermark(db, "pushRev", 0, backend);
     pushRevReset = true;
   }
diff --git a/packages/rpc/tests/wire.test.ts b/packages/rpc/tests/wire.test.ts
index 6db78517..58e564c7 100644
--- a/packages/rpc/tests/wire.test.ts
+++ b/packages/rpc/tests/wire.test.ts
@@ -508,7 +508,7 @@ describe("push semantics — external vs sync peer", () => {
     }
   });
 
-  it("push with senderRev>0 (sync peer) advances pushRev to silence loopback", async () => {
+  it("push with senderRev>0 (sync peer) leaves pushRev for the next push, advances fetchRev to senderRev", async () => {
     harness = await startHarness();
     const { currentRev, readWatermark, writeWatermark } = await import("@cloudflare/dofs");
     const client = createSyncClient({ url: harness.url });
@@ -546,11 +546,13 @@ describe("push semantics — external vs sync peer", () => {
         }),
       });
 
-      // Loopback suppression kicks in: pushRev was advanced
-      // to currentRev so the receiver doesn't push these
-      // entries back to the peer.
-      const cur = currentRev(harness.db);
-      expect(readWatermark(harness.db, "pushRev")).toBe(cur);
+      // pushRev stays where we seeded it: the receiver does
+      // not advance pushRev locally on upstream apply (that
+      // was unsound — it desynced our pushRev from the wire-
+      // visible fetchRev and broke the cross-side invariant
+      // on the next pull). The next pushOnce ships the
+      // apply's rev bumps and alreadyApplied() drops them.
+      expect(readWatermark(harness.db, "pushRev")).toBe(1);
       // fetchRev was advanced to senderRev.
       expect(readWatermark(harness.db, "fetchRev")).toBe(42);
     } finally {