docs(index): LocalDb consumer pins — OtOpcUa 0.1.3 -> 0.2.1 (R4 merged), skew note updated for 0.3.0

dotnet pack with no project argument packs whatever solution sits in the caller's
CWD, so invoking this script by absolute path from another repo's checkout packed
and published 25 unrelated packages to the live Gitea feed under its name. Now it
cds to its own repo root and names the solution and an absolute output path.

ZB.MOM.WW.LocalDb/Directory.Build.props

@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
-    <Version>0.2.1</Version>
+    <Version>0.3.0</Version>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a

ZB.MOM.WW.LocalDb/README.md

@@ -10,7 +10,7 @@ Change capture uses SQLite triggers writing a single HLC-stamped oplog inside th
 consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
 last-writer-wins over a hybrid logical clock. All timestamps are UTC.
 
-## Packages (all `0.2.1`)
+## Packages (all `0.3.0`)
 
 | Package | Contents |
 |---|---|
@@ -20,6 +20,30 @@ last-writer-wins over a hybrid logical clock. All timestamps are UTC.
 
 > Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
 >
+> - `0.3.0` (2026-08-15) — **the snapshot-resync flag is per-table.** On-disk bookkeeping schema
+>   goes to **v3** (new `__localdb_snapshot_state`, upgraded in place on open); wire-compatible with
+>   `0.1.x`/`0.2.x` peers, and `lib_schema_version` deliberately stays `1` because it is compared
+>   fail-closed for equality.
+>   - **`RegisterReplicated(table, baselineExistingRows: true)` now flags a resync for THAT TABLE.**
+>     Through `0.2.x` the flag was one bit for the whole database, so baselining one table
+>     re-streamed every registered table in full — and on a pair enabling replication late, both
+>     nodes did that to each other at once. A snapshot now streams only the flagged tables.
+>   - **The database-wide flag survives, unchanged, for the cases that genuinely need it**: an oplog
+>     cap prune or an impossible peer ack loses deltas for the oplog as a whole, so nothing short of
+>     every table is sound. What moved is only where a *baseline* records its debt.
+>   - **Additive wire extension, negotiated by capability, not by version.** `Handshake` gains
+>     `supports_partial_snapshot`; `SnapshotBegin` gains a `tables` scope where **empty means "all
+>     tables"** — which is precisely what a pre-`0.3.0` sender emits and what a pre-`0.3.0` receiver
+>     assumes. A peer that does not advertise the capability is served a **full** snapshot, because a
+>     scoped one would make it advance `last_applied_remote_seq` past deltas it never received.
+>   - **A partial snapshot moves no watermark on either side.** The receiver keeps
+>     `last_applied_remote_seq` (it is one number for the whole peer oplog, and the untouched tables'
+>     deltas still sit below the snapshot's as-of); the sender keeps pumping from where it was. The
+>     overlap re-delivers rows the snapshot already carried, which LWW absorbs.
+>   - **Snapshot debt now clears on acknowledgement, not on enqueue** (new `SnapshotAck`, exchanged
+>     only between `0.3.0` peers). A crash part-way through a snapshot re-streams the same scope on
+>     reconnect instead of losing it. Against a pre-`0.3.0` peer the `0.2.x` clear-on-enqueue
+>     behaviour is kept — there is no ack to wait for.
 > - `0.2.1` (2026-08-14) — two adversarial-review findings against `0.2.0`. Wire-compatible with
 >   `0.1.x`/`0.2.0` peers; no proto change, no `lib_schema_version` bump, no on-disk schema change.
 >   - **`DeregisterReplicated` no longer leaves the HLC without a crash anchor.** Clock recovery on
@@ -177,10 +201,15 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
   delta/snapshot batches), so receiving never blocks on the ability to ack.
 - **Resync** — when a peer's watermark falls below the pruned oplog horizon, it is sent a
   full snapshot of the registered tables (streamed, keyset-paged) and deltas resume above
-  the snapshot's as-of seq. **Each node decides from its OWN state** whether it owes the
-  peer a snapshot. The snapshot is streamed *after* the handshake, alongside a running
-  receive loop, so a node sending one still applies what the peer sends it — a mutual
-  snapshot exchanges under flow control instead of being buffered whole in memory.
+  the snapshot's as-of seq. When instead a single table was *baselined*, only that table is
+  streamed and **no watermark moves on either side** — a partial snapshot speaks for nothing
+  outside its own tables, so the deltas it overlaps are still pumped and merged by LWW.
+  **Each node decides from its OWN state** whether it owes the peer a snapshot, and a peer
+  that does not advertise `supports_partial_snapshot` is sent a full one. The snapshot is
+  streamed *after* the handshake, alongside a running receive loop, so a node sending one
+  still applies what the peer sends it — a mutual snapshot exchanges under flow control
+  instead of being buffered whole in memory. Between `0.3.0` peers the sender's flags are
+  cleared by the receiver's `SnapshotAck`, so a crash mid-snapshot re-streams it.
 - **Maintenance** — a background service on **both** roles enforces the oplog row/age caps
   and prunes expired tombstones every `MaintenanceInterval`.
 
@@ -209,17 +238,18 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
 - **CDC semantics** — rows that existed **before** `RegisterReplicated` are not captured
   and are not snapshotted; only changes after registration are replicated. **Opt out with
   `RegisterReplicated(table, baselineExistingRows: true)`**, which seeds the row-version
-  ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync so
-  they reach the peer. The floor is below every genuine stamp, so a baselined row loses
+  ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync
+  **for that table** so they reach the peer. The floor is below every genuine stamp, so a baselined row loses
   every comparison against a real remote write and wins only where the peer has no version
   of that key at all. Two nodes that baseline the same key both hold HLC `0` and resolve by
   the node-id tie-break — convergent, but arbitrary as to which content survives.
-- **The snapshot flag is per-database, not per-table.** Baselining one table flags a resync
-  that streams **every** registered table in full, and on a pair enabling replication late
-  both nodes do that to each other simultaneously — the largest transfer this library
-  performs. Bounded (batch budget out, bounded inbox in), but **purge or trim high-volume
-  tables such as event logs before enabling replication late**, and expect one slow first
-  sync. A per-table flag needs an on-disk schema change and is not in `0.2.x`.
+- **A snapshot still ships whole tables.** The flag is per-table as of `0.3.0`, so baselining
+  one table streams that table and no other — but *that* table travels in full, and two
+  cases still stream **everything**: an oplog cap prune / impossible peer ack (delta
+  continuity is lost for the oplog as a whole), and any peer older than `0.3.0` (which
+  cannot be told a snapshot is partial). So **purge or trim high-volume tables such as event
+  logs before baselining them**, and expect one slow first sync. Bounded either way (batch
+  budget out, bounded inbox in).
 - **REAL ±Infinity** is carried in SQLite's JSON dialect on the wire (fine for the
   SQLite↔SQLite peers this library targets).
 - `AddZbLocalDb` / `AddZbLocalDbReplication` are **first-registration-wins** (idempotent).
@@ -256,8 +286,15 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 
 ## Ops notes
 
-- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags
-  `needs_snapshot` (disk safety over delta continuity); the peer then snapshot-resyncs.
+- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags the
+  **database-wide** `needs_snapshot` (disk safety over delta continuity); the peer then gets a
+  full snapshot. Per-table debt lives separately in `__localdb_snapshot_state` and is what
+  baselining writes.
+- **Upgrading a pair to `0.3.0`** needs no coordination: the two nodes negotiate scoped snapshots
+  by capability, so a `0.3.0` node paired with a `0.2.x` one simply keeps sending full snapshots
+  until both are upgraded. Rolling the file forward is one-way — a `0.2.x` build reopening a v3
+  file leaves `__localdb_snapshot_state` in place and ignores it, so any per-table debt recorded
+  by `0.3.0` is invisible to it (roll forward again, or re-baseline).
 - **Turning a table off** — `DeregisterReplicated(table)` drops its capture triggers and prunes
   its oplog + ledger rows. It must be applied **symmetrically**: the handshake compares
   registered-table digests fail-closed, so a node that drops a table its peer still
@@ -281,14 +318,16 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 
 ## Testing
 
-188 tests, all offline / no live dependencies — unit (HLC ordering + restart
+199 tests, all offline / no live dependencies — unit (HLC ordering + restart
 persistence, deregistration's HLC anchor surviving an ungraceful reopen, trigger SQL
 generation, LWW decision table, watermark/pruning, digest computation, schema-version
 upgrade, deregistration idempotency, baseline-floor LWW safety, O(1) depth vs a seeded
 backlog, byte-budget chunking, bounded-inbox backpressure under a slow reader), integration
 over two real SQLite DBs on an in-memory (and real) gRPC pipe (bidirectional convergence,
 delete/update races, offline-accumulate → reconnect, prune → snapshot resync, mutual
-snapshot over a capacity-1 transport), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
+snapshot over a capacity-1 transport, single-table baseline streaming only that table,
+mixed-version interop where a peer without `supports_partial_snapshot` gets a full one,
+snapshot debt surviving until the peer's `SnapshotAck`), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
 Family conventions apply (`TreatWarningsAsErrors`).
 
 Build / test from this directory:

ZB.MOM.WW.LocalDb/build/push.sh

@@ -16,9 +16,19 @@ set -euo pipefail
 : "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
 : "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"
 
-dotnet pack -c Release -o ./artifacts
+# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
+# with neither a project argument nor a fixed working directory packs whatever solution
+# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
+# script's name. Invoked as `bash /path/to/ZB.MOM.WW.LocalDb/build/push.sh` from another
+# repo's checkout, the relative form packed and published 25 unrelated packages (recovered
+# with `dotnet nuget delete`, but only because the token happened to carry delete rights).
+# A publish script must be unambiguous about WHAT it publishes no matter where it is run.
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
 
-dotnet nuget push "./artifacts/*.nupkg" \
+dotnet pack ZB.MOM.WW.LocalDb.slnx -c Release -o "$REPO_ROOT/artifacts"
+
+dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
   --source "$GITEA_NUGET_SOURCE" \
   --api-key "$GITEA_NUGET_KEY" \
   --skip-duplicate

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Contracts/Protos/localdb_sync.proto

@@ -16,6 +16,7 @@ message SyncMessage {
     SnapshotBegin snapshot_begin = 5; // sender switches to snapshot mode
     SnapshotBatch snapshot_batch = 6;
     SnapshotComplete snapshot_complete = 7;
+    SnapshotAck snapshot_ack = 8;     // 0.3.0+; only sent when the peer advertised support
   }
 }
 
@@ -24,6 +25,13 @@ message Handshake {
   uint32 lib_schema_version = 2;
   repeated TableDigest tables = 3;    // fail-closed on any mismatch
   int64 last_applied_remote_seq = 4;  // highest of YOUR oplog seqs I have applied
+  // 0.3.0+ capability, additive: I understand a snapshot scoped to a SUBSET of the registered
+  // tables (SnapshotBegin.tables) and I will answer a completed snapshot with a SnapshotAck.
+  // A pre-0.3.0 peer leaves this false — proto3 default — and a 0.3.0 sender then promotes any
+  // scoped snapshot to a full one and clears its flags on enqueue, exactly as 0.2.x did. This is
+  // why lib_schema_version stays 1: it is compared fail-closed for EQUALITY, so bumping it would
+  // stop 0.1.x/0.2.x pairs syncing outright.
+  bool supports_partial_snapshot = 5;
 }
 message TableDigest { string table_name = 1; string digest = 2; }
 message HandshakeAck { string node_id = 1; bool snapshot_required = 2; }
@@ -40,7 +48,12 @@ message OplogEntry {
 message DeltaBatch { repeated OplogEntry entries = 1; }
 message DeltaAck { int64 applied_thru_seq = 1; }
 
-message SnapshotBegin { int64 as_of_seq = 1; }
+// tables: the SCOPE of this snapshot (0.3.0+). EMPTY means every registered table — a full
+// snapshot, which is the only thing a pre-0.3.0 sender ever emits and the only thing a
+// pre-0.3.0 receiver can safely assume. A non-empty list means the snapshot is PARTIAL, so the
+// receiver must NOT advance last_applied_remote_seq to as_of_seq (deltas below it for the tables
+// NOT listed have not been delivered); it merges the rows through LWW and leaves the watermark.
+message SnapshotBegin { int64 as_of_seq = 1; repeated string tables = 2; }
 message SnapshotBatch { string table_name = 1; repeated SnapshotRow rows = 2; }
 message SnapshotRow {
   string pk_json = 1;
@@ -50,3 +63,8 @@ message SnapshotRow {
   bool is_tombstone = 5;
 }
 message SnapshotComplete { int64 as_of_seq = 1; }
+// Sent by the RECEIVER once a snapshot has been durably applied, so the sender clears the
+// per-table snapshot flags only after the peer actually holds the rows (a crash mid-snapshot then
+// re-streams the same scope on reconnect). Only exchanged when both peers advertised
+// supports_partial_snapshot; a pre-0.3.0 pair keeps the 0.2.x clear-on-enqueue behaviour.
+message SnapshotAck { int64 as_of_seq = 1; }

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

@@ -233,10 +233,61 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
         db.ExecuteAsync(
             "UPDATE __localdb_peer_state SET last_applied_remote_seq = 0, last_seen_hlc = 0 WHERE id = 1", null, ct);
 
-    /// <summary>Sets or clears the snapshot-resync-required flag.</summary>
+    /// <summary>
+    /// Sets or clears the DATABASE-WIDE snapshot-resync flag — the one that means "every registered
+    /// table". Set by the oplog-horizon paths (cap prune, impossible peer ack), which invalidate
+    /// delta continuity for all tables at once. Per-table flags live in
+    /// <c>__localdb_snapshot_state</c>; see <see cref="GetTablesNeedingSnapshotAsync"/>.
+    /// </summary>
     public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) =>
         db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct);
 
+    /// <summary>
+    /// The tables individually flagged as owing the peer a snapshot (schema v3), in ordinal name
+    /// order. Independent of the database-wide flag: baselining one table flags only that table.
+    /// </summary>
+    public Task<IReadOnlyList<string>> GetTablesNeedingSnapshotAsync(CancellationToken ct = default) =>
+        db.QueryAsync(
+            "SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
+            static r => r.GetString(0), null, ct);
+
+    /// <summary>
+    /// Clears the snapshot debt a completed snapshot discharged: the named tables when
+    /// <paramref name="tables"/> is non-empty, otherwise <b>everything</b> — the database-wide flag
+    /// and every per-table row, because a full snapshot subsumes each of them.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// Rows are deleted rather than zeroed: absence is the resting state, and keeping a row per
+    /// table that has ever been baselined would accumulate bookkeeping that means nothing. Runs in
+    /// one transaction so a full snapshot's two clears cannot half-apply and leave the pair
+    /// re-snapshotting on every reconnect.
+    /// </para>
+    /// <para>
+    /// A scoped clear deliberately does NOT touch the database-wide flag. If the cap enforcer
+    /// flagged a whole-database resync while this snapshot was in flight, that debt is real and
+    /// unrelated — the snapshot only carried some of the tables — so it must survive. The full
+    /// clear may take it down because a full snapshot carries every row state as of its as-of seq,
+    /// which is exactly what the pruned deltas would have delivered.
+    /// </para>
+    /// </remarks>
+    public async Task ClearSnapshotDebtAsync(IReadOnlyList<string> tables, CancellationToken ct = default)
+    {
+        await using var tx = await db.BeginTransactionAsync(ct);
+        if (tables.Count == 0)
+        {
+            await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1", null, ct);
+            await tx.ExecuteAsync("DELETE FROM __localdb_snapshot_state", null, ct);
+        }
+        else
+        {
+            foreach (var table in tables)
+                await tx.ExecuteAsync(
+                    "DELETE FROM __localdb_snapshot_state WHERE table_name = @t", new { t = table }, ct);
+        }
+        await tx.CommitAsync(ct);
+    }
+
     /// <summary>
     /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps
     /// <c>last_sync_utc</c>. Both watermarks are MAX-guarded (monotonic): a snapshot completing with a

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

@@ -19,21 +19,57 @@ internal interface ISnapshotApplier
 }
 
 /// <summary>
-/// Streams a full snapshot of every registered table to a peer that has fallen behind the pruned
-/// oplog horizon. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
+/// Which tables a snapshot covers. An EMPTY <see cref="Tables"/> is the wire's own encoding of
+/// "all registered tables" (see <c>SnapshotBegin.tables</c>) and is what every pre-0.3.0 peer both
+/// sends and expects, so it is deliberately the same value here rather than a separate flag.
+/// </summary>
+/// <remarks>
+/// The distinction is not cosmetic: a FULL snapshot carries every row state as of its
+/// <c>as_of_seq</c>, so the receiver may jump its watermark there and the sender may skip the
+/// deltas it covered. A PARTIAL one carries only the listed tables, so neither side may move a
+/// watermark that also speaks for the tables it left out.
+/// </remarks>
+internal sealed record SnapshotScope(IReadOnlyList<string> Tables)
+{
+    /// <summary>Every registered table — the only scope a pre-0.3.0 peer can be sent.</summary>
+    public static readonly SnapshotScope Full = new([]);
+
+    public bool IsPartial => Tables.Count > 0;
+
+    public bool Includes(string tableName) =>
+        Tables.Count == 0 || Tables.Contains(tableName, StringComparer.Ordinal);
+}
+
+/// <summary>
+/// Streams a snapshot of the tables in the requested <see cref="SnapshotScope"/> — every registered
+/// table for a peer that has fallen behind the pruned oplog horizon, or just the baselined ones for
+/// a scoped resync. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
 /// to each live table, pages it into <see cref="SnapshotBatch"/> messages through the gated data
 /// lane, and brackets them with SnapshotBegin/SnapshotComplete carrying the max oplog seq as-of.
 /// </summary>
+/// <remarks>
+/// Clearing the snapshot debt is deliberately NOT done here: whether the flags may be cleared at all
+/// depends on the peer's capability (a 0.3.0 peer acknowledges the snapshot, a pre-0.3.0 one cannot),
+/// which is session state. <see cref="SyncSession"/> owns that decision; this type only streams.
+/// </remarks>
 internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger)
 {
-    /// <summary>Streams the full snapshot; returns the as-of oplog seq it covers, so the session can skip already-covered tail deltas.</summary>
-    public async Task<long> SendAsync(Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
+    /// <summary>
+    /// Streams the snapshot for <paramref name="scope"/>; returns the as-of oplog seq it covers, so
+    /// the session can skip already-covered tail deltas (a FULL snapshot only — a partial one speaks
+    /// for nothing outside its tables).
+    /// </summary>
+    public async Task<long> SendAsync(SnapshotScope scope, Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
     {
         var asOfSeq = await store.GetMaxSeqAsync(ct);
-        await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct);
+        var begin = new SnapshotBegin { AsOfSeq = asOfSeq };
+        begin.Tables.AddRange(scope.Tables);
+        await send(new SyncMessage { SnapshotBegin = begin }, ct);
 
         var totalRows = 0L;
-        foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
+        foreach (var table in db.ReplicatedTables.Values
+                     .Where(t => scope.Includes(t.Name))
+                     .OrderBy(t => t.Name, StringComparer.Ordinal))
         {
             var sql = BuildPageSql(table);
             // Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes
@@ -111,16 +147,9 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
         }
 
         await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = asOfSeq } }, ct);
-        // The sender owed this snapshot (its own needs_snapshot); clearing it here — after the whole
-        // snapshot is on the wire — stops it re-streaming on the next reconnect. Narrow window: the
-        // messages are enqueued on the data lane, not yet transport-delivered, so a crash/drop after
-        // this clear could lose the snapshot with the flag already down. The reconnect handshake's
-        // gap check (peer watermark below the pruned horizon) re-detects that in every case where the
-        // oplog is non-empty; the sole undetectable corner is an EMPTY oplog (nothing to gap-check),
-        // where the peer stays behind until new local writes recreate a horizon. Accepted: closing it
-        // would need transport-level delivery acks for snapshot frames.
-        await store.SetNeedsSnapshotAsync(false, ct);
-        logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows);
+        logger.LogInformation(
+            "Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows, scope {Scope}).",
+            asOfSeq, totalRows, scope.IsPartial ? string.Join(",", scope.Tables) : "all");
         return asOfSeq;
     }
 
@@ -181,6 +210,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
     private long _asOfSeq;
     private long _rows;
     private long _maxHlc;
+    private bool _partial;
 
     // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge
     // reaches the database through the applier and store; kept for construction symmetry in the factory.
@@ -197,7 +227,13 @@ internal sealed class SnapshotApplier : ISnapshotApplier
         _asOfSeq = begin.AsOfSeq;
         _rows = 0;
         _maxHlc = 0;
-        _logger.LogInformation("Snapshot receive begun (as_of_seq {AsOfSeq}).", _asOfSeq);
+        // A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can
+        // emit) means every registered table. Read here rather than on complete because
+        // SnapshotComplete does not repeat it — the receive loop guarantees a Begin precedes both.
+        _partial = begin.Tables.Count > 0;
+        _logger.LogInformation(
+            "Snapshot receive begun (as_of_seq {AsOfSeq}, scope {Scope}).",
+            _asOfSeq, _partial ? string.Join(",", begin.Tables) : "all");
         return Task.CompletedTask;
     }
 
@@ -219,9 +255,20 @@ internal sealed class SnapshotApplier : ISnapshotApplier
 
     public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct)
     {
+        // A PARTIAL snapshot may not move the inbound seq watermark. That watermark is one number
+        // for the whole peer oplog, and jumping it to as_of_seq would declare every table applied
+        // through that point — including the ones this snapshot deliberately left out, whose deltas
+        // below as_of_seq the sender would then never resend. Only the HLC is carried across (seq 0
+        // is a no-op against the MAX guard), so the clock still cannot regress below what the peer
+        // has stamped. The sender makes the mirror-image concession: it does not advance its own
+        // sent-thru watermark past a partial snapshot, so those deltas are still pumped normally
+        // and merely re-deliver rows the snapshot already carried, which LWW absorbs.
+        var seq = _partial ? 0 : complete.AsOfSeq;
         // SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq
         // above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed.
-        await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, _maxHlc, ct);
-        _logger.LogInformation("Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows).", complete.AsOfSeq, _rows);
+        await _store.SetLastAppliedRemoteSeqAsync(seq, _maxHlc, ct);
+        _logger.LogInformation(
+            "Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows, partial {Partial}).",
+            complete.AsOfSeq, _rows, _partial);
     }
 }

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

@@ -47,6 +47,17 @@ internal sealed class SyncSession
     // single receive-loop task, so no synchronization is needed.
     private bool _snapshotReceiving;
 
+    // Whether the peer speaks the 0.3.0 snapshot extensions: scoped snapshots
+    // (SnapshotBegin.tables) and the SnapshotAck that answers a completed one. Written once during
+    // the handshake — which completes before either loop starts — then read-only.
+    private bool _peerSupportsPartialSnapshot;
+
+    // The scope of the snapshot this node is streaming, held until the peer's SnapshotAck discharges
+    // it; null on every session that owes none. Assigned on the pump task strictly BEFORE the
+    // snapshot's first message is enqueued and read on the receive loop when the ack lands, so the
+    // ordering is causal — volatile only to make the cross-thread publication explicit.
+    private SnapshotScope? _sentSnapshotScope;
+
     public SyncSession(
         SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger,
         Func<DateTimeOffset>? utcNow = null, LocalDbMetrics? metrics = null, SyncStatus? status = null)
@@ -63,11 +74,12 @@ internal sealed class SyncSession
 
     /// <summary>
     /// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not
-    /// the peer's) before deltas resume. Receives a gated send delegate that enqueues on the bounded
-    /// data lane (never the raw transport); returns the as-of oplog seq the snapshot covers so the
-    /// pump can skip already-covered tail deltas. Wired to <see cref="SnapshotStreamer.SendAsync"/>.
+    /// the peer's) before deltas resume. Receives the negotiated <see cref="SnapshotScope"/> and a
+    /// gated send delegate that enqueues on the bounded data lane (never the raw transport); returns
+    /// the as-of oplog seq the snapshot covers so the pump can skip already-covered tail deltas.
+    /// Wired to <see cref="SnapshotStreamer.SendAsync"/>.
     /// </summary>
-    public Func<Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
+    public Func<SnapshotScope, Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
 
     /// <summary>
     /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive,
@@ -99,7 +111,7 @@ internal sealed class SyncSession
         var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token);
         try
         {
-            var snapshotRequired = await HandshakeAsync(duplex, control.Writer, linked.Token);
+            var snapshotScope = await HandshakeAsync(duplex, control.Writer, linked.Token);
 
             // The receive loop starts BEFORE any snapshot this node owes is streamed. That ordering
             // is load-bearing, not incidental: a snapshot sent from inside the handshake left the
@@ -111,7 +123,7 @@ internal sealed class SyncSession
             // ordinary flow control. Interleaving is already safe by design: snapshot rows and
             // deltas both merge through LWW, and the snapshot's watermark write is MAX-guarded.
             var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token);
-            var pump = SnapshotThenPumpLoopAsync(snapshotRequired, GatedDataSend, data.Writer, linked.Token);
+            var pump = SnapshotThenPumpLoopAsync(snapshotScope, GatedDataSend, data.Writer, linked.Token);
             var loops = new[] { pump, receive, writer };
 
             // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
@@ -135,12 +147,12 @@ internal sealed class SyncSession
     }
 
     /// <summary>
-    /// Runs the two-phase fail-closed handshake and returns whether THIS node owes the peer a full
-    /// snapshot (its own computation from its own state — the peer's answer drives the peer's send
-    /// direction, symmetrically). Sending that snapshot is deliberately NOT part of the handshake;
-    /// see <see cref="RunAsync"/>.
+    /// Runs the two-phase fail-closed handshake and returns the snapshot THIS node owes the peer —
+    /// null for none, otherwise the <see cref="SnapshotScope"/> to stream (its own computation from
+    /// its own state; the peer's answer drives the peer's send direction, symmetrically). Sending
+    /// that snapshot is deliberately NOT part of the handshake; see <see cref="RunAsync"/>.
     /// </summary>
-    private async Task<bool> HandshakeAsync(
+    private async Task<SnapshotScope?> HandshakeAsync(
         SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
     {
         var peerState = await _store.GetPeerStateAsync(ct);
@@ -158,6 +170,11 @@ internal sealed class SyncSession
 
         ValidateHandshake(localHandshake, peerHandshake);
 
+        // Read before anything can act on it. A pre-0.3.0 peer leaves the field at its proto3
+        // default, so "false" covers both "explicitly unsupported" and "older build" without a
+        // version check — which is the point of advertising a capability rather than a version.
+        _peerSupportsPartialSnapshot = peerHandshake.SupportsPartialSnapshot;
+
         // 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
@@ -191,16 +208,16 @@ internal sealed class SyncSession
         _sentThruSeq = peerAppliedThruSeq;
         Interlocked.Exchange(ref _peerAckedSeq, peerAppliedThruSeq);
 
-        var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct);
+        var snapshotScope = await ComputeSnapshotScopeAsync(peerHandshake, peerState, ct);
         await controlLane.WriteAsync(
-            new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotRequired } }, ct);
+            new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotScope is not null } }, ct);
 
         var peerAckMsg = await ReceiveOneAsync(duplex, ct);
         if (peerAckMsg.MsgCase != SyncMessage.MsgOneofCase.HandshakeAck)
             throw new InvalidOperationException(
                 $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
 
-        return snapshotRequired;
+        return snapshotScope;
     }
 
     /// <summary>
@@ -216,25 +233,55 @@ internal sealed class SyncSession
     /// than racing the transport.
     /// </remarks>
     private async Task SnapshotThenPumpLoopAsync(
-        bool snapshotRequired, Func<SyncMessage, CancellationToken, Task> dataSend,
+        SnapshotScope? snapshotScope, Func<SyncMessage, CancellationToken, Task> dataSend,
         ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
     {
-        if (snapshotRequired)
+        if (snapshotScope is not null)
         {
             if (SnapshotSender is null)
                 throw new NotSupportedException("snapshot required but no snapshot sender configured");
-            var snapshotAsOfSeq = await SnapshotSender(dataSend, ct);
+            // Published BEFORE the first frame is enqueued, so the peer's SnapshotAck can never
+            // arrive on the receive loop while this is still null.
+            Volatile.Write(ref _sentSnapshotScope, snapshotScope);
+            var snapshotAsOfSeq = await SnapshotSender(snapshotScope, dataSend, ct);
             _metrics?.RecordSnapshotSent();
             _status?.SetLastSyncUtc(_utcNow());
-            // The snapshot already carries every row state up to as-of; skipping the pump past it
-            // avoids re-sending tail deltas the peer would only discard via LWW.
-            if (snapshotAsOfSeq > _sentThruSeq)
+            // A FULL snapshot carries every row state up to as-of, so skipping the pump past it
+            // avoids re-sending tail deltas the peer would only discard via LWW. A PARTIAL one
+            // speaks only for its own tables, so the pump must still deliver every delta below
+            // as-of — the peer has not moved its watermark either (see SnapshotApplier.OnComplete).
+            if (!snapshotScope.IsPartial && snapshotAsOfSeq > _sentThruSeq)
                 _sentThruSeq = snapshotAsOfSeq;
+
+            // A peer that cannot acknowledge the snapshot leaves only the 0.2.x option: clear once
+            // the frames are on the data lane. The window that opens (enqueued, not yet delivered —
+            // a crash there loses the snapshot with the debt already discharged) is why 0.3.0 peers
+            // wait for the ack instead. The reconnect handshake's gap check re-detects it whenever
+            // the oplog is non-empty; an EMPTY oplog remains the undetectable corner.
+            if (!_peerSupportsPartialSnapshot)
+                await ClearSentSnapshotDebtAsync(ct);
         }
 
         await PumpLoopAsync(dataLane, ct);
     }
 
+    /// <summary>
+    /// Discharges the snapshot debt this session's outbound snapshot covered, exactly once.
+    /// </summary>
+    /// <remarks>
+    /// Taking the scope out of the field before clearing makes this idempotent, which matters
+    /// because two paths reach it — the pre-0.3.0 clear-on-enqueue above and the peer's SnapshotAck
+    /// — and a peer that acknowledges anyway (a newer build than its handshake claimed) must not
+    /// clear a second time on top of flags a later baseline may already have re-raised.
+    /// </remarks>
+    private async Task ClearSentSnapshotDebtAsync(CancellationToken ct)
+    {
+        var scope = Interlocked.Exchange(ref _sentSnapshotScope, null);
+        if (scope is null)
+            return;
+        await _store.ClearSnapshotDebtAsync(scope.Tables, ct);
+    }
+
     // The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's
     // inbound processing is never starved of its acks behind a backlog of DeltaBatches.
     private static async Task WriterLoopAsync(
@@ -384,6 +431,20 @@ internal sealed class SyncSession
                         _snapshotReceiving = false;
                         _metrics?.RecordSnapshotReceived();
                         _status?.SetLastSyncUtc(_utcNow());
+                        // Acknowledge only to a peer that asked for it. Sent AFTER the applier's
+                        // durable watermark write, so the ack means "I hold these rows", which is
+                        // what lets the sender drop its per-table flags. On the unbounded control
+                        // lane, like DeltaAck: an ack must never queue behind a data backlog.
+                        if (_peerSupportsPartialSnapshot)
+                            await controlLane.WriteAsync(
+                                new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = msg.SnapshotComplete.AsOfSeq } }, ct);
+                        break;
+                    case SyncMessage.MsgOneofCase.SnapshotAck:
+                        // The peer has durably applied the snapshot THIS node streamed, so the
+                        // tables it covered no longer owe one. Nothing to reconcile against the
+                        // ack's as-of seq: only one snapshot is ever sent per session, so the scope
+                        // held in _sentSnapshotScope is unambiguously the one being acknowledged.
+                        await ClearSentSnapshotDebtAsync(ct);
                         break;
                     default:
                         throw new InvalidOperationException(
@@ -565,6 +626,9 @@ internal sealed class SyncSession
             NodeId = _db.NodeId,
             LibSchemaVersion = LibSchemaVersion,
             LastAppliedRemoteSeq = peerState.LastAppliedRemoteSeq,
+            // Unconditional: this build always understands a scoped snapshot and always answers a
+            // completed one. It is a statement about the code, not about configuration or state.
+            SupportsPartialSnapshot = true,
         };
         foreach (var table in _db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
             handshake.Tables.Add(new TableDigest { TableName = table.Name, Digest = table.Digest });
@@ -595,11 +659,63 @@ internal sealed class SyncSession
             "Replication handshake rejected (fail-closed): " + string.Join("; ", parts) + ".");
     }
 
-    private async Task<bool> ComputeSnapshotRequiredAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
+    /// <summary>
+    /// Decides what snapshot this node owes the peer: null for none, <see cref="SnapshotScope.Full"/>
+    /// for every registered table, or a scoped snapshot naming just the tables that need one.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The two sources of snapshot debt are genuinely different in reach, which is why they are
+    /// stored separately. An <b>oplog-horizon</b> gap — the peer's watermark below the oldest delta
+    /// still streamable, which is what a cap prune or an impossible peer ack leaves behind — means
+    /// deltas have been lost for the oplog as a whole, so nothing short of every table is sound. A
+    /// <b>baseline</b> means one table's pre-existing rows were never captured, and says nothing at
+    /// all about the others.
+    /// </para>
+    /// <para>
+    /// A scope that happens to name every registered table is returned as Full rather than as a
+    /// list: it covers exactly what a full snapshot covers, so it should also get a full snapshot's
+    /// watermark advance on both sides, and it needs no promotion for a pre-0.3.0 peer.
+    /// </para>
+    /// <para>
+    /// Flags naming tables that are not registered are ignored (they can only stream nothing).
+    /// <c>DeregisterReplicated</c> drops them with the table's ledger rows, so the surviving way to
+    /// hold one is to drop the SQL table without deregistering — which the handshake's fail-closed
+    /// digest comparison rejects for its own reasons.
+    /// </para>
+    /// </remarks>
+    private async Task<SnapshotScope?> ComputeSnapshotScopeAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
     {
-        if (peerState.NeedsSnapshot)
-            return true;
+        if (peerState.NeedsSnapshot || await HasHorizonGapAsync(peerHandshake, peerState, ct))
+            return SnapshotScope.Full;
 
+        var flagged = await _store.GetTablesNeedingSnapshotAsync(ct);
+        if (flagged.Count == 0)
+            return null;
+
+        var scoped = flagged.Where(_db.ReplicatedTables.ContainsKey).ToList();
+        if (scoped.Count == 0)
+            return null;
+        if (scoped.Count >= _db.ReplicatedTables.Count)
+            return SnapshotScope.Full;
+
+        // A peer that would read SnapshotBegin.tables as "all tables" and then advance its watermark
+        // past deltas this snapshot did not carry gets the full snapshot instead. Wider than asked
+        // for, but it is the only shape the older protocol can express correctly — and it is what
+        // 0.2.x did for every baseline.
+        if (!_peerSupportsPartialSnapshot)
+        {
+            _logger.LogInformation(
+                "Peer {PeerNodeId} does not support scoped snapshots; promoting the snapshot of {Tables} to a full one.",
+                PeerNodeId, string.Join(",", scoped));
+            return SnapshotScope.Full;
+        }
+
+        return new SnapshotScope(scoped);
+    }
+
+    private async Task<bool> HasHorizonGapAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
+    {
         var minRows = await _db.QueryAsync(
             "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
         var minSeq = minRows[0];

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

@@ -6,7 +6,7 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
 /// <summary>
 /// Shared construction of a <see cref="SyncSession"/> and its dependencies (OplogStore + LwwApplier)
 /// from a <see cref="SqliteLocalDb"/> and the bound options. Both the service and client adapters use
-/// this so they build an identical session; snapshot hooks are left null (Task 12 fills them in).
+/// this so they build an identical session.
 /// </summary>
 internal static class SyncSessionFactory
 {

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs

@@ -50,7 +50,8 @@ public interface ILocalDb
     /// <param name="baselineExistingRows">
     /// When true, every row already in the table that has no <c>__localdb_row_version</c> entry is
     /// given one at the <b>baseline floor</b> — HLC <c>0</c>, stamped with this node's id — and, if
-    /// anything was seeded, a snapshot resync is flagged so the peer actually receives them.
+    /// anything was seeded, a snapshot resync is flagged <b>for this table</b> so the peer actually
+    /// receives them.
     /// </param>
     /// <remarks>
     /// <para>
@@ -80,16 +81,29 @@ public interface ILocalDb
     /// and no snapshot is flagged. Passing true on every boot is therefore free after the first.
     /// </para>
     /// <para>
-    /// <b>Cost, and why it is worth planning for.</b> The snapshot flag is a single per-DATABASE
-    /// flag, and a snapshot streams EVERY registered table in full — so baselining one table
-    /// re-sends all of them, and a pair that enables replication late has both nodes doing that to
-    /// each other at once. On a database holding a high-volume table (an event log at its retention
-    /// cap, say) that is the largest transfer this library ever performs, in both directions
-    /// simultaneously. It is bounded work, not runaway: outbound messages respect
-    /// <c>MaxBatchBytes</c>, and the receive side is bounded by the transport's inbox. But the
-    /// operational advice is to <b>purge or trim high-volume tables before enabling replication
-    /// late</b>, and to expect one slow first sync rather than a slow steady state. A per-table
-    /// snapshot flag would narrow this; it needs an on-disk schema change and is not in 0.2.x.
+    /// <b>Cost, and what bounds it (0.3.0).</b> The snapshot flag is <b>per-table</b>
+    /// (<c>__localdb_snapshot_state</c>, on-disk schema v3), so baselining one table asks the peer
+    /// for that table alone — the tables around it are not re-streamed. Through 0.2.x the flag was a
+    /// single per-database bit and baselining anything re-sent everything, in both directions at
+    /// once on a pair enabling replication late.
+    /// </para>
+    /// <para>
+    /// Two cases still stream every registered table, by design. A <b>pre-0.3.0 peer</b> cannot be
+    /// told a snapshot is partial (it would advance its watermark past deltas it never received),
+    /// so a scoped snapshot is promoted to a full one for it. And the <b>oplog-horizon</b> paths —
+    /// a cap prune, an impossible peer ack — invalidate delta continuity for every table at once
+    /// and still set the database-wide flag. So the operational advice survives, narrowed: baseline
+    /// a high-volume table (an event log at its retention cap, say) deliberately, and <b>purge or
+    /// trim it first</b>, because that ONE table's rows still travel in full. The work is bounded
+    /// either way — outbound messages respect <c>MaxBatchBytes</c>, the receive side is bounded by
+    /// the transport's inbox.
+    /// </para>
+    /// <para>
+    /// <b>Crash safety.</b> Between 0.3.0 peers a table's flag is cleared only when the receiver
+    /// acknowledges the completed snapshot, so an ungraceful exit part-way through re-streams the
+    /// same scope on reconnect rather than losing it. Against a pre-0.3.0 peer (which sends no such
+    /// acknowledgement) the 0.2.x behaviour is kept: the flag clears once the snapshot is on the
+    /// wire, and the handshake's horizon gap check is the net.
     /// </para>
     /// </remarks>
     ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows);

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs

@@ -27,9 +27,20 @@ internal static class LocalDbSchema
     /// <c>hlc</c>-predicate delete. It cost a b-tree insert on the CAPTURE path — i.e. inside every
     /// consumer write transaction — to serve nothing.
     /// </description></item>
+    /// <item><description>
+    /// <b>3</b> — adds <c>__localdb_snapshot_state</c>, the PER-TABLE snapshot-resync flag. Until
+    /// v3 the only flag was <c>__localdb_peer_state.needs_snapshot</c>, one bit for the whole
+    /// database, so baselining a single table re-streamed EVERY registered table in full — in both
+    /// directions at once on a pair enabling replication late. The database-wide bit survives with
+    /// its original meaning (the oplog-horizon and cap-prune paths genuinely invalidate every
+    /// table); the new table narrows the baseline case to the tables that actually need it. The
+    /// transition itself is a version stamp only — <see cref="Ddl"/> creates the table with
+    /// <c>IF NOT EXISTS</c> on every open, so an older file gains it whatever its stamp says, and
+    /// an empty flag table means exactly what it meant before: nothing per-table is owed.
+    /// </description></item>
     /// </list>
     /// </remarks>
-    internal const long CurrentSchemaVersion = 2;
+    internal const long CurrentSchemaVersion = 3;
 
     private const string Ddl = """
         CREATE TABLE IF NOT EXISTS __localdb_meta (
@@ -73,6 +84,14 @@ internal static class LocalDbSchema
           id INTEGER PRIMARY KEY CHECK (id = 1),
           applying INTEGER NOT NULL DEFAULT 0
         );
+        -- Per-table snapshot-resync flags (schema v3). A row exists only while that table owes the
+        -- peer a snapshot; the whole-database __localdb_peer_state.needs_snapshot bit still means
+        -- "every registered table" and is what the cap/horizon paths set.
+        -- WITHOUT ROWID: table_name IS the only access path (read as a set, cleared by name).
+        CREATE TABLE IF NOT EXISTS __localdb_snapshot_state (
+          table_name TEXT NOT NULL PRIMARY KEY,
+          needs_snapshot INTEGER NOT NULL DEFAULT 0
+        ) WITHOUT ROWID;
         CREATE TABLE IF NOT EXISTS __localdb_dead_letter (
           id INTEGER PRIMARY KEY AUTOINCREMENT,
           received_utc TEXT NOT NULL,
@@ -143,6 +162,12 @@ internal static class LocalDbSchema
         if (current < 2)
             steps.Add("DROP INDEX IF EXISTS __localdb_oplog_hlc;");
 
+        // 2 -> 3: __localdb_snapshot_state. No step: the shared Ddl above already ran
+        // CREATE TABLE IF NOT EXISTS for it in this same transaction, and an empty flag table is
+        // the correct v3 state for a file that has never baselined a table. Repeating the DDL here
+        // would be a second source of truth for the same layout. The stamp still moves, because
+        // schema_version records what the file's layout IS, not which statements produced it.
+
         if (steps.Count > 0)
         {
             using var ddl = connection.CreateCommand();

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs

@@ -276,7 +276,12 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
                 prune.Transaction = tx;
                 prune.CommandText =
                     "DELETE FROM __localdb_oplog WHERE table_name = $t;" +
-                    "DELETE FROM __localdb_row_version WHERE table_name = $t;";
+                    "DELETE FROM __localdb_row_version WHERE table_name = $t;" +
+                    // A pending per-table snapshot flag for a table nothing streams any more would
+                    // never clear — the streamer skips unregistered tables, so the flag would sit
+                    // in the file forever, and re-registering later would find a stale "already
+                    // owed" it did not ask for. Dropped with the ledger it refers to.
+                    "DELETE FROM __localdb_snapshot_state WHERE table_name = $t;";
                 prune.Parameters.AddWithValue("$t", tableName);
                 prunedRows = prune.ExecuteNonQuery();
             }
@@ -308,12 +313,12 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
     /// the same snapshot anyway, via a cap-prune.
     /// </para>
     /// <para>
-    /// <b>The flag is per-DATABASE, not per-table</b>, and a snapshot streams every registered table
-    /// in full — so baselining one table re-sends all of them, and both nodes of a late-enabled pair
-    /// do it to each other at once. Bounded (batch budget outbound, bounded inbox inbound) but
-    /// potentially the largest transfer this library performs; see
-    /// <see cref="ILocalDb.RegisterReplicated(string, bool)"/> for the operational advice. Narrowing
-    /// it to a per-table flag needs an on-disk schema change and is deferred.
+    /// <b>The flag is PER-TABLE</b> (<c>__localdb_snapshot_state</c>, schema v3): baselining one
+    /// table asks for a snapshot of that table alone, not of every registered table. The
+    /// database-wide <c>__localdb_peer_state.needs_snapshot</c> bit is untouched here and keeps its
+    /// original meaning — the oplog-horizon and cap-prune paths invalidate every table's delta
+    /// continuity at once, and only those paths set it. A peer too old to understand a scoped
+    /// snapshot is served a full one instead; the negotiation lives in the replication package.
     /// </para>
     /// </remarks>
     private void BaselineExistingRows(SqliteTransaction tx, ReplicatedTable table)
@@ -339,7 +344,10 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
 
         using var flag = _master.CreateCommand();
         flag.Transaction = tx;
-        flag.CommandText = "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1";
+        flag.CommandText =
+            "INSERT INTO __localdb_snapshot_state (table_name, needs_snapshot) VALUES ($t, 1) " +
+            "ON CONFLICT(table_name) DO UPDATE SET needs_snapshot = 1";
+        flag.Parameters.AddWithValue("$t", table.Name);
         flag.ExecuteNonQuery();
     }
 

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/BatchBudgetTests.cs

@@ -250,7 +250,7 @@ public sealed class BatchBudgetTests : IDisposable
         var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
 
         var sent = new List<SyncMessage>();
-        await streamer.SendAsync((m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
+        await streamer.SendAsync(SnapshotScope.Full, (m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
 
         var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
             .Select(m => m.SnapshotBatch).ToList();
@@ -279,7 +279,7 @@ public sealed class BatchBudgetTests : IDisposable
         var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
 
         var sent = new List<SyncMessage>();
-        await streamer.SendAsync((m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
+        await streamer.SendAsync(SnapshotScope.Full, (m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
 
         var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
             .Select(m => m.SnapshotBatch).ToList();

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LocalDbSchemaTests.cs

@@ -51,7 +51,8 @@ public sealed class LocalDbSchemaTests : IDisposable
         foreach (var expected in new[]
                  {
                      "__localdb_meta", "__localdb_oplog", "__localdb_peer_state",
-                     "__localdb_row_version", "__localdb_applying", "__localdb_dead_letter"
+                     "__localdb_row_version", "__localdb_applying", "__localdb_dead_letter",
+                     "__localdb_snapshot_state",
                  })
             Assert.Contains(expected, tables);
 
@@ -85,6 +86,44 @@ public sealed class LocalDbSchemaTests : IDisposable
             ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
     }
 
+    [Fact]
+    public void EnsureCreated_UpgradesV2File_AddsPerTableSnapshotState()
+    {
+        using var conn = Open();
+
+        // A file exactly as schema v2 left it: everything except __localdb_snapshot_state, stamped 2.
+        LocalDbSchema.EnsureCreated(conn);
+        Exec(conn, "DROP TABLE __localdb_snapshot_state;" +
+                   "UPDATE __localdb_meta SET schema_version = 2 WHERE id = 1;");
+
+        LocalDbSchema.EnsureCreated(conn);
+
+        Assert.Equal(1, ScalarLong(conn,
+            "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='__localdb_snapshot_state'"));
+        // Empty is the right v3 state for a file that never baselined anything: no table owes a
+        // snapshot, and the database-wide flag it was carrying keeps whatever value it had.
+        Assert.Equal(0, ScalarLong(conn, "SELECT COUNT(*) FROM __localdb_snapshot_state"));
+        Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
+            ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
+    }
+
+    [Fact]
+    public void EnsureCreated_Upgrade_PreservesTheDatabaseWideSnapshotFlag()
+    {
+        // A v2 file mid-resync (cap prune flagged a full snapshot, the peer was down) must still owe
+        // that snapshot after the upgrade. v3 narrows where NEW debt is recorded; it does not
+        // reinterpret debt already on disk.
+        using var conn = Open();
+        LocalDbSchema.EnsureCreated(conn);
+        Exec(conn, "DROP TABLE __localdb_snapshot_state;" +
+                   "UPDATE __localdb_meta SET schema_version = 2 WHERE id = 1;" +
+                   "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1;");
+
+        LocalDbSchema.EnsureCreated(conn);
+
+        Assert.Equal(1, ScalarLong(conn, "SELECT needs_snapshot FROM __localdb_peer_state WHERE id=1"));
+    }
+
     [Fact]
     public void EnsureCreated_Upgrade_PreservesNodeIdentityAndOplogRows()
     {

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs

@@ -291,6 +291,45 @@ public sealed class RegistrationTests : IDisposable
         // Without the flag the seeding would be inert: no gap exists at the handshake, so the peer
         // would be told it needs no snapshot and the baselined rows would never travel.
         Assert.True(await NeedsSnapshot(db));
+        // ...and it is flagged for THIS TABLE ONLY. Through 0.2.x this set the one database-wide
+        // bit, so baselining `orders` re-streamed every other registered table with it.
+        Assert.False(await NeedsFullSnapshot(db));
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_FlagsOnlyTheBaselinedTable()
+    {
+        // The per-table flag's reason to exist: opting one table into replication late must not
+        // schedule a full re-stream of its unrelated neighbours.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+        await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W')");
+
+        db.RegisterReplicated("widgets");
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        Assert.True(await NeedsSnapshot(db, "orders"));
+        Assert.False(await NeedsSnapshot(db, "widgets"));
+        Assert.False(await NeedsFullSnapshot(db));
+    }
+
+    [Fact]
+    public async Task Deregister_DropsThePendingSnapshotFlag()
+    {
+        // A flag naming a table nothing streams any more can never clear itself — the streamer skips
+        // unregistered tables — so it would sit in the file and re-appear as unasked-for debt if the
+        // table were registered again.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+        Assert.True(await NeedsSnapshot(db));
+
+        db.DeregisterReplicated("orders");
+
+        Assert.False(await NeedsSnapshot(db));
     }
 
     [Fact]
@@ -301,7 +340,8 @@ public sealed class RegistrationTests : IDisposable
         await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
 
         db.RegisterReplicated("orders", baselineExistingRows: true);
-        await db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1");
+        // Stand in for the snapshot having been delivered and acknowledged.
+        await db.ExecuteAsync("DELETE FROM __localdb_snapshot_state");
 
         db.RegisterReplicated("orders", baselineExistingRows: true);
 
@@ -480,7 +520,17 @@ public sealed class RegistrationTests : IDisposable
     private static async Task<long> CountAsync(SqliteLocalDb db, string sql) =>
         (await db.QueryAsync(sql, r => r.GetInt64(0)))[0];
 
-    private static async Task<bool> NeedsSnapshot(SqliteLocalDb db) =>
+    /// <summary>Whether <paramref name="table"/> individually owes the peer a snapshot (schema v3).</summary>
+    private static async Task<bool> NeedsSnapshot(SqliteLocalDb db, string table = "orders") =>
+        (await db.QueryAsync(
+            "SELECT COUNT(*) FROM __localdb_snapshot_state WHERE table_name = @t AND needs_snapshot <> 0",
+            r => r.GetInt64(0), new { t = table }))[0] != 0;
+
+    /// <summary>
+    /// The DATABASE-WIDE flag — "every registered table". Baselining must never set it: that is the
+    /// whole point of the per-table flags, and setting it would restore the 0.2.x behaviour silently.
+    /// </summary>
+    private static async Task<bool> NeedsFullSnapshot(SqliteLocalDb db) =>
         (await db.QueryAsync(
             "SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0)))[0] != 0;
 }

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotResyncTests.cs

@@ -89,9 +89,19 @@ public sealed class SnapshotResyncTests : IDisposable
         return (duplex, toSession.Writer, fromSession.Reader);
     }
 
-    private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0)
+    // supportsPartialSnapshot defaults to FALSE, which is exactly what a 0.1.x/0.2.x peer puts on
+    // the wire (the field does not exist in its proto, so proto3 decodes the default). Scripted
+    // peers are therefore old peers unless a test says otherwise.
+    private static Handshake MatchingHandshake(
+        SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0, bool supportsPartialSnapshot = false)
     {
-        var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = lastAppliedRemoteSeq };
+        var h = new Handshake
+        {
+            NodeId = peerNodeId,
+            LibSchemaVersion = 1,
+            LastAppliedRemoteSeq = lastAppliedRemoteSeq,
+            SupportsPartialSnapshot = supportsPartialSnapshot,
+        };
         foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
             h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest });
         return h;
@@ -114,14 +124,30 @@ public sealed class SnapshotResyncTests : IDisposable
 
     // Completes the two-phase handshake against a real session: the peer never requests a snapshot.
     private static async Task DriveHandshakeAsync(
-        SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession, CancellationToken ct)
+        SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession,
+        CancellationToken ct, bool supportsPartialSnapshot = false)
     {
         await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
-        await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct);
+        await toSession.WriteAsync(
+            new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer", supportsPartialSnapshot: supportsPartialSnapshot) }, ct);
         await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
         await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
     }
 
+    // Reads outbound messages until (and including) the first of the given kind.
+    private static async Task<List<SyncMessage>> DrainThroughAsync(
+        ChannelReader<SyncMessage> reader, SyncMessage.MsgOneofCase last, CancellationToken ct)
+    {
+        var all = new List<SyncMessage>();
+        while (true)
+        {
+            var msg = await NextAsync(reader, ct);
+            all.Add(msg);
+            if (msg.MsgCase == last)
+                return all;
+        }
+    }
+
     private static async Task SwallowAsync(Task task)
     {
         try { await task; }
@@ -159,6 +185,12 @@ public sealed class SnapshotResyncTests : IDisposable
         return r[0];
     }
 
+    /// <summary>The tables individually flagged as owing the peer a snapshot (schema v3).</summary>
+    private static async Task<List<string>> PendingSnapshotTables(SqliteLocalDb db) =>
+        (await db.QueryAsync(
+            "SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
+            x => x.GetString(0))).ToList();
+
     private static async Task<long> OplogCount(SqliteLocalDb db)
     {
         var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
@@ -585,6 +617,261 @@ public sealed class SnapshotResyncTests : IDisposable
         await SwallowAsync(run);
     }
 
+    // ---- Per-table snapshot scoping (0.3.0) ----------------------------------------------
+
+    private const string WidgetsSql = "CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)";
+
+    /// <summary>
+    /// A node with two registered tables where exactly one — <c>orders</c> — was baselined.
+    /// <c>widgets</c> is registered BEFORE its rows are written, so ordinary capture ledgers them;
+    /// <c>orders</c> is registered after, so only baselining can. That asymmetry is what makes the
+    /// scope observable: a full snapshot carries both tables' ledgers, a scoped one carries orders
+    /// alone — and both tables hold rows either way, so an empty widgets stream cannot be mistaken
+    /// for "there was nothing to send".
+    /// </summary>
+    private async Task<Side> NewBaselinedSideAsync(ReplicationOptions? options = null)
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        await db.ExecuteAsync(OrdersSql);
+        await db.ExecuteAsync(WidgetsSql);
+
+        // Predates registration: invisible to CDC, so only baselining puts it in the ledger.
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'PRE',1),(2,'PRE',2)");
+
+        db.RegisterReplicated("widgets");
+        await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W1'),(2,'W2')");
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        return NewSessionOver(db, options);
+    }
+
+    /// <summary>The peer of <see cref="NewBaselinedSideAsync"/>: the same two tables, empty, owing nothing.</summary>
+    private async Task<Side> NewTwoTableSideAsync(ReplicationOptions? options = null)
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        await db.ExecuteAsync(OrdersSql);
+        await db.ExecuteAsync(WidgetsSql);
+        db.RegisterReplicated("orders");
+        db.RegisterReplicated("widgets");
+        return NewSessionOver(db, options);
+    }
+
+    private static List<string> SnapshotTablesIn(IEnumerable<SyncMessage> messages) =>
+        messages.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
+            .Select(m => m.SnapshotBatch.TableName)
+            .Distinct(StringComparer.Ordinal)
+            .Order(StringComparer.Ordinal)
+            .ToList();
+
+    [Fact]
+    public async Task Baselining_OneTable_StreamsOnlyThatTable()
+    {
+        // The whole point of the per-table flag. Through 0.2.x the flag was one bit for the whole
+        // database, so baselining `orders` re-streamed `widgets` (and everything else) in full — in
+        // both directions at once on a pair enabling replication late, which on a database holding
+        // an event log at its retention cap is the largest transfer this library performs.
+        var a = await NewBaselinedSideAsync();
+        Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
+        Assert.Equal(0, await NeedsSnapshot(a.Db)); // NOT the database-wide flag
+
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
+        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
+
+        var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
+        Assert.Equal(["orders"], begin.Tables);
+        Assert.Equal(["orders"], SnapshotTablesIn(frames));
+        // The baselined rows really are in there — the scope narrowed the snapshot, it did not empty it.
+        var pks = frames.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
+            .SelectMany(m => m.SnapshotBatch.Rows).Select(r => r.PkJson).Order(StringComparer.Ordinal);
+        Assert.Equal(["{\"id\":1}", "{\"id\":2}"], pks);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Baselining_AgainstAPeerWithoutPartialSupport_StreamsEveryTable()
+    {
+        // Mixed-version interop, the hard constraint. A pre-0.3.0 peer ignores SnapshotBegin.tables
+        // and would advance last_applied_remote_seq to as_of_seq on complete — which for a SCOPED
+        // snapshot silently discards every delta below as_of for the tables it left out. There is no
+        // way to express "partial" to that peer, so it is served the 0.2.x answer: everything.
+        var a = await NewBaselinedSideAsync();
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        // supportsPartialSnapshot omitted == what a 0.1.x/0.2.x handshake decodes to.
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
+        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
+
+        var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
+        Assert.Empty(begin.Tables); // empty == full, the only scope the old wire can express
+        Assert.Equal(["orders", "widgets"], SnapshotTablesIn(frames));
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task PartialSnapshot_DebtClearsOnlyWhenThePeerAcknowledges()
+    {
+        // Crash safety: an ungraceful exit part-way through a snapshot must re-stream it, so the
+        // flag cannot come down when the frames are merely enqueued. The peer's SnapshotAck is
+        // written after its own durable watermark write, so it means "I hold these rows".
+        var a = await NewBaselinedSideAsync();
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
+        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
+        var complete = frames[^1].SnapshotComplete;
+
+        // The entire snapshot is on the wire and unacknowledged: the debt must still stand.
+        Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
+
+        await toSession.WriteAsync(new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = complete.AsOfSeq } }, cts.Token);
+        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Snapshot_ToAPeerWithoutPartialSupport_ClearsOnEnqueue()
+    {
+        // The 0.2.x behaviour, kept deliberately: a peer that sends no SnapshotAck would otherwise
+        // leave the debt standing forever and re-stream the snapshot on every reconnect.
+        var a = await NewBaselinedSideAsync();
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
+        await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
+
+        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task PartialSnapshot_DoesNotAdvanceTheReceiversWatermark()
+    {
+        // The receiver's half of the same invariant. last_applied_remote_seq is ONE number for the
+        // whole peer oplog; moving it to a partial snapshot's as_of would declare the untouched
+        // tables applied through that point, and their deltas below it would never be resent.
+        var a = await NewSideAsync();
+        await a.Store.SetLastAppliedRemoteSeqAsync(10, 0);
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
+
+        var begin = new SnapshotBegin { AsOfSeq = 900 };
+        begin.Tables.Add("orders");
+        await toSession.WriteAsync(new SyncMessage { SnapshotBegin = begin }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
+        }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 900 } }, cts.Token);
+
+        // The ack proves the complete was processed — and that a 0.3.0 receiver answers one at all.
+        var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotAck, cts.Token);
+        Assert.Equal(900, ack.SnapshotAck.AsOfSeq);
+
+        Assert.Equal(10, await LastAppliedRemoteSeq(a.Db)); // NOT 900
+        Assert.Single(await ReadOrders(a.Db));              // the rows still merged
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task FullSnapshot_ToAnOldPeer_DrawsNoSnapshotAck()
+    {
+        // Symmetry check on the negotiation: a 0.3.0 receiver must not put a message a pre-0.3.0
+        // sender has never heard of on the wire back to it.
+        var a = await NewSideAsync();
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 300 } }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
+        }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 300 } }, cts.Token);
+
+        // A full snapshot from an old peer still advances the watermark — that is the 0.2.x contract.
+        await WaitForAsync(async () => await LastAppliedRemoteSeq(a.Db) == 300, RunTimeout);
+        // Barrier: an ack for this delta proves everything before it was processed, so an absent
+        // SnapshotAck is absent rather than merely late.
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
+            {
+                Seq = 301, TableName = "orders", PkJson = "{\"id\":2}",
+                RowJson = "{\"id\":2,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
+            } } },
+        }, cts.Token);
+        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+        Assert.False(fromSession.TryPeek(out var stray) && stray.MsgCase == SyncMessage.MsgOneofCase.SnapshotAck);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task BaselinedTable_ReachesARealPeer_WithoutRestreamingTheOthers()
+    {
+        // End to end over two real databases: the baselined table's pre-existing rows arrive via the
+        // scoped snapshot, the untouched table's rows arrive as ordinary deltas, and the debt clears.
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
+        var a = await NewBaselinedSideAsync(options);
+        var b = await NewTwoTableSideAsync(options);
+
+        var (da, db) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(db, cts.Token);
+
+        async Task<bool> Converged() =>
+            (await ReadOrders(b.Db)).Count == 2
+            && (await b.Db.QueryAsync("SELECT COUNT(*) FROM widgets", x => x.GetInt64(0)))[0] == 2;
+
+        await WaitForAsync(Converged, RunTimeout);
+        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
+        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
+
+        // A partial snapshot speaks for nothing outside its tables, so neither side may skip the
+        // deltas it covered: B's watermark tracks the oplog, not the snapshot's as-of.
+        Assert.Equal(0, await NeedsSnapshot(a.Db));
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
     [Fact]
     public async Task Sender_ClearsNeedsSnapshotFlag()
     {

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs

@@ -553,8 +553,9 @@ public sealed class SyncSessionTests : IDisposable
             await a.Store.SetNeedsSnapshotAsync(true, default);
             var (duplex, toSession, fromSession) = ScriptedPeer();
             var sent = new TaskCompletionSource();
-            a.Session.SnapshotSender = async (send, token) =>
+            a.Session.SnapshotSender = async (scope, send, token) =>
             {
+                Assert.False(scope.IsPartial); // the database-wide flag always means every table
                 sent.TrySetResult();
                 await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token);
                 return 7;