fix(localdb): a rebuilt peer's own writes were silently dropped

The other half of the rebuilt-peer story, and only reachable once 0.1.2 made
back-fill work: with the wiped node repopulated and writing again, its writes
never reached its peer.

last_applied_remote_seq is a watermark in the PEER's seq space, and a rebuilt
peer is a new database whose oplog numbers from 1. Two things then went wrong,
and both had to be fixed — the first alone leaves data stuck:

- The healthy node kept advertising the OLD peer's watermark, and the sending
  side seeds its pump from exactly that number, so the rebuilt node skipped its
  entire oplog. The watermark (and the observed peer clock) is now reset when
  the peer's node id changes. last_acked_seq is deliberately NOT reset: it
  describes our own oplog's pruned horizon and is what tells a rebuilt peer it
  needs a snapshot — clearing it would turn 0.1.2's back-fill back off.

- A peer claiming to have applied more of our stream than we have ever produced
  can only be remembering a previous incarnation of us. That claim is now
  rejected in favour of starting from the beginning; LWW makes the re-send
  harmless. This mirrors the clamp RecordPeerAckAsync already applies to acks.

Found on the OtOpcUa docker-dev rig, where the healthy node held watermark 10
while the rebuilt peer's oplog was seqs 1-3 and its last_acked stayed 0 — the
peer never accepted a single one. Covered by a test that converges with one
peer, replaces it with a fresh database, and requires a write on the new peer
to arrive: RED before this fix, green after. 149/149 pass.

Version 0.1.3.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs

@@ -142,11 +142,38 @@ internal sealed class SyncSession
 
         ValidateHandshake(localHandshake, peerHandshake);
 
+        // A different node id on the far end means the peer was replaced, not merely restarted — a
+        // rebuilt host, a restored-from-nothing volume. Its oplog numbers from 1 again, so the
+        // watermark we hold (in the OLD peer's seq space) would silently swallow its first N
+        // writes. Forget it before anything reads it.
+        if (peerState.PeerNodeId is not null && peerState.PeerNodeId != peerHandshake.NodeId)
+        {
+            _logger.LogWarning(
+                "Replication peer identity changed ({OldPeer} -> {NewPeer}): the peer was rebuilt, so its seq space restarted. Resetting the inbound watermark.",
+                peerState.PeerNodeId, peerHandshake.NodeId);
+            await _store.ResetInboundWatermarkAsync(ct);
+            peerState = peerState with { LastAppliedRemoteSeq = 0, LastSeenHlc = 0 };
+        }
+
         await _store.SetPeerNodeIdAsync(peerHandshake.NodeId, ct);
         PeerNodeId = peerHandshake.NodeId;
         _status?.SetPeerNodeId(peerHandshake.NodeId);
-        _sentThruSeq = peerHandshake.LastAppliedRemoteSeq;
-        Interlocked.Exchange(ref _peerAckedSeq, peerHandshake.LastAppliedRemoteSeq);
+        // The peer's watermark decides where our pump starts, so an impossible claim silently
+        // skips real data. It can only exceed what we have ever produced if the peer is remembering
+        // a previous incarnation of THIS node — i.e. WE were rebuilt and our seqs restarted at 1 —
+        // in which case its number says nothing about the stream we hold now, and starting from it
+        // would skip our entire oplog. Fall back to the beginning; LWW makes re-sending harmless.
+        var knownMaxSeq = await _store.GetKnownMaxSeqAsync(ct);
+        var peerAppliedThruSeq = peerHandshake.LastAppliedRemoteSeq <= knownMaxSeq
+            ? peerHandshake.LastAppliedRemoteSeq
+            : 0;
+        if (peerAppliedThruSeq != peerHandshake.LastAppliedRemoteSeq)
+            _logger.LogWarning(
+                "Peer claims to have applied our seq {PeerClaim}, beyond the {KnownMax} we have ever produced — this node was rebuilt. Re-sending our oplog from the start.",
+                peerHandshake.LastAppliedRemoteSeq, knownMaxSeq);
+
+        _sentThruSeq = peerAppliedThruSeq;
+        Interlocked.Exchange(ref _peerAckedSeq, peerAppliedThruSeq);
 
         var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct);
         await controlLane.WriteAsync(