diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 3907c58a01..2b83ada77d 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -42,6 +42,27 @@ type batchPublisher struct {
 	log        *service.Logger
 	cacheLSN   func(ctx context.Context, lsn replication.LSN) error
 	shutSig    *shutdown.Signaller
+
+	// snapshotAckWG counts published snapshot batches that have not yet been
+	// acknowledged downstream. The snapshot->streaming handoff blocks on it so
+	// the post-snapshot LSN is never persisted while snapshot rows are in flight.
+	snapshotAckWG sync.WaitGroup
+	// persistMu serializes resolve+persist pairs. The ordered tracker hands
+	// out monotonically increasing frontiers, but ack functions and
+	// CheckpointWindow run on different goroutines: without a shared critical
+	// section around resolveFn()+cacheLSN, two persists can land out of order
+	// and regress the cached resume position.
+	persistMu sync.Mutex
+	// pendingCheckpointLSN mirrors the CheckpointLSN of the most recently
+	// added message (or a stronger drained-window LSN, see CheckpointWindow):
+	// the start LSN of the last transaction whose rows are all published, the
+	// only value safe to persist as a resume position. Guarded by batcherMu,
+	// so at flush time it always belongs to the flushed batch's last message.
+	pendingCheckpointLSN replication.LSN
+	// buffered counts messages currently held by the batcher (guarded by
+	// batcherMu). CheckpointWindow uses it to decide between deferring the
+	// window checkpoint to the buffered batch and registering a marker.
+	buffered int
 }
 
 // newBatchPublisher creates an instance of batchPublisher.
@@ -80,7 +101,11 @@ func (p *batchPublisher) loop() {
 			return
 		}
 
+		// UntilNext reads the batcher's internal state, which concurrent
+		// Publish calls mutate under batcherMu — take the same lock.
+		p.batcherMu.Lock()
 		tNext, exists := p.batcher.UntilNext()
+		p.batcherMu.Unlock()
 		if !exists {
 			if flushBatchTicker != nil {
 				flushBatchTicker.Stop()
@@ -104,9 +129,14 @@ func (p *batchPublisher) loop() {
 		adjustTimedFlush()
 		select {
 		case <-flushBatch:
-			var sendBatch service.MessageBatch
+			var (
+				tracked  *trackedBatch
+				trackErr error
+			)
 
-			// Wrap this in a closure to make locking/unlocking easier.
+			// Wrap this in a closure to make locking/unlocking easier. Track
+			// happens under the same lock as the flush so the checkpoint
+			// sequence matches flush order.
 			func() {
 				p.batcherMu.Lock()
 				defer p.batcherMu.Unlock()
@@ -119,13 +149,19 @@ func (p *batchPublisher) loop() {
 					return
 				}
 
+				var sendBatch service.MessageBatch
 				if sendBatch, _ = p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
 					return
 				}
+				p.buffered = 0
+				tracked, trackErr = p.trackBatchLocked(closeAtLeisureCtx, sendBatch)
 			}()
+			if trackErr != nil {
+				return
+			}
 
-			if len(sendBatch) > 0 {
-				if err := p.publishBatch(closeAtLeisureCtx, sendBatch); err != nil {
+			if tracked != nil {
+				if err := p.sendTracked(closeAtLeisureCtx, tracked); err != nil {
 					return
 				}
 			}
@@ -176,10 +212,21 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: s})
 	}
 
-	var flushedBatch []*service.Message
+	// Flush and Track must be atomic: Track order defines the checkpoint
+	// sequence, so another flusher (the timed-flush loop) must not interleave
+	// between our flush and our Track. Only the channel send happens outside
+	// the lock.
+	var tracked *trackedBatch
 	b.batcherMu.Lock()
+	b.pendingCheckpointLSN = m.CheckpointLSN
 	if b.batcher.Add(msg) {
-		flushedBatch, err = b.batcher.Flush(ctx)
+		var flushedBatch []*service.Message
+		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
+			b.buffered = 0
+			tracked, err = b.trackBatchLocked(ctx, flushedBatch)
+		}
+	} else {
+		b.buffered++
 	}
 	b.batcherMu.Unlock()
 	if err != nil {
@@ -187,8 +234,8 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	}
 
 	// If a batch was flushed, publish it outside the lock
-	if len(flushedBatch) > 0 {
-		if err := b.publishBatch(ctx, flushedBatch); err != nil {
+	if tracked != nil {
+		if err := b.sendTracked(ctx, tracked); err != nil {
 			return fmt.Errorf("publishing flushed batch: %w", err)
 		}
 	}
@@ -196,40 +243,153 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	return nil
 }
 
-func (b *batchPublisher) publishBatch(ctx context.Context, batch service.MessageBatch) error {
-	if len(batch) == 0 {
-		return nil
-	}
+// trackedBatch pairs a ready-to-send asyncMessage with the bookkeeping needed
+// to roll back its snapshot-gate slot if the send fails.
+type trackedBatch struct {
+	msg        asyncMessage
+	isSnapshot bool
+}
 
+// trackBatchLocked registers the batch with the ordered checkpoint tracker and
+// builds its ack function. It MUST be called with batcherMu held: Track order
+// defines the checkpoint sequence, so it has to match flush order exactly.
+func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.MessageBatch) (*trackedBatch, error) {
 	lastMsg := batch[len(batch)-1]
-	var checkpointLSN []byte
-	// snapshot records don't have a lsn as we don't track those
-	if lsn, ok := lastMsg.MetaGet("lsn"); ok {
-		checkpointLSN = replication.LSN(lsn)
+	// Checkpoint only the pending checkpoint LSN: the last transaction whose
+	// rows are all published. The row's own lsn must never be persisted — all
+	// rows of a transaction share a start LSN and resume is exclusive (> lsn),
+	// so persisting it mid-transaction would skip the transaction's remaining
+	// rows on restart. Snapshot rows never carry one; we don't track those.
+	checkpointLSN := []byte(b.pendingCheckpointLSN)
+
+	// Snapshot batches are tracked so the snapshot->streaming handoff can block
+	// until they are acknowledged downstream (see waitSnapshotAcks).
+	isSnapshotBatch := false
+	if op, ok := lastMsg.MetaGet("operation"); ok && op == replication.MessageOperationRead.String() {
+		isSnapshotBatch = true
 	}
 
 	resolveFn, err := b.checkpoint.Track(ctx, checkpointLSN, int64(len(batch)))
 	if err != nil {
-		return fmt.Errorf("tracking LSN checkpoint for batch: %w", err)
-	}
-	msg := asyncMessage{
-		msg: batch,
-		ackFn: func(ctx context.Context, _ error) error {
-			lsn := resolveFn()
-			if lsn != nil && len(*lsn) != 0 {
-				return b.cacheLSN(ctx, *lsn)
-			}
-			return nil
+		return nil, fmt.Errorf("tracking LSN checkpoint for batch: %w", err)
+	}
+	if isSnapshotBatch {
+		b.snapshotAckWG.Add(1)
+	}
+	return &trackedBatch{
+		isSnapshot: isSnapshotBatch,
+		msg: asyncMessage{
+			msg: batch,
+			// The ack error is deliberately ignored: nacks are replayed by
+			// auto_replay_nacks (the default), and disabling that is a
+			// documented opt-in to DROP rejected messages, so the checkpoint
+			// must advance past them rather than pin the tracker.
+			ackFn: func(ctx context.Context, _ error) error {
+				if isSnapshotBatch {
+					defer b.snapshotAckWG.Done()
+				}
+				b.persistMu.Lock()
+				defer b.persistMu.Unlock()
+				lsn := resolveFn()
+				if lsn != nil && len(*lsn) != 0 {
+					return b.cacheLSN(ctx, *lsn)
+				}
+				return nil
+			},
 		},
+	}, nil
+}
+
+// sendTracked hands a tracked batch to ReadBatch. Must be called WITHOUT
+// batcherMu held (the send blocks until consumed). A failed send releases the
+// batch's snapshot-gate slot.
+func (b *batchPublisher) sendTracked(ctx context.Context, tracked *trackedBatch) error {
+	select {
+	case b.msgChan <- tracked.msg:
+		return nil
+	case <-ctx.Done():
+		if tracked.isSnapshot {
+			b.snapshotAckWG.Done()
+		}
+		return ctx.Err()
 	}
+}
+
+// waitSnapshotAcks blocks until every published snapshot batch has been
+// acknowledged (or nacked) downstream, or until ctx is cancelled. Nacked
+// batches release the gate too: redelivery is owned by auto_replay_nacks,
+// and disabling that is a documented opt-in to drop rejections. The ctx
+// escape prevents a permanently-failing downstream from wedging shutdown.
+func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
+	drained := make(chan struct{})
+	go func() {
+		// May outlive this call if ctx fires first; bounded by process lifetime.
+		b.snapshotAckWG.Wait()
+		close(drained)
+	}()
 	select {
-	case b.msgChan <- msg:
+	case <-drained:
 		return nil
 	case <-ctx.Done():
 		return ctx.Err()
 	}
 }
 
+// CheckpointWindow records that every transaction up to and including lsn is
+// fully published (a polling window drained), giving the stream an exact
+// resume position instead of lagging one transaction behind (which would
+// re-deliver the final transaction of a burst on every restart).
+//
+// The user's batching policy stays in charge of batch sizes: if rows from the
+// window are still buffered, the window-end LSN simply becomes their batch's
+// checkpoint payload (safe, and stronger than the last row's transaction
+// boundary). Only when the batcher is empty is an immediately-resolved marker
+// slot registered, so lsn persists once every published batch is acked.
+func (b *batchPublisher) CheckpointWindow(ctx context.Context, lsn replication.LSN) error {
+	b.batcherMu.Lock()
+	if b.buffered > 0 {
+		b.pendingCheckpointLSN = lsn
+		b.batcherMu.Unlock()
+		return nil
+	}
+	resolveFn, err := b.checkpoint.Track(ctx, lsn, 1)
+	b.batcherMu.Unlock()
+	if err != nil {
+		return fmt.Errorf("tracking window checkpoint: %w", err)
+	}
+	// Resolve the marker immediately: if everything before it is already
+	// acked this persists lsn now; otherwise the last outstanding ack's
+	// resolve will surface it.
+	b.persistMu.Lock()
+	defer b.persistMu.Unlock()
+	if resolved := resolveFn(); resolved != nil && len(*resolved) != 0 {
+		return b.cacheLSN(ctx, *resolved)
+	}
+	return nil
+}
+
+// flushCurrent flushes any partial batch still held by the batcher and
+// publishes it, leaving the publisher loop running. Used at the
+// snapshot->streaming handoff so every snapshot row is published (and can be
+// awaited via waitSnapshotAcks) before the post-snapshot LSN is persisted.
+func (b *batchPublisher) flushCurrent(ctx context.Context) error {
+	if b.batcher == nil {
+		return nil
+	}
+	var tracked *trackedBatch
+	b.batcherMu.Lock()
+	remaining, err := b.batcher.Flush(ctx)
+	if err == nil && len(remaining) > 0 {
+		b.buffered = 0
+		tracked, err = b.trackBatchLocked(ctx, remaining)
+	}
+	b.batcherMu.Unlock()
+	if err != nil || tracked == nil {
+		return err
+	}
+	return b.sendTracked(ctx, tracked)
+}
+
 func (b *batchPublisher) msgs() <-chan asyncMessage {
 	return b.msgChan
 }
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
new file mode 100644
index 0000000000..f6d6931453
--- /dev/null
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -0,0 +1,462 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package mssqlserver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Jeffail/checkpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+	"github.com/redpanda-data/connect/v4/internal/impl/mssqlserver/replication"
+)
+
+func TestSnapshotAckGate(t *testing.T) {
+	t.Run("blocks until the snapshot batch is acked", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
+
+		done := make(chan error, 1)
+		go func() { done <- publisher.waitSnapshotAcks(ctx) }()
+
+		select {
+		case err := <-done:
+			t.Fatalf("waitSnapshotAcks returned before the snapshot batch was acked: %v", err)
+		case <-time.After(100 * time.Millisecond):
+		}
+
+		require.NoError(t, msg.ackFn(ctx, nil))
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+		case <-time.After(5 * time.Second):
+			t.Fatal("waitSnapshotAcks did not return after the snapshot batch was acked")
+		}
+	})
+
+	t.Run("a nack also releases the gate", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
+		// Nacks count as settled: replay is owned by auto_replay_nacks, and
+		// disabling it is a documented opt-in to drop rejections.
+		require.NoError(t, msg.ackFn(ctx, errors.New("downstream failure")))
+		require.NoError(t, publisher.waitSnapshotAcks(ctx))
+	})
+
+	t.Run("streaming batches do not hold the gate", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		// Published but never acked: must not block the gate.
+		publishAndReceive(t, ctx, publisher, streamingEvent("00000030", ""))
+
+		require.NoError(t, publisher.waitSnapshotAcks(ctx))
+	})
+
+	t.Run("context cancellation escapes the gate", func(t *testing.T) {
+		publisher, _ := newTestBatchPublisher(t)
+
+		ctx, cancel := context.WithCancel(t.Context())
+		publishAndReceive(t, ctx, publisher, snapshotEvent())
+
+		done := make(chan error, 1)
+		go func() { done <- publisher.waitSnapshotAcks(ctx) }()
+		cancel()
+
+		select {
+		case err := <-done:
+			require.ErrorIs(t, err, context.Canceled)
+		case <-time.After(5 * time.Second):
+			t.Fatal("waitSnapshotAcks did not return after context cancellation")
+		}
+	})
+}
+
+func TestFlushCurrent(t *testing.T) {
+	ctx := t.Context()
+	// Count=100 keeps published events buffered in the batcher until flushed.
+	publisher, _ := newTestBatchPublisherWithCount(t, 100)
+
+	publishEvent := func() {
+		t.Helper()
+		require.NoError(t, publisher.Publish(ctx, snapshotEvent()))
+	}
+	receive := func(failMsg string) {
+		t.Helper()
+		got := make(chan asyncMessage, 1)
+		go func() { got <- <-publisher.msgs() }()
+		require.NoError(t, publisher.flushCurrent(ctx))
+		select {
+		case m := <-got:
+			require.Len(t, m.msg, 1)
+		case <-time.After(5 * time.Second):
+			t.Fatal(failMsg)
+		}
+	}
+
+	publishEvent()
+	receive("flushCurrent did not publish the buffered partial batch")
+
+	// The loop must still be alive after flushCurrent: a second
+	// publish+flush must work identically.
+	publishEvent()
+	receive("publisher loop no longer functional after flushCurrent")
+}
+
+func TestCheckpointSelection(t *testing.T) {
+	t.Run("persists the transaction boundary, never the row's own lsn", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
+		require.NoError(t, am.ackFn(ctx, nil))
+
+		lsns := cachedLSNs()
+		require.Len(t, lsns, 1)
+		require.Equal(t, "00000041", string(lsns[0]),
+			"the checkpoint must be the last fully-published transaction boundary, not the row's own LSN")
+	})
+
+	t.Run("no boundary yet (first transaction) persists nothing", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, am.ackFn(ctx, nil))
+
+		require.Empty(t, cachedLSNs(),
+			"a batch ending mid-transaction (no prior complete transaction) must not persist any LSN")
+	})
+
+	t.Run("a nack resolves too: auto_replay_nacks off is an opt-in drop", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		b1 := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
+		b2 := publishAndReceive(t, ctx, publisher, streamingEvent("00000043", "00000042"))
+
+		// A nacked batch is deleted per the auto_replay_nacks contract: its
+		// slot resolves so the stream continues past it instead of pinning
+		// the tracker and back-pressuring forever.
+		require.NoError(t, b1.ackFn(ctx, errors.New("downstream failure")))
+		require.NoError(t, b2.ackFn(ctx, nil))
+
+		lsns := cachedLSNs()
+		require.NotEmpty(t, lsns, "the checkpoint must continue advancing past a dropped batch")
+		require.Equal(t, "00000042", string(lsns[len(lsns)-1]))
+	})
+}
+
+func TestCheckpointWindow(t *testing.T) {
+	t.Run("persists immediately when all prior batches are acked", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, am.ackFn(ctx, nil))
+		require.Empty(t, cachedLSNs(), "mid-transaction batch must not persist anything on its own")
+
+		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+
+		lsns := cachedLSNs()
+		require.Len(t, lsns, 1)
+		require.Equal(t, "00000042", string(lsns[0]),
+			"a drained window must checkpoint its exact end position")
+	})
+
+	t.Run("waits for outstanding acks before surfacing", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+		require.Empty(t, cachedLSNs(), "the window end must not persist while its batches are unacked")
+
+		require.NoError(t, am.ackFn(ctx, nil))
+		lsns := cachedLSNs()
+		require.Len(t, lsns, 1)
+		require.Equal(t, "00000042", string(lsns[0]))
+	})
+
+	t.Run("a nacked batch settles the window checkpoint too", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+
+		// A nacked batch is deleted per the auto_replay_nacks contract, so
+		// the window checkpoint behind it still persists.
+		require.NoError(t, am.ackFn(ctx, errors.New("downstream failure")))
+		lsns := cachedLSNs()
+		require.NotEmpty(t, lsns)
+		require.Equal(t, "00000042", string(lsns[len(lsns)-1]))
+	})
+}
+
+func TestCheckpointWindowDefersToBufferedBatch(t *testing.T) {
+	ctx := t.Context()
+	// Count=100 keeps published events buffered: the window checkpoint must
+	// ride on the eventual batch instead of forcing a flush (which would
+	// override the user's batching policy).
+	publisher, cachedLSNs := newTestBatchPublisherWithCount(t, 100)
+
+	require.NoError(t, publisher.Publish(ctx, streamingEvent("00000042", "")))
+	require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+	require.Empty(t, cachedLSNs(), "a deferred window checkpoint must not persist before its batch is acked")
+
+	// No batch may have been force-flushed by CheckpointWindow.
+	select {
+	case m := <-publisher.msgs():
+		t.Fatalf("CheckpointWindow force-flushed a batch of %d messages, overriding the batching policy", len(m.msg))
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	// When the batch eventually flushes and acks, it carries the window LSN.
+	got := make(chan asyncMessage, 1)
+	go func() { got <- <-publisher.msgs() }()
+	require.NoError(t, publisher.flushCurrent(ctx))
+	var am asyncMessage
+	select {
+	case am = <-got:
+	case <-time.After(5 * time.Second):
+		t.Fatal("buffered batch was never flushed")
+	}
+	require.NoError(t, am.ackFn(ctx, nil))
+
+	lsns := cachedLSNs()
+	require.Len(t, lsns, 1)
+	require.Equal(t, "00000042", string(lsns[0]), "the drained-window LSN must ride on the buffered batch's checkpoint")
+}
+
+// TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
+// count-triggered flush in Publish and the timed-flush loop) and asserts the
+// persisted checkpoint never regresses when batches are acked in delivery
+// order. Before Track was moved under the batcher mutex, the two flushers
+// could interleave between flush and Track, registering batches with the
+// ordered tracker in the wrong order and persisting a regressing LSN. Run with
+// -race to also catch the underlying data race structurally.
+func TestTrackOrderUnderConcurrentFlush(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](1000)
+
+	// Count 2 + a tiny period keeps both flush paths active concurrently.
+	batcher, err := (service.BatchPolicy{Count: 2, Period: "1ms"}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	var (
+		mu        sync.Mutex
+		persisted []replication.LSN
+	)
+	publisher.cacheLSN = func(_ context.Context, lsn replication.LSN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		persisted = append(persisted, lsn)
+		return nil
+	}
+
+	// Consumer: ack every batch immediately, in delivery order.
+	consumerDone := make(chan struct{})
+	consumerCtx, stopConsumer := context.WithCancel(ctx)
+	go func() {
+		defer close(consumerDone)
+		for {
+			select {
+			case m := <-publisher.msgs():
+				_ = m.ackFn(ctx, nil)
+			case <-consumerCtx.Done():
+				return
+			}
+		}
+	}()
+
+	const events = 500
+	for i := range events {
+		// %08d keeps lexicographic order == numeric order, like real LSNs.
+		lsn := fmt.Sprintf("%08d", i)
+		require.NoError(t, publisher.Publish(ctx, streamingEvent(lsn, lsn)))
+	}
+	require.NoError(t, publisher.flushCurrent(ctx))
+
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(persisted) > 0 && string(persisted[len(persisted)-1]) == fmt.Sprintf("%08d", events-1)
+	}, 10*time.Second, 10*time.Millisecond, "final LSN was never persisted")
+	stopConsumer()
+	<-consumerDone
+
+	mu.Lock()
+	defer mu.Unlock()
+	for i := 1; i < len(persisted); i++ {
+		require.GreaterOrEqual(t, string(persisted[i]), string(persisted[i-1]),
+			"persisted checkpoint regressed at index %d: %v", i, persisted)
+	}
+}
+
+// TestPersistOrderUnderConcurrentAcksAndWindows locks in that the cached
+// resume position never regresses when batch acks (pipeline goroutines) and
+// CheckpointWindow markers (stream goroutine) persist concurrently: the
+// resolve+persist pair must be a single critical section, otherwise two
+// persists can land out of order.
+func TestPersistOrderUnderConcurrentAcksAndWindows(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](1000)
+
+	batcher, err := (service.BatchPolicy{Count: 1}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	var (
+		mu        sync.Mutex
+		persisted []replication.LSN
+	)
+	publisher.cacheLSN = func(_ context.Context, lsn replication.LSN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		persisted = append(persisted, lsn)
+		return nil
+	}
+
+	// Consumer: ack every batch on its own goroutine so acks complete out of
+	// order relative to each other and to the window markers.
+	var ackWG sync.WaitGroup
+	consumerDone := make(chan struct{})
+	consumerCtx, stopConsumer := context.WithCancel(ctx)
+	go func() {
+		defer close(consumerDone)
+		for {
+			select {
+			case m := <-publisher.msgs():
+				ackWG.Go(func() {
+					_ = m.ackFn(ctx, nil)
+				})
+			case <-consumerCtx.Done():
+				return
+			}
+		}
+	}()
+
+	const events = 400
+	for i := range events {
+		lsn := fmt.Sprintf("%08d", i)
+		require.NoError(t, publisher.Publish(ctx, streamingEvent(lsn, lsn)))
+		// A drained polling window ends every 10 rows; its end LSN persists
+		// via an immediately-resolved marker racing the in-flight acks.
+		if i%10 == 9 {
+			require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN(lsn)))
+		}
+	}
+
+	finalLSN := fmt.Sprintf("%08d", events-1)
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(persisted) > 0 && string(persisted[len(persisted)-1]) == finalLSN
+	}, 10*time.Second, 10*time.Millisecond, "final LSN was never persisted")
+	stopConsumer()
+	<-consumerDone
+	ackWG.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	for i := 1; i < len(persisted); i++ {
+		require.GreaterOrEqual(t, string(persisted[i]), string(persisted[i-1]),
+			"persisted checkpoint regressed at index %d: %v", i, persisted)
+	}
+}
+
+// newTestBatchPublisher builds a publisher whose batcher flushes on every
+// published event (count=1), so tests drive the production
+// Publish->trackBatchLocked->sendTracked path directly.
+func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.LSN) {
+	t.Helper()
+	return newTestBatchPublisherWithCount(t, 1)
+}
+
+func newTestBatchPublisherWithCount(t *testing.T, count int) (*batchPublisher, func() []replication.LSN) {
+	t.Helper()
+
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](100)
+
+	batcher, err := (service.BatchPolicy{Count: count}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	var (
+		mu         sync.Mutex
+		cachedLSNs []replication.LSN
+	)
+	publisher.cacheLSN = func(_ context.Context, lsn replication.LSN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		cachedLSNs = append(cachedLSNs, lsn)
+		return nil
+	}
+
+	cachedLSNsFn := func() []replication.LSN {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]replication.LSN(nil), cachedLSNs...)
+	}
+
+	return publisher, cachedLSNsFn
+}
+
+func snapshotEvent() replication.MessageEvent {
+	return replication.MessageEvent{
+		Schema:    "dbo",
+		Table:     "t",
+		Operation: replication.MessageOperationRead.String(),
+		Data:      map[string]any{"a": 1},
+	}
+}
+
+func streamingEvent(lsn, checkpointLSN string) replication.MessageEvent {
+	return replication.MessageEvent{
+		Schema:        "dbo",
+		Table:         "t",
+		Operation:     replication.MessageOperationInsert.String(),
+		LSN:           replication.LSN(lsn),
+		CheckpointLSN: replication.LSN(checkpointLSN),
+		Data:          map[string]any{"a": 1},
+	}
+}
+
+// publishAndReceive publishes a single event through the production Publish
+// path (count=1 batcher: every event flushes, tracks, and sends immediately)
+// and returns the delivered asyncMessage.
+func publishAndReceive(t *testing.T, ctx context.Context, publisher *batchPublisher, event replication.MessageEvent) asyncMessage {
+	t.Helper()
+	go func() {
+		_ = publisher.Publish(ctx, event)
+	}()
+	return <-publisher.msgs()
+}
diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc.go b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
index 4b837c7f9b..4034ccd745 100644
--- a/internal/impl/mssqlserver/input_mssqlserver_cdc.go
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
@@ -366,6 +366,24 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 				i.stopSig.TriggerHasStopped()
 				return
 			}
+
+			// Flush the partial snapshot batch still held by the batcher, then
+			// block until every snapshot batch is acknowledged downstream.
+			// Persisting the LSN any earlier would let a crash in this window
+			// skip un-acked snapshot rows on restart. Blocks until acks drain
+			// or soft-stop (no timeout, by design; see postgres_cdc's
+			// equivalent barrier).
+			if err = i.publisher.flushCurrent(softCtx); err != nil {
+				i.log.Errorf("Failed to flush remaining snapshot batches. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				i.stopSig.TriggerHasStopped()
+				return
+			}
+			if err = i.publisher.waitSnapshotAcks(softCtx); err != nil {
+				i.log.Infof("Interrupted while waiting for snapshot acknowledgements. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				i.stopSig.TriggerHasStopped()
+				return
+			}
+
 			if err = i.cacheLSN(softCtx, maxLSN); err != nil {
 				if i.stopSig.IsHardStopSignalled() {
 					i.log.Errorf("Shutting down snapshotting process: %s", err)
diff --git a/internal/impl/mssqlserver/integration_test.go b/internal/impl/mssqlserver/integration_test.go
index 359193f5f8..7f8c7efb9d 100644
--- a/internal/impl/mssqlserver/integration_test.go
+++ b/internal/impl/mssqlserver/integration_test.go
@@ -11,9 +11,11 @@ package mssqlserver_test
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -429,6 +431,241 @@ microsoft_sql_server_cdc:
 	require.NoError(t, stream.StopWithin(time.Second*10))
 }
 
+// TestIntegration_MicrosoftSQLServerCDC_SnapshotAckBarrier verifies that a
+// crash during the snapshot->streaming handoff (after snapshot rows are
+// emitted but before they are acknowledged) does not lose data: because the
+// post-snapshot LSN is only persisted once every snapshot batch is acked, the
+// snapshot must re-run on restart. See CON-504.
+func TestIntegration_MicrosoftSQLServerCDC_SnapshotAckBarrier(t *testing.T) {
+	integration.CheckSkip(t)
+
+	connStr, db := mssqlservertest.SetupTestWithMicrosoftSQLServerVersion(t)
+	require.NoError(t, db.CreateTableWithCDCEnabledIfNotExists(t.Context(), "dbo.barrier", "CREATE TABLE dbo.barrier (id INT IDENTITY(1,1) PRIMARY KEY);"))
+
+	const rowCount = 5
+	for range rowCount {
+		db.MustExec("INSERT INTO dbo.barrier DEFAULT VALUES")
+	}
+	db.WaitForCDCChanges(t.Context(), rowCount, "dbo.barrier")
+
+	// batching.count == rowCount forces all snapshot rows into a single output
+	// batch, so the run-1 consumer receives them all at once and can then block
+	// without acking - reproducing the "emitted but not yet acked" handoff state.
+	cfg := fmt.Sprintf(`
+microsoft_sql_server_cdc:
+  connection_string: %s
+  stream_snapshot: true
+  checkpoint_cache: ""
+  include: ["dbo.barrier"]
+  batching:
+    count: %d
+    period: 1h`, connStr, rowCount)
+
+	// Run 1: receive the snapshot rows but never acknowledge them, then
+	// simulate a crash by cancelling the run before the LSN can be persisted.
+	t.Log("Launching run 1 (blocked consumer, simulated crash)...")
+	received := make(chan struct{}, 1)
+	run1Builder := service.NewStreamBuilder()
+	require.NoError(t, run1Builder.AddInputYAML(cfg))
+	require.NoError(t, run1Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run1Builder.AddBatchConsumerFunc(func(ctx context.Context, _ service.MessageBatch) error {
+		select {
+		case received <- struct{}{}:
+		default:
+		}
+		// Block without acking until the simulated crash cancels our context.
+		<-ctx.Done()
+		return ctx.Err()
+	}))
+	run1, err := run1Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run1.Resources())
+
+	run1Ctx, crash := context.WithCancel(t.Context())
+	run1Done := make(chan struct{})
+	go func() {
+		defer close(run1Done)
+		_ = run1.Run(run1Ctx)
+	}()
+
+	select {
+	case <-received:
+	case <-time.After(5 * time.Minute):
+		t.Fatal("snapshot rows were never delivered to the run-1 output")
+	}
+	// Give the input time to reach the ack barrier (and, in the buggy version,
+	// to persist the post-snapshot LSN) before we crash.
+	time.Sleep(5 * time.Second)
+	crash()
+	select {
+	case <-run1Done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("run 1 did not stop after the simulated crash")
+	}
+
+	// The barrier must have prevented the post-snapshot LSN from being
+	// persisted, since the snapshot rows were never acknowledged. Without it a
+	// cached LSN would exist here and the snapshot would be skipped on
+	// restart, silently losing the un-acked rows.
+	var checkpoints int
+	require.NoError(t, db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM rpcn.CdcCheckpointCache").Scan(&checkpoints))
+	require.Zero(t, checkpoints, "post-snapshot LSN must not be persisted before snapshot rows are acknowledged")
+
+	// Run 2: restart against the same checkpoint cache. Since run 1 never
+	// acked the snapshot, no LSN was cached, so the snapshot re-runs and every
+	// row is delivered again.
+	t.Log("Launching run 2 (verifying the snapshot re-runs)...")
+	var (
+		readsMu sync.Mutex
+		reads   int
+	)
+	run2Builder := service.NewStreamBuilder()
+	require.NoError(t, run2Builder.AddInputYAML(cfg))
+	require.NoError(t, run2Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run2Builder.AddBatchConsumerFunc(func(_ context.Context, mb service.MessageBatch) error {
+		readsMu.Lock()
+		defer readsMu.Unlock()
+		for _, msg := range mb {
+			if op, _ := msg.MetaGet("operation"); op == "read" {
+				reads++
+			}
+		}
+		return nil
+	}))
+	run2, err := run2Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run2.Resources())
+	go func() {
+		if err := run2.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) {
+			t.Error(err)
+		}
+	}()
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		readsMu.Lock()
+		defer readsMu.Unlock()
+		assert.Equal(c, rowCount, reads, "snapshot should have re-run and re-delivered every row after the crash")
+	}, 5*time.Minute, 500*time.Millisecond)
+	require.NoError(t, run2.StopWithin(time.Second*30))
+}
+
+// TestIntegration_MicrosoftSQLServerCDC_TransactionSplitAcrossBatches verifies
+// that acking a batch which ends mid-transaction never persists that
+// transaction's own start LSN: all rows of a transaction share a start LSN and
+// resume is exclusive (> lsn), so doing so would skip the transaction's
+// remaining rows after a crash. The checkpoint may only advance to the last
+// fully-published transaction boundary. See CON-504.
+func TestIntegration_MicrosoftSQLServerCDC_TransactionSplitAcrossBatches(t *testing.T) {
+	integration.CheckSkip(t)
+
+	connStr, db := mssqlservertest.SetupTestWithMicrosoftSQLServerVersion(t)
+	require.NoError(t, db.CreateTableWithCDCEnabledIfNotExists(t.Context(), "dbo.splittx", "CREATE TABLE dbo.splittx (id INT IDENTITY(1,1) PRIMARY KEY, val INT NOT NULL);"))
+
+	// T1: a single-row transaction, establishing a prior transaction boundary.
+	// T2: four rows committed in ONE transaction - they all share a start LSN.
+	db.MustExec("INSERT INTO dbo.splittx (val) VALUES (101)")
+	db.MustExec("BEGIN TRAN; INSERT INTO dbo.splittx (val) VALUES (102); INSERT INTO dbo.splittx (val) VALUES (103); INSERT INTO dbo.splittx (val) VALUES (104); INSERT INTO dbo.splittx (val) VALUES (105); COMMIT")
+	db.WaitForCDCChanges(t.Context(), 5, "dbo.splittx")
+
+	// batching.count = 2 splits T2 across batches: [T1r1, T2r1], [T2r2, T2r3], ...
+	cfg := fmt.Sprintf(`
+microsoft_sql_server_cdc:
+  connection_string: %s
+  stream_snapshot: false
+  checkpoint_cache: ""
+  include: ["dbo.splittx"]
+  batching:
+    count: 2
+    period: 1h`, connStr)
+
+	// Run 1: ack ONLY the first batch (which ends on T2's first row), block on
+	// everything after it, then crash once the ack's checkpoint write lands.
+	t.Log("Launching run 1 (ack first batch only, simulated crash)...")
+	var firstBatch atomic.Bool
+	firstBatch.Store(true)
+	run1Builder := service.NewStreamBuilder()
+	require.NoError(t, run1Builder.AddInputYAML(cfg))
+	require.NoError(t, run1Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run1Builder.AddBatchConsumerFunc(func(ctx context.Context, _ service.MessageBatch) error {
+		if firstBatch.CompareAndSwap(true, false) {
+			return nil // ack the first batch
+		}
+		<-ctx.Done()
+		return ctx.Err()
+	}))
+	run1, err := run1Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run1.Resources())
+
+	run1Ctx, crash := context.WithCancel(t.Context())
+	run1Done := make(chan struct{})
+	go func() {
+		defer close(run1Done)
+		_ = run1.Run(run1Ctx)
+	}()
+
+	// Wait for the first batch's ack to persist a checkpoint, then crash.
+	require.Eventually(t, func() bool {
+		var checkpoints int
+		if err := db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM rpcn.CdcCheckpointCache").Scan(&checkpoints); err != nil {
+			return false
+		}
+		return checkpoints == 1
+	}, 5*time.Minute, 500*time.Millisecond, "the first batch's ack never persisted a checkpoint")
+	crash()
+	select {
+	case <-run1Done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("run 1 did not stop after the simulated crash")
+	}
+
+	// Run 2: restart. The checkpoint must point at the T1/T2 boundary, so all
+	// of T2 is redelivered - especially rows 102-105's tail (103, 104, 105),
+	// which the pre-fix code skipped by persisting T2's own start LSN.
+	t.Log("Launching run 2 (verifying the split transaction replays in full)...")
+	var (
+		seenMu sync.Mutex
+		seen   = map[int]bool{}
+	)
+	run2Builder := service.NewStreamBuilder()
+	require.NoError(t, run2Builder.AddInputYAML(cfg))
+	require.NoError(t, run2Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run2Builder.AddBatchConsumerFunc(func(_ context.Context, mb service.MessageBatch) error {
+		seenMu.Lock()
+		defer seenMu.Unlock()
+		for _, msg := range mb {
+			var row struct {
+				Val int `json:"val"`
+			}
+			b, err := msg.AsBytes()
+			if err != nil {
+				return err
+			}
+			if err := json.Unmarshal(b, &row); err == nil && row.Val != 0 {
+				seen[row.Val] = true
+			}
+		}
+		return nil
+	}))
+	run2, err := run2Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run2.Resources())
+	go func() {
+		if err := run2.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) {
+			t.Error(err)
+		}
+	}()
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		seenMu.Lock()
+		defer seenMu.Unlock()
+		for _, val := range []int{102, 103, 104, 105} {
+			assert.Truef(c, seen[val], "row val=%d from the split transaction was never redelivered (checkpoint advanced past a partially-delivered transaction)", val)
+		}
+	}, 5*time.Minute, 500*time.Millisecond)
+	require.NoError(t, run2.StopWithin(time.Second*30))
+}
+
 func TestIntegration_MicrosoftSQLServerCDC_ResumesFromCheckpoint(t *testing.T) {
 	integration.CheckSkip(t)
 
diff --git a/internal/impl/mssqlserver/replication/snapshot_test.go b/internal/impl/mssqlserver/replication/snapshot_test.go
index a89ed0d361..2b8be4f569 100644
--- a/internal/impl/mssqlserver/replication/snapshot_test.go
+++ b/internal/impl/mssqlserver/replication/snapshot_test.go
@@ -158,6 +158,10 @@ func (m *publisherStub) Publish(_ context.Context, msg replication.MessageEvent)
 	return nil
 }
 
+func (*publisherStub) CheckpointWindow(context.Context, replication.LSN) error {
+	return nil
+}
+
 func (m *publisherStub) count() int {
 	m.mu.Lock()
 	defer m.mu.Unlock()
diff --git a/internal/impl/mssqlserver/replication/stream.go b/internal/impl/mssqlserver/replication/stream.go
index 39145380e3..1d03720dba 100644
--- a/internal/impl/mssqlserver/replication/stream.go
+++ b/internal/impl/mssqlserver/replication/stream.go
@@ -332,9 +332,37 @@ func mapScannedValue(val any, colType *sql.ColumnType) any {
 	return val
 }
 
+// txnBoundary tracks transaction boundaries in the globally LSN-ordered row
+// stream. All rows of one transaction share a __$start_lsn, so an LSN change
+// between consecutive rows proves the previous transaction is fully read (and,
+// because rows are published synchronously in read order, fully published).
+type txnBoundary struct {
+	prev         LSN
+	lastComplete LSN
+}
+
+// Observe records the current row's LSN and returns the start LSN of the most
+// recent transaction whose rows have all been observed — empty until the first
+// boundary is crossed.
+func (t *txnBoundary) Observe(lsn LSN) LSN {
+	if len(t.prev) != 0 && !bytes.Equal(lsn, t.prev) {
+		t.lastComplete = t.prev
+	}
+	// Copy: the iterator may reuse the underlying array on the next scan.
+	t.prev = append(t.prev[:0:0], lsn...)
+	return t.lastComplete
+}
+
 // ChangePublisher is responsible for handling and processing of a replication.MessageEvent.
 type ChangePublisher interface {
 	Publish(ctx context.Context, msg MessageEvent) error
+	// CheckpointWindow records that every transaction up to and including lsn
+	// has been fully published (a polling window drained). Once all batches
+	// published before this call are acknowledged, lsn may be persisted as the
+	// resume position — without it the final transaction of a burst would only
+	// be checkpointed when a later transaction appears, re-delivering it on
+	// every restart of an idle stream.
+	CheckpointWindow(ctx context.Context, lsn LSN) error
 }
 
 // ChangeTableStream tracks and streams all change events from the configured change
@@ -365,6 +393,9 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 		startLSN LSN // load last checkpoint; nil means start from beginning in tables
 		endLSN   LSN // often set to fn_cdc_get_max_lsn(); nil means no upper bound
 		lastLSN  LSN
+		// boundary computes each row's CheckpointLSN: the last transaction
+		// whose rows are all published, the only safe resume position.
+		boundary txnBoundary
 	)
 
 	if len(startPos) != 0 {
@@ -415,13 +446,14 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 			cur := item.iter.current
 
 			msg := MessageEvent{
-				Table:       item.iter.table.Name,
-				Schema:      item.iter.table.Schema,
-				Data:        cur.columns,
-				LSN:         cur.startLSN,
-				Operation:   cur.operation.String(),
-				ColumnNames: item.iter.userColNames,
-				ColumnTypes: item.iter.userColTypes,
+				Table:         item.iter.table.Name,
+				Schema:        item.iter.table.Schema,
+				Data:          cur.columns,
+				LSN:           cur.startLSN,
+				CheckpointLSN: boundary.Observe(cur.startLSN),
+				Operation:     cur.operation.String(),
+				ColumnNames:   item.iter.userColNames,
+				ColumnTypes:   item.iter.userColTypes,
 			}
 
 			if err := r.publisher.Publish(ctx, msg); err != nil {
@@ -450,6 +482,12 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 
 		if len(lastLSN) != 0 {
 			if !bytes.Equal(startLSN, lastLSN) {
+				// The window is drained: every transaction <= lastLSN is fully
+				// published, so the exact end position may be checkpointed once
+				// the window's batches are acked.
+				if err := r.publisher.CheckpointWindow(ctx, lastLSN); err != nil {
+					return fmt.Errorf("checkpointing window end: %w", err)
+				}
 				startLSN = lastLSN
 			} else {
 				r.log.Debug("No more changes across all change tables, backing off...")
diff --git a/internal/impl/mssqlserver/replication/stream_message.go b/internal/impl/mssqlserver/replication/stream_message.go
index 27143a3c05..ffab898aed 100644
--- a/internal/impl/mssqlserver/replication/stream_message.go
+++ b/internal/impl/mssqlserver/replication/stream_message.go
@@ -83,11 +83,17 @@ func (op OpType) String() string {
 
 // MessageEvent represents a single change from Table's change table in the database.
 type MessageEvent struct {
-	LSN       LSN    `json:"start_lsn"`
-	Operation string `json:"operation"`
-	Schema    string `json:"schema"`
-	Table     string `json:"table"`
-	Data      any    `json:"data"`
+	LSN LSN `json:"start_lsn"`
+	// CheckpointLSN is the start LSN of the most recent transaction whose rows
+	// have all been published — the only value safe to persist as a resume
+	// position (resume is exclusive and all rows of a transaction share a
+	// start LSN). Empty for snapshot rows and until the first transaction
+	// boundary is observed.
+	CheckpointLSN LSN    `json:"-"`
+	Operation     string `json:"operation"`
+	Schema        string `json:"schema"`
+	Table         string `json:"table"`
+	Data          any    `json:"data"`
 
 	// ColumnNames and ColumnTypes carry user-defined column metadata (excluding
 	// MSSQL system columns with __$ prefix). They are used to build schema
diff --git a/internal/impl/mssqlserver/replication/stream_test.go b/internal/impl/mssqlserver/replication/stream_test.go
new file mode 100644
index 0000000000..ce41aae709
--- /dev/null
+++ b/internal/impl/mssqlserver/replication/stream_test.go
@@ -0,0 +1,41 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package replication
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestTxnBoundaryObserve(t *testing.T) {
+	var b txnBoundary
+
+	// Sequence AAABBC: the last complete transaction only advances when the
+	// LSN changes, and always lags one transaction behind the current row.
+	observations := []struct {
+		lsn          string
+		lastComplete string // "" = no complete transaction yet
+	}{
+		{"A", ""},
+		{"A", ""},
+		{"A", ""},
+		{"B", "A"},
+		{"B", "A"},
+		{"C", "B"},
+	}
+	for i, o := range observations {
+		got := b.Observe(LSN(o.lsn))
+		if o.lastComplete == "" {
+			require.Emptyf(t, got, "observation %d (lsn %s)", i, o.lsn)
+		} else {
+			require.Equalf(t, o.lastComplete, string(got), "observation %d (lsn %s)", i, o.lsn)
+		}
+	}
+}