diff --git a/.github/workflows/jepsen-test.yml b/.github/workflows/jepsen-test.yml
index 2061f6e46..a002ddf04 100644
--- a/.github/workflows/jepsen-test.yml
+++ b/.github/workflows/jepsen-test.yml
@@ -64,6 +64,7 @@ jobs:
           go mod download
           go build -o /tmp/elastickv-bin .
           go build -o /tmp/elastickv-admin-bin ./cmd/elastickv-admin
+          go build -o /tmp/elastickv-raftadmin-bin ./cmd/raftadmin
       - name: Warm Leiningen Maven cache
         working-directory: jepsen
         run: |
@@ -90,7 +91,8 @@ jobs:
           BOOTSTRAP_MEMBERS="n1=127.0.0.1:50051,n2=127.0.0.1:50052,n3=127.0.0.1:50053"
           RAFT_REDIS_MAP="127.0.0.1:50051=127.0.0.1:63791,127.0.0.1:50052=127.0.0.1:63792,127.0.0.1:50053=127.0.0.1:63793"
           RAFT_S3_MAP="127.0.0.1:50051=127.0.0.1:63901,127.0.0.1:50052=127.0.0.1:63902,127.0.0.1:50053=127.0.0.1:63903"
-          RAFT_DYNAMO_MAP="127.0.0.1:50051=127.0.0.1:63801,127.0.0.1:50052=127.0.0.1:63802,127.0.0.1:50053=127.0.0.1:63803"
+          # No raftDynamoMap: DynamoDB workloads must exercise follower gRPC
+          # forwarding and leader-issued seals instead of HTTP leader proxying.
           RAFT_SQS_MAP="127.0.0.1:50051=127.0.0.1:63501,127.0.0.1:50052=127.0.0.1:63502,127.0.0.1:50053=127.0.0.1:63503"
 
           # Encrypted leg: one random owner-only file KEK shared by the
@@ -132,7 +134,6 @@ jobs:
               --raftBootstrapMembers "$BOOTSTRAP_MEMBERS" \
               --raftRedisMap "$RAFT_REDIS_MAP" \
               --raftS3Map "$RAFT_S3_MAP" \
-              --raftDynamoMap "$RAFT_DYNAMO_MAP" \
               --raftSqsMap "$RAFT_SQS_MAP" \
               "${ENC_ARGS[@]}" \
               > "/tmp/elastickv-demo-n${node}.log" 2>&1 &
@@ -183,6 +184,26 @@ jobs:
         timeout-minutes: 3
         run: |
           timeout 120 ~/lein run -m elastickv.dynamodb-workload --local --time-limit 5 --rate 5 --concurrency 5 --dynamo-ports 63801,63802,63803 --host 127.0.0.1
+      - name: Run DynamoDB seal workload with recorded leader changes
+        working-directory: jepsen
+        run: |
+          printf 'run_sha=%s\n' "$GITHUB_SHA" > /tmp/dynamodb-seal-source.txt
+          timeout 150 ~/lein run -m elastickv.dynamodb-seal-workload --local \
+            --time-limit 15 --rate 30 --concurrency 9 --max-txn-length 4 \
+            --dynamo-ports 63801,63802,63803 --host 127.0.0.1 \
+            --raftadmin-bin /tmp/elastickv-raftadmin-bin \
+            --raft-addresses 127.0.0.1:50051,127.0.0.1:50052,127.0.0.1:50053 \
+            --leader-change-interval 2
+      - name: Upload DynamoDB seal acceptance history
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: dynamodb-seal-encryption-${{ matrix.encryption }}
+          path: |
+            jepsen/store/elastickv-dynamodb-snapshot-seal/
+            /tmp/dynamodb-seal-source.txt
+          retention-days: 7
+          if-no-files-found: ignore
       - name: Run DynamoDB per-type Jepsen workloads against elastickv
         working-directory: jepsen
         timeout-minutes: 10
diff --git a/adapter/dynamodb_item_write.go b/adapter/dynamodb_item_write.go
index 2c4d21b86..a0be1ce41 100644
--- a/adapter/dynamodb_item_write.go
+++ b/adapter/dynamodb_item_write.go
@@ -495,7 +495,7 @@ func validatePutItemAdminPathKey(in putItemInput, schema *dynamoTableSchema) err
 
 func (d *DynamoDBServer) commitItemWrite(ctx context.Context, readTimestamp kv.ReadTimestamp, req *kv.OperationGroup[kv.OP]) error {
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-	_, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req)
+	_, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req)
 	if err != nil {
 		return errors.WithStack(err)
 	}
diff --git a/adapter/dynamodb_locks.go b/adapter/dynamodb_locks.go
index df2375c79..75b1bcd87 100644
--- a/adapter/dynamodb_locks.go
+++ b/adapter/dynamodb_locks.go
@@ -391,8 +391,9 @@ func (d *DynamoDBServer) tryTxnWriteTimestamp(ctx context.Context, label string,
 		var resp *kv.CoordinateResponse
 		if sealTS == 0 {
 			resp, err = kv.DispatchWithLeaderIssuedCommitTS(floor.WithDispatchVoucher(ctx), d.coordinator, request)
+			err = normalizeDynamoWriteDispatchError(err)
 		} else {
-			resp, err = kv.DispatchWithReadTimestamp(floor.WithDispatchVoucher(ctx), d.coordinator, request)
+			resp, err = d.dispatchWriteWithReadTimestamp(floor.WithDispatchVoucher(ctx), request)
 		}
 		if err != nil {
 			return kv.ReadTimestamp{}, errors.WithStack(err)
diff --git a/adapter/dynamodb_migration.go b/adapter/dynamodb_migration.go
index 7c6f5bb8e..c3f86d15c 100644
--- a/adapter/dynamodb_migration.go
+++ b/adapter/dynamodb_migration.go
@@ -97,7 +97,7 @@ func (d *DynamoDBServer) startLegacyTableKeyMigration(
 	}
 	req.StartTS = readTS
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
+	if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
 		return errors.WithStack(err)
 	}
 	return nil
@@ -186,7 +186,7 @@ func (d *DynamoDBServer) migrateLegacyItem(
 		req.StartTS = readTS
 		req.ReadKeys = append(req.ReadKeys, sealKey)
 		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
+		if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err == nil {
 			return nil
 		} else if !isRetryableTransactWriteError(err) {
 			return errors.WithStack(err)
@@ -338,7 +338,7 @@ func (d *DynamoDBServer) finalizeLegacyTableMigration(ctx context.Context, schem
 		},
 	}
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
+	if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
 		return errors.WithStack(err)
 	}
 	d.launchDeletedTableCleanup(cleanup, schema.TableName, oldGeneration)
diff --git a/adapter/dynamodb_schema.go b/adapter/dynamodb_schema.go
index 96d12c120..a05a715b2 100644
--- a/adapter/dynamodb_schema.go
+++ b/adapter/dynamodb_schema.go
@@ -205,7 +205,7 @@ func (d *DynamoDBServer) createTableWithRetry(ctx context.Context, tableName str
 		}
 		req.StartTS = readTS
 		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
+		if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err == nil {
 			return nil
 		} else if !isRetryableTransactWriteError(err) {
 			return errors.WithStack(err)
@@ -327,7 +327,7 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
 			},
 		}
 		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
+		if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
 			if !isRetryableTransactWriteError(err) {
 				return errors.WithStack(err)
 			}
@@ -386,6 +386,7 @@ func (d *DynamoDBServer) dispatchDeletedTableCleanupPrefix(ctx context.Context,
 				{Op: kv.DelPrefix, Key: prefix},
 			},
 		})
+		err = normalizeDynamoWriteDispatchError(err)
 		if err == nil {
 			return nil
 		}
@@ -412,7 +413,7 @@ func (d *DynamoDBServer) dispatchDeleteBatch(ctx context.Context, keys [][]byte)
 	}
 	_, err := d.coordinator.Dispatch(ctx, req)
 	if err != nil {
-		return errors.WithStack(err)
+		return errors.WithStack(normalizeDynamoWriteDispatchError(err))
 	}
 	return nil
 }
diff --git a/adapter/dynamodb_snapshot_seal_benchmark_test.go b/adapter/dynamodb_snapshot_seal_benchmark_test.go
new file mode 100644
index 000000000..9b0f4d752
--- /dev/null
+++ b/adapter/dynamodb_snapshot_seal_benchmark_test.go
@@ -0,0 +1,102 @@
+package adapter
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+type sealBenchmarkCoordinator struct {
+	kv.Coordinator
+	seals     atomic.Uint64
+	conflicts atomic.Uint64
+}
+
+func (c *sealBenchmarkCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if isDynamoSnapshotSealRequest(req) {
+		c.seals.Add(1)
+	}
+	resp, err := c.Coordinator.Dispatch(ctx, req)
+	if errors.Is(err, store.ErrWriteConflict) {
+		c.conflicts.Add(1)
+	}
+	return resp, err
+}
+
+// A single durable Raft voter isolates the extra seal/OCC cost. This is not a
+// networked quorum or a production throughput claim. Separate frontends avoid
+// serializing the workload with a shared server's item-lock stripes.
+func BenchmarkDynamoDBSnapshotSealRealRaft(b *testing.B) {
+	for _, mode := range []string{"independent", "hot_item", "collision"} {
+		b.Run(mode, func(b *testing.B) {
+			ctx := context.Background()
+			st := store.NewMVCCStore()
+			defer st.Close()
+			server, _, stop := sealRaftRestartableServer(b, b.TempDir(), st)
+			defer stop()
+			schema := dedupItemTable()
+			require.NoError(b, server.createTableWithRetry(ctx, schema.TableName, schema))
+			keys := sealBenchmarkKeys(b, schema, mode)
+			for _, key := range keys {
+				_, err := server.putItemWithRetry(ctx, putItemInput{
+					TableName: schema.TableName, Item: map[string]attributeValue{"pk": newStringAttributeValue(key)},
+				})
+				require.NoError(b, err)
+			}
+			coord := &sealBenchmarkCoordinator{Coordinator: server.coordinator}
+			var worker atomic.Uint64
+			b.ReportAllocs()
+			b.ResetTimer()
+			b.RunParallel(func(pb *testing.PB) {
+				local := NewDynamoDBServer(nil, st, coord)
+				key := keys[(worker.Add(1)-1)%uint64(len(keys))]
+				in := putItemInput{TableName: schema.TableName, Item: map[string]attributeValue{
+					"pk": newStringAttributeValue(key), "value": newStringAttributeValue("constant-size replacement"),
+				}}
+				for pb.Next() {
+					if _, err := local.putItemWithRetry(ctx, in); err != nil {
+						b.Error(err)
+						return
+					}
+				}
+			})
+			b.StopTimer()
+			b.ReportMetric(float64(coord.seals.Load())/float64(b.N), "seals/op")
+			b.ReportMetric(float64(coord.conflicts.Load())/float64(b.N), "conflicts/op")
+		})
+	}
+}
+
+func sealBenchmarkKeys(b *testing.B, schema *dynamoTableSchema, mode string) []string {
+	b.Helper()
+	if mode == "hot_item" {
+		return []string{"hot"}
+	}
+	const count = 4
+	buckets := make(map[string][]string)
+	independent := make([]string, 0, count)
+	for i := range 1 << 20 {
+		key := fmt.Sprintf("seal-bench-%d", i)
+		item, err := schema.itemKeyFromAttributes(map[string]attributeValue{"pk": newStringAttributeValue(key)})
+		require.NoError(b, err)
+		seal := string(dynamoItemSnapshotSealKey(schema.TableName, schema.Generation, item))
+		if mode == "independent" && len(buckets[seal]) == 0 {
+			independent = append(independent, key)
+			if len(independent) == count {
+				return independent
+			}
+		}
+		buckets[seal] = append(buckets[seal], key)
+		if mode == "collision" && len(buckets[seal]) == count {
+			return buckets[seal]
+		}
+	}
+	b.Fatal("could not construct seal bucket fixture")
+	return nil
+}
diff --git a/adapter/dynamodb_snapshot_seal_raft_test.go b/adapter/dynamodb_snapshot_seal_raft_test.go
new file mode 100644
index 000000000..7a79bfcfd
--- /dev/null
+++ b/adapter/dynamodb_snapshot_seal_raft_test.go
@@ -0,0 +1,424 @@
+package adapter
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/bootjp/elastickv/distribution"
+	"github.com/bootjp/elastickv/internal/raftengine"
+	etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd"
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	json "github.com/goccy/go-json"
+	"github.com/stretchr/testify/require"
+)
+
+// Pause after the adapter allocates C, but before the real coordinator proposes
+// it. No store/FSM/transport is mocked: the later seal must reject this plan.
+type delayedSealRaftCoordinator struct {
+	kv.Coordinator
+	blocked       atomic.Bool
+	entered       chan uint64
+	release       chan struct{}
+	result        chan error
+	unstamped     bool
+	freshCommitTS uint64
+}
+
+func (c *delayedSealRaftCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if req == nil || !req.IsTxn || (req.CommitTS == 0 && !c.unstamped) || isDynamoSnapshotSealRequest(req) || !c.blocked.CompareAndSwap(false, true) {
+		return c.Coordinator.Dispatch(ctx, req)
+	}
+	stamp := req.CommitTS
+	if stamp == 0 {
+		stamp = req.StartTS
+	}
+	c.entered <- stamp
+	select {
+	case <-c.release:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	return c.dispatchReleased(ctx, req)
+}
+
+func (c *delayedSealRaftCoordinator) dispatchReleased(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if c.freshCommitTS != 0 {
+		req.CommitTS = c.freshCommitTS
+	}
+	resp, err := c.Coordinator.Dispatch(ctx, req)
+	c.result <- err
+	return resp, err
+}
+
+func newDelayedSealRaftServer(t *testing.T, server *DynamoDBServer, unstamped bool) (*DynamoDBServer, *delayedSealRaftCoordinator, func()) {
+	t.Helper()
+	coord := &delayedSealRaftCoordinator{
+		Coordinator: server.coordinator,
+		entered:     make(chan uint64, 1),
+		release:     make(chan struct{}),
+		result:      make(chan error, 1),
+		unstamped:   unstamped,
+	}
+	var once sync.Once
+	release := func() { once.Do(func() { close(coord.release) }) }
+	t.Cleanup(release)
+	return NewDynamoDBServer(nil, server.store, coord), coord, release
+}
+
+func sealRaftSeed(t *testing.T, ctx context.Context, server *DynamoDBServer, schema *dynamoTableSchema) {
+	t.Helper()
+	require.NoError(t, server.createTableWithRetry(ctx, schema.TableName, schema))
+	_, err := server.putItemWithRetry(ctx, putItemInput{TableName: schema.TableName, Item: map[string]attributeValue{
+		"pk": newStringAttributeValue(dedupTestKey), "val": numberListAttr("1"),
+	}})
+	require.NoError(t, err)
+}
+
+func sealRaftAppend(table, value string) updateItemInput {
+	in := appendListInput()
+	in.TableName = table
+	in.ExpressionAttributeValues[":val"] = numberListAttr(value)
+	return in
+}
+
+func sealRaftTxnUpdate(table string) transactWriteItem {
+	in := sealRaftAppend(table, "3")
+	return transactWriteItem{Update: &transactUpdateInput{
+		TableName: in.TableName, Key: in.Key, UpdateExpression: in.UpdateExpression,
+		ExpressionAttributeNames: in.ExpressionAttributeNames, ExpressionAttributeValues: in.ExpressionAttributeValues,
+	}}
+}
+
+func sealRaftDelayedAppend(ctx context.Context, server *DynamoDBServer, table string) <-chan error {
+	done := make(chan error, 1)
+	go func() {
+		_, err := server.updateItemWithRetry(ctx, sealRaftAppend(table, "2"))
+		done <- err
+	}()
+	return done
+}
+
+func sealRaftReceive[T any](t *testing.T, ctx context.Context, ch <-chan T) T {
+	t.Helper()
+	select {
+	case value := <-ch:
+		return value
+	case <-ctx.Done():
+		t.Fatalf("waiting for delayed Raft write: %v", ctx.Err())
+		var zero T
+		return zero
+	}
+}
+
+func sealRaftHTTP(t *testing.T, ctx context.Context, address, action string, input any) {
+	t.Helper()
+	require.NoError(t, sealRaftHTTPCall(ctx, address, action, input))
+}
+
+func sealRaftHTTPCall(ctx context.Context, address, action string, input any) error {
+	body, err := json.Marshal(input)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://"+address+"/", bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("X-Amz-Target", "DynamoDB_20120810."+action)
+	req.Header.Set("Content-Type", "application/x-amz-json-1.0")
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	response, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("DynamoDB HTTP %d: %s", resp.StatusCode, response)
+	}
+	return nil
+}
+
+func TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress(t *testing.T) {
+	for _, restamp := range []bool{false, true} {
+		t.Run(map[bool]string{false: "delayed-commit-watermark", true: "fresh-commit-stale-read-seal"}[restamp], func(t *testing.T) {
+			testSnapshotSealRaftLeaderChange(t, restamp)
+		})
+	}
+}
+
+func testSnapshotSealRaftLeaderChange(t *testing.T, restamp bool) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+	schema := dedupItemTable()
+	leader := nodes[0].dynamoServer
+	sealRaftSeed(t, ctx, leader, schema)
+	delayed, coord, release := newDelayedSealRaftServer(t, leader, false)
+	defer release()
+	done := sealRaftDelayedAppend(ctx, delayed, schema.TableName)
+	commitTS := sealRaftReceive(t, ctx, coord.entered)
+	sealRaftAdvanceWatermark(t, ctx, leader.coordinator, commitTS)
+
+	admin, ok := nodes[0].engine.(raftengine.Admin)
+	require.True(t, ok)
+	require.NoError(t, admin.TransferLeadershipToServer(ctx, "1", nodes[1].raftAddress))
+	require.Eventually(t, func() bool {
+		return nodes[1].engine.State() == raftengine.StateLeader &&
+			nodes[0].engine.State() == raftengine.StateFollower && nodes[2].engine.State() == raftengine.StateFollower &&
+			nodes[0].engine.Leader().Address == nodes[1].raftAddress &&
+			nodes[2].engine.Leader().Address == nodes[1].raftAddress
+	}, 5*time.Second, 10*time.Millisecond)
+	follower := nodes[2].dynamoServer
+	require.Empty(t, follower.leaderDynamo, "HTTP forwarding would hide the follower timestamp path")
+	second, secondCoord, releaseSecond := newDelayedSealRaftServer(t, follower, true)
+	defer releaseSecond()
+	httpServer := httptest.NewServer(http.HandlerFunc(second.handle))
+	defer func() {
+		releaseSecond()
+		cancel()
+		httpServer.Close()
+	}()
+	txnDone := make(chan error, 1)
+	go func() {
+		txnDone <- sealRaftHTTPCall(ctx, strings.TrimPrefix(httpServer.URL, "http://"), "TransactWriteItems", transactWriteItemsInput{
+			TransactItems: []transactWriteItem{sealRaftTxnUpdate(schema.TableName)},
+		})
+	}()
+	var sealed uint64
+	select {
+	case sealed = <-secondCoord.entered:
+	case err := <-txnDone:
+		t.Fatalf("transaction finished before the planned-write barrier: %v", err)
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	require.Greater(t, sealed, commitTS)
+	if restamp {
+		// Keep the stale read plan but allocate its commit timestamp after R.
+		// The global apply-order fence must not hide whether the seal rejects it.
+		fresh, err := kv.NextTimestampAfterThrough(ctx, nodes[1].dynamoServer.coordinator, sealed, "seal regression: fresh commit for stale read plan")
+		require.NoError(t, err)
+		require.Greater(t, fresh, sealed)
+		coord.freshCommitTS = fresh
+	}
+	release()
+	conflict := sealRaftReceive(t, ctx, coord.result)
+	require.ErrorContains(t, conflict, "write conflict", "the delayed write must never be acknowledged")
+	if restamp {
+		require.ErrorContains(t, conflict, dynamoSealPrefix, "a fresh commit timestamp bypasses the global fence, so the seal itself must reject the stale read")
+	}
+	require.NoError(t, sealRaftReceive(t, ctx, done))
+	releaseSecond()
+	require.NoError(t, sealRaftReceive(t, ctx, txnDone))
+	require.Equal(t, []string{"1", "2", "3"}, readListValues(t, nodes[1].dynamoServer, schema))
+	// Also exercise the follower's single-item legacy timestamp allocation.
+	sealRaftHTTP(t, ctx, nodes[2].dynamoAddress, "UpdateItem", sealRaftAppend(schema.TableName, "4"))
+	require.Equal(t, []string{"1", "2", "3", "4"}, readListValues(t, nodes[1].dynamoServer, schema))
+	t.Logf("delayed C=%d, successor seal R=%d, old write rejected before transaction commit", commitTS, sealed)
+}
+
+func sealRaftAdvanceWatermark(t *testing.T, ctx context.Context, coord kv.Coordinator, pending uint64) {
+	t.Helper()
+	ts, err := kv.NextTimestampAfterThrough(ctx, coord, pending, "advance unrelated watermark")
+	require.NoError(t, err)
+	require.Greater(t, ts, pending)
+	_, err = coord.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
+		CommitTS: ts,
+		Elems:    []*kv.Elem[kv.OP]{{Op: kv.Put, Key: []byte("watermark"), Value: []byte("x")}},
+	})
+	require.NoError(t, err)
+}
+
+func TestDynamoDBRealRaftFollowerConcurrentCreateTable(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+	input := map[string]any{
+		"TableName":             "concurrent_seal_setup",
+		"AttributeDefinitions":  []map[string]string{{"AttributeName": "pk", "AttributeType": "S"}},
+		"KeySchema":             []map[string]string{{"AttributeName": "pk", "KeyType": "HASH"}},
+		"ProvisionedThroughput": map[string]int{"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+	}
+	const clients = 9
+	done := make(chan error, clients)
+	for i := range clients {
+		address := nodes[1+i%2].dynamoAddress
+		go func() { done <- sealRaftHTTPCall(ctx, address, "CreateTable", input) }()
+	}
+	created := 0
+	for range clients {
+		err := sealRaftReceive(t, ctx, done)
+		if err == nil {
+			created++
+		} else {
+			require.ErrorContains(t, err, "ResourceInUseException", "concurrent setup must not leak a forwarded OCC conflict")
+		}
+	}
+	require.Equal(t, 1, created)
+}
+
+func TestDynamoDBSnapshotSealRealRaftTwoDataGroups(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	left, right := dedupItemTable(), dedupItemTable()
+	left.TableName, right.TableName = "alpha", "beta"
+	split := kv.RouteKey(dynamoTableMetaKey(right.TableName))
+	require.Negative(t, bytes.Compare(kv.RouteKey(dynamoTableMetaKey(left.TableName)), split))
+	routes := distribution.NewEngine()
+	routes.UpdateRoute(nil, split, 1)
+	routes.UpdateRoute(split, nil, 2)
+	hlc := kv.NewHLC()
+	st1, st2 := store.NewMVCCStore(), store.NewMVCCStore()
+	eng1, stop1 := newSingleRaftForDistributionE2E(t, "seal-group-1", kv.NewKvFSMWithHLC(st1, hlc))
+	defer stop1()
+	eng2, stop2 := newSingleRaftForDistributionE2E(t, "seal-group-2", kv.NewKvFSMWithHLC(st2, hlc))
+	defer stop2()
+	groups := map[uint64]*kv.ShardGroup{
+		1: {Engine: eng1, Store: st1, Txn: kv.NewLeaderProxyWithEngine(eng1)},
+		2: {Engine: eng2, Store: st2, Txn: kv.NewLeaderProxyWithEngine(eng2)},
+	}
+	st := kv.NewShardStore(routes, groups)
+	defer func() {
+		stop1()
+		stop2()
+		require.NoError(t, st.Close())
+	}()
+	coordinator := kv.NewShardedCoordinator(routes, groups, 1, hlc, st)
+	server := NewDynamoDBServer(nil, st, coordinator)
+	sealRaftSeed(t, ctx, server, left)
+	sealRaftSeed(t, ctx, server, right)
+	delayed, coord, release := newDelayedSealRaftServer(t, server, false)
+	defer release()
+	done := sealRaftDelayedAppend(ctx, delayed, right.TableName)
+	commitTS := sealRaftReceive(t, ctx, coord.entered)
+	sealRaftAdvanceWatermark(t, ctx, coordinator, commitTS)
+
+	// This independent frontend has no shared per-item locks with the writer.
+	second, secondCoord, releaseSecond := newDelayedSealRaftServer(t, server, true)
+	defer releaseSecond()
+	input := transactWriteItemsInput{TransactItems: []transactWriteItem{
+		sealRaftTxnUpdate(left.TableName), sealRaftTxnUpdate(right.TableName),
+	}}
+	body, err := json.Marshal(input)
+	require.NoError(t, err)
+	recorder := httptest.NewRecorder()
+	txnDone := make(chan struct{})
+	go func() {
+		second.transactWriteItems(recorder, httptest.NewRequestWithContext(ctx, http.MethodPost, "/", bytes.NewReader(body)))
+		close(txnDone)
+	}()
+	sealed := sealRaftReceive(t, ctx, secondCoord.entered)
+	require.Greater(t, sealed, commitTS)
+	key := map[string]attributeValue{"pk": newStringAttributeValue(dedupTestKey)}
+	leftKey, err := left.itemKeyFromAttributes(key)
+	require.NoError(t, err)
+	rightKey, err := right.itemKeyFromAttributes(key)
+	require.NoError(t, err)
+	seal1, seal2 := dynamoItemSnapshotSealKey(left.TableName, 1, leftKey), dynamoItemSnapshotSealKey(right.TableName, 1, rightKey)
+	ts1, found, err := st1.LatestCommitTS(ctx, seal1)
+	require.NoError(t, err)
+	require.True(t, found)
+	ts2, found, err := st2.LatestCommitTS(ctx, seal2)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Greater(t, ts1, commitTS)
+	require.Equal(t, ts1, ts2, "both real Raft groups must seal at the same R")
+	release()
+	require.ErrorIs(t, sealRaftReceive(t, ctx, coord.result), store.ErrWriteConflict)
+	require.NoError(t, sealRaftReceive(t, ctx, done))
+	releaseSecond()
+	sealRaftReceive(t, ctx, txnDone)
+	require.Equal(t, http.StatusOK, recorder.Code, "%s", recorder.Body.String())
+	require.Equal(t, []string{"1", "3"}, readListValues(t, second, left))
+	require.Equal(t, []string{"1", "2", "3"}, readListValues(t, second, right))
+	t.Logf("two data groups: delayed C=%d, shared R=%d", commitTS, ts1)
+}
+
+func TestDynamoDBSnapshotSealRealRaftSnapshotRestart(t *testing.T) {
+	t.Setenv("ELASTICKV_RAFT_SNAPSHOT_COUNT", "1")
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	dir := t.TempDir()
+	st := store.NewMVCCStore()
+	server, engine, stop := sealRaftRestartableServer(t, dir, st)
+	schema := dedupItemTable()
+	sealRaftSeed(t, ctx, server, schema)
+	key, err := schema.itemKeyFromAttributes(map[string]attributeValue{"pk": newStringAttributeValue(dedupTestKey)})
+	require.NoError(t, err)
+	seal := dynamoItemSnapshotSealKey(schema.TableName, 1, key)
+	before, found, err := st.LatestCommitTS(ctx, seal)
+	require.NoError(t, err)
+	require.True(t, found)
+	stamp, err := server.beginTxnWriteTimestamp(ctx, "restart seal", [][]byte{seal})
+	require.NoError(t, err)
+	sealed := stamp.Timestamp()
+	require.Greater(t, sealed, before)
+	index := engine.Status().AppliedIndex
+	require.Eventually(t, func() bool { return engine.Status().LastSnapshotIndex >= index }, 5*time.Second, 10*time.Millisecond)
+	stop()
+	require.NoError(t, st.Close())
+
+	// Reopen the same Raft directory with a genuinely empty state machine.
+	// The item and seal below must come from its durable snapshot/WAL, not RAM.
+	restored := store.NewMVCCStore()
+	restarted, _, stopRestarted := sealRaftRestartableServer(t, dir, restored)
+	defer restored.Close()
+	defer stopRestarted()
+	sealTS, found, err := restored.LatestCommitTS(ctx, seal)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Equal(t, sealed, sealTS)
+	require.Equal(t, []string{"1"}, readListValues(t, restarted, schema))
+	_, err = restarted.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
+		IsTxn: true, StartTS: before, ReadKeys: [][]byte{seal},
+		Elems: []*kv.Elem[kv.OP]{{Op: kv.Put, Key: key, Value: []byte("stale plan")}},
+	})
+	require.ErrorIs(t, err, store.ErrWriteConflict)
+	_, err = restarted.updateItemWithRetry(ctx, sealRaftAppend(schema.TableName, "2"))
+	require.NoError(t, err)
+	require.Equal(t, []string{"1", "2"}, readListValues(t, restarted, schema))
+}
+
+func sealRaftRestartableServer(t testing.TB, dir string, st store.MVCCStore) (*DynamoDBServer, raftengine.Engine, func()) {
+	t.Helper()
+	hlc := kv.NewHLC()
+	factory := etcdraftengine.NewFactory(etcdraftengine.FactoryConfig{
+		TickInterval: 10 * time.Millisecond, HeartbeatTick: 1, ElectionTick: 10,
+		MaxSizePerMsg: 1 << 20, MaxInflightMsg: 256,
+	})
+	result, err := factory.Create(raftengine.FactoryConfig{
+		LocalID: "seal-restart", LocalAddress: "seal-restart", DataDir: dir, Bootstrap: true,
+		StateMachine: kv.NewKvFSMWithHLC(st, hlc),
+	})
+	require.NoError(t, err)
+	var once sync.Once
+	stop := func() {
+		once.Do(func() {
+			require.NoError(t, result.Engine.Close())
+			if result.Close != nil {
+				require.NoError(t, result.Close())
+			}
+		})
+	}
+	t.Cleanup(stop)
+	require.Eventually(t, func() bool { return result.Engine.State() == raftengine.StateLeader }, 5*time.Second, 10*time.Millisecond)
+	coord := kv.NewCoordinatorWithEngine(kv.NewTransactionWithProposer(result.Engine), result.Engine, kv.WithHLC(hlc))
+	return NewDynamoDBServer(nil, st, coord), result.Engine, stop
+}
diff --git a/adapter/dynamodb_transact.go b/adapter/dynamodb_transact.go
index ba371af6b..9d8553d9f 100644
--- a/adapter/dynamodb_transact.go
+++ b/adapter/dynamodb_transact.go
@@ -752,7 +752,7 @@ func (d *DynamoDBServer) runTransactWriteAttempt(
 		return true, nil
 	}
 	dispatchCtx := reqs.readTimestamp.WithDispatchVoucher(ctx)
-	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, reqs.group); err != nil {
+	if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, reqs.group); err != nil {
 		wrapped := errors.WithStack(err)
 		if !isRetryableTransactWriteError(err) {
 			return false, wrapped
diff --git a/adapter/dynamodb_write_error.go b/adapter/dynamodb_write_error.go
new file mode 100644
index 000000000..4caaf4554
--- /dev/null
+++ b/adapter/dynamodb_write_error.go
@@ -0,0 +1,40 @@
+package adapter
+
+import (
+	"context"
+	"strings"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func (d *DynamoDBServer) dispatchWriteWithReadTimestamp(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	resp, err := kv.DispatchWithReadTimestamp(ctx, d.coordinator, req)
+	return resp, normalizeDynamoWriteDispatchError(err)
+}
+
+// Forwarded OCC conflicts lose their Go error chain. Restore only the known
+// wire envelope so item retries retain the existing exact-timestamp dedup path.
+func normalizeDynamoWriteDispatchError(err error) error {
+	if err == nil {
+		return nil
+	}
+	type statusCarrier interface{ GRPCStatus() *status.Status }
+	var carrier statusCarrier
+	if !errors.As(err, &carrier) {
+		return err
+	}
+	st := carrier.GRPCStatus()
+	if st == nil || (st.Code() != codes.Unknown && st.Code() != codes.Aborted) || !strings.HasPrefix(st.Message(), "key: ") {
+		return err
+	}
+	for _, sentinel := range []error{store.ErrWriteConflict, kv.ErrTxnLocked} {
+		if strings.HasSuffix(st.Message(), ": "+sentinel.Error()) {
+			return errors.WithStack(errors.Mark(err, sentinel))
+		}
+	}
+	return err
+}
diff --git a/adapter/dynamodb_write_error_test.go b/adapter/dynamodb_write_error_test.go
new file mode 100644
index 000000000..3612f6750
--- /dev/null
+++ b/adapter/dynamodb_write_error_test.go
@@ -0,0 +1,131 @@
+package adapter
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func TestNormalizeDynamoWriteDispatchError(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name string
+		err  error
+		want error
+	}{
+		{"nil", nil, nil},
+		{"local conflict", store.NewWriteConflictError([]byte("k")), store.ErrWriteConflict},
+		{"unknown conflict", status.Error(codes.Unknown, "key: k: write conflict"), store.ErrWriteConflict},
+		{"wrapped aborted conflict", errors.WithStack(status.Error(codes.Aborted, "key: k: write conflict")), store.ErrWriteConflict},
+		{"local lock", kv.NewTxnLockedError([]byte("k")), kv.ErrTxnLocked},
+		{"unknown lock", status.Error(codes.Unknown, "key: k: txn locked"), kv.ErrTxnLocked},
+		{"wrapped aborted lock", errors.WithStack(status.Error(codes.Aborted, "key: k (pending): txn locked")), kv.ErrTxnLocked},
+		{"plain text", errors.New("key: k: write conflict"), nil},
+		{"missing key envelope", status.Error(codes.Unknown, "write conflict"), nil},
+		{"wrong code", status.Error(codes.Internal, "key: k: write conflict"), nil},
+		{"embedded sentinel", status.Error(codes.Unknown, "key: write conflict: invalid value"), nil},
+		{"plain lock", status.Error(codes.Unknown, "txn locked"), nil},
+		{"wrong lock code", status.Error(codes.Internal, "key: k: txn locked"), nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			normalized := normalizeDynamoWriteDispatchError(tc.err)
+			switch {
+			case tc.want != nil:
+				require.True(t, errors.Is(normalized, tc.want), "missing normalized sentinel %v: %v", tc.want, normalized)
+				require.True(t, shouldPreserveTransactWriteAttempt(normalized))
+			case tc.err != nil:
+				require.False(t, isRetryableTransactWriteError(normalized))
+				require.Same(t, tc.err, normalized)
+			default:
+				require.NoError(t, normalized)
+			}
+		})
+	}
+}
+
+type forwardedDynamoLockCoordinator struct {
+	kv.Coordinator
+	rejected atomic.Bool
+	failure  error
+	seal     bool
+}
+
+func (c *forwardedDynamoLockCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if req != nil && req.IsTxn && isDynamoSnapshotSealRequest(req) == c.seal && c.rejected.CompareAndSwap(false, true) {
+		if c.failure != nil {
+			return nil, c.failure
+		}
+		return nil, status.Error(codes.Unknown, "key: pending: txn locked")
+	}
+	return c.Coordinator.Dispatch(ctx, req)
+}
+
+func TestDynamoDBRetriesForwardedSnapshotSealConflict(t *testing.T) {
+	t.Parallel()
+	for _, sentinel := range []error{store.ErrWriteConflict, kv.ErrTxnLocked} {
+		t.Run(sentinel.Error(), func(t *testing.T) {
+			st := store.NewMVCCStore()
+			defer st.Close()
+			schema := dedupItemTable()
+			seedDedupItem(t, st, schema, "1")
+			coord := &forwardedDynamoLockCoordinator{
+				Coordinator: newOCCAdapterCoordinator(st),
+				seal:        true,
+				failure:     status.Error(codes.Aborted, "key: !ddb|seal|pending: "+sentinel.Error()),
+			}
+			server := NewDynamoDBServer(nil, st, coord)
+			_, err := server.updateItemWithRetry(context.Background(), appendListInput())
+			require.NoError(t, err)
+			require.True(t, coord.rejected.Load())
+			require.Equal(t, []string{"1", "3"}, readListValues(t, server, schema))
+		})
+	}
+}
+
+func TestDynamoDBCreateTableRetriesForwardedConflict(t *testing.T) {
+	t.Parallel()
+	st := store.NewMVCCStore()
+	defer st.Close()
+	coord := &forwardedDynamoLockCoordinator{
+		Coordinator: newOCCAdapterCoordinator(st),
+		failure:     status.Error(codes.Unknown, "key: !ddb|meta|table|test: write conflict"),
+	}
+	server := NewDynamoDBServer(nil, st, coord)
+	schema := dedupItemTable()
+	require.NoError(t, server.createTableWithRetry(context.Background(), schema.TableName, schema))
+	require.True(t, coord.rejected.Load())
+	exists, err := server.tableExistsAt(context.Background(), schema.TableName, consistentReadLatestTS)
+	require.NoError(t, err)
+	require.True(t, exists)
+}
+
+func TestDynamoDBRetriesForwardedTxnLock(t *testing.T) {
+	t.Parallel()
+	for _, txn := range []bool{false, true} {
+		t.Run(map[bool]string{false: "item", true: "transaction"}[txn], func(t *testing.T) {
+			st := store.NewMVCCStore()
+			defer st.Close()
+			schema := dedupItemTable()
+			seedDedupItem(t, st, schema, "1")
+			coord := &forwardedDynamoLockCoordinator{Coordinator: newOCCAdapterCoordinator(st)}
+			server := NewDynamoDBServer(nil, st, coord)
+			if txn {
+				require.NoError(t, server.transactWriteItemsWithRetry(context.Background(), transactWriteItemsInput{
+					TransactItems: []transactWriteItem{sealRaftTxnUpdate(schema.TableName)},
+				}))
+			} else {
+				_, err := server.updateItemWithRetry(context.Background(), appendListInput())
+				require.NoError(t, err)
+			}
+			require.True(t, coord.rejected.Load())
+			require.Equal(t, []string{"1", "3"}, readListValues(t, server, schema))
+		})
+	}
+}
diff --git a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
index 7b43e83ae..2a2731775 100644
--- a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
+++ b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
@@ -90,11 +90,82 @@ This is three-node smoke evidence, not evidence of the follower path with the HT
 [`elastickv-dynamodb-test`](../../jepsen/src/elastickv/dynamodb_workload.clj) disables the nemesis in local mode.
 The multi-table workload exists in [`dynamodb_multi_table_workload.clj`](../../jepsen/src/elastickv/dynamodb_multi_table_workload.clj), but that existence and its unit tests are not a recorded execution of the seal failure schedule across data groups.
 
-### Remaining acceptance work
+### Follow-up acceptance tests (2026-10-22)
 
-- Run the allocated/unproposed-write regression through real Raft with follower ingress and an election, retaining the history and final item assertion.
-- Repeat the delayed-write schedule with two touched data groups, and cover seal persistence through snapshot restore and process restart.
-- Exercise follower ingress with the HTTP leader map disabled and leader changes in Jepsen, retaining the exact revision, topology, nemesis configuration, histories, and checker results.
-- Measure the extra seal transaction under hot-item contention, digest collisions, and independent buckets. No seal-specific benchmark function or recorded contention measurement was found in the audited tracked files, and #1259 explicitly reports production throughput as unmeasured. Existing Redis benchmarks are not evidence for this requirement.
+The audit above describes main at `429f5ee7`; the following additions address its validation gaps without replacing #1259's seal protocol.
 
-These are validation and measurement gaps. This audit found the seal protocol and all item entry-point wiring already present; it does not add a second implementation or claim that the remaining acceptance work has passed.
+[`dynamodb_snapshot_seal_raft_test.go`](../../adapter/dynamodb_snapshot_seal_raft_test.go) uses the existing etcd/Raft test fixtures:
+
+- `TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress` holds a write after allocating C, advances an unrelated key, transfers real three-voter Raft leadership, and sends HTTP to a follower with no HTTP leader map. It pauses the second transaction after sealing and planning, but before commit. The old write must fail on the **seal key**, before any second item write commits. Both retried appends must survive, in order, without duplicates. The test also writes through the follower's single-item path. The leader-change case passed ten consecutive race-enabled runs after waiting for every replica to recognize the new leader.
+- `TestDynamoDBSnapshotSealRealRaftTwoDataGroups` repeats the delayed-write schedule using two independent durable one-voter Raft groups, one per table route. It checks that the stored seals have the same R in the two different stores, rejects the old write, and asserts both tables' contents. This is not a two-group, three-replica-per-group topology.
+- `TestDynamoDBSnapshotSealRealRaftSnapshotRestart` forces a durable Raft snapshot, closes the engine and store, and opens the same Raft directory with an empty FSM. It checks the exact restored seal timestamp and item, rejects a stale plan against the restored seal, then accepts a fresh append. This fixture is a cold engine/FSM restart inside the test process, not an OS process-kill experiment.
+
+These tests exposed two runtime gaps, rather than a missing seal implementation:
+
+1. Forwarded write conflicts and transaction locks arrive as gRPC statuses without the local sentinel chain. [`normalizeDynamoWriteDispatchError`](../../adapter/dynamodb_write_error.go) recognizes only known `Unknown`/`Aborted` key-conflict/key-lock envelopes. All DynamoDB write dispatches normalize failures: item/transaction commits, snapshot seals, table schema/migration, and cleanup. The common read-timestamp dispatch helper preserves the existing voucher handling; leader-issued seal and nontransactional cleanup dispatches retain their original timestamp paths. Existing dedup and recompute behavior remains in place. `TestDynamoDBRetriesForwardedTxnLock`, `TestDynamoDBRetriesForwardedSnapshotSealConflict`, and `TestDynamoDBCreateTableRetriesForwardedConflict` cover these retry boundaries. The shared retry predicates, including their SQS callers, are unchanged.
+2. [`buildLeaderDynamo`](../../main.go) previously synthesized a self-only HTTP map even when the flag was omitted. A follower could not resolve a remote leader in that map. An omitted DynamoDB map now selects the existing coordinator gRPC path; explicit maps retain the previous local-entry completion. [`main_dynamodb_follower_test.go`](../../main_dynamodb_follower_test.go) covers both modes and malformed configuration. Redis, S3, and SQS map builders are unchanged.
+
+All production callers of the changed functions were audited with `rg`: three `commitItemWrite` callers (legacy, first dedup attempt, reused dedup attempt), one `runTransactWriteAttempt` caller, and one `buildLeaderDynamo` caller in runtime configuration.
+The follow-up dispatch audit covers all eight reusable read-timestamp write sites plus the leader-issued seal and two direct cleanup dispatches; no DynamoDB write dispatch remains outside normalization. `TestDynamoDBRealRaftFollowerConcurrentCreateTable` sends nine concurrent HTTP setup requests through followers and requires one creation plus ResourceInUse responses, never leaked OCC errors; five race-enabled repetitions passed.
+
+### Follower-ingress Jepsen evidence
+
+[`dynamodb_seal_workload.clj`](../../jepsen/src/elastickv/dynamodb_seal_workload.clj) reuses the existing DynamoDB append workload and Jepsen generator/checker library. Its local nemesis invokes the repository's `raftadmin`, verifies that leadership actually moves to a different node, and records the endpoints in the history. A separate checker rejects zero verified transfers or any failed transfer; a successful append checker alone is insufficient.
+Its DB setup deletes the previous `jepsen_append` table once before any client setup begins; the existing clients then recreate an empty generation. Only a missing table is ignored. This prevents a prior append history on the same cluster from contaminating the new check.
+
+The local run `20261022T231204.299+0900` used three separate server processes on loopback, no `raftDynamoMap`, 15 seconds, rate 30, concurrency 9, maximum transaction length 4, and a two-second transfer interval. The checker output was:
+
+```clojure
+{:append {:valid? true},
+ :leader-changes {:valid? true, :verified-leader-changes 5, :failures []},
+ :valid? true}
+```
+
+The raw history, `test.jepsen`, checker results, and workload log remain in `jepsen/store/elastickv-dynamodb-snapshot-seal/20261022T231204.299+0900/` in the acceptance worktree. The external launcher retains the per-node logs and topology separately. This is short-run safety evidence, not a claim of production availability or a partition/kill campaign; indeterminate client results are permitted by the existing workload.
+
+The [`Jepsen CI workflow`](../../.github/workflows/jepsen-test.yml) now omits the HTTP leader map and runs this workload in both encryption legs. Its artifact retains the seal history and tested revision even on success. Those new CI executions must be checked separately from the local run.
+
+### OS-process snapshot/restart evidence
+
+At source revision `69f2d932d03dfb45a1cd535418f8bbb076d8a43c`, a second local run (`20261022T232208.506+0900`) enabled `ELASTICKV_RAFT_SNAPSHOT_COUNT=20` on three separate server processes. The append and leader-change checkers passed with six verified transfers and no failed transfers. After the workload stopped, all three processes were terminated and reopened from their original data directories and bootstrap membership. The pre/post statuses were:
+
+| Node | Snapshot index before/after | Applied index before/after | Term before/after |
+| --- | ---: | ---: | ---: |
+| n1 | 401 / 401 | 404 / 407 | 8 / 9 |
+| n2 | 402 / 402 | 404 / 407 | 8 / 9 |
+| n3 | 402 / 402 | 404 / 407 | 8 / 9 |
+
+Consistent scans before and after restart matched all 17 items, including list order and values. The per-node logs, topology, and complete pre/post status plus item JSON are retained under `/private/tmp/elastickv-seal-jepsen-luUUDo/`; the Jepsen history remains in the acceptance worktree. The server binary SHA-256 was `6260694477d19fb7c6c9628accebfa5dcacbe6761ba7848e822b84420ca7f022`. The Go snapshot regression separately checks exact seal-timestamp preservation and stale-write rejection; the process experiment checks the persisted real-cluster data path, not just an in-memory restore.
+
+### Contention measurements
+
+[`BenchmarkDynamoDBSnapshotSealRealRaft`](../../adapter/dynamodb_snapshot_seal_benchmark_test.go) uses a durable single-voter Raft engine, four independent frontends, and constant-size PutItem replacements. The independent case uses four different buckets; the collision case searches for four different item keys mapping to one actual digest bucket. A hot-item case uses one item across frontends. Local item locks are not shared between frontends.
+
+Command: `go test ./adapter -run '^$' -bench '^BenchmarkDynamoDBSnapshotSealRealRaft$' -benchtime=100x -count=3 -cpu=4 -benchmem` on Darwin/arm64, Apple M1 Max.
+
+| Case | ns/op range (three samples) | OCC conflicts/op | Seal dispatches/op |
+| --- | ---: | ---: | ---: |
+| Independent buckets | 9,750,054-10,273,370 | 0 | 1.00 |
+| Hot item | 21,095,990-22,904,492 | 1.82-2.23 | 2.46-2.72 |
+| Digest collision | 21,698,244-23,050,694 | 1.97-2.01 | 2.59-2.62 |
+
+All measured operations completed. Collision/hot-item contention increased retries and seal work in this fixture. These measurements include adapter and local durable Raft costs, not network quorum latency, Pebble production compaction, or a before/after throughput delta against an unsafe seal-free implementation. Production throughput remains unmeasured.
+
+### Verification and remaining gates
+
+- The real-Raft tests, all DynamoDB and existing one-phase dedup tests passed with `-race`; the leader-change case also passed ten consecutive runs.
+- `TestBuildLeaderDynamoWithoutHTTPMap` and scoped main/adapter lint passed. Lint used the repository's Go 1.26.6 toolchain because the locally installed analyzer cannot consume Go 1.27 export data.
+- The complete Jepsen unit suite after review fixes passed: 209 tests, 586 assertions, zero failures/errors.
+- Full local `go test ./...` is not a pass: the pre-existing FUSE test references Linux-only `syscall.Statfs_t.Namelen` on macOS, an unchanged backup renewal test hit its deadline (five isolated reruns passed), and vet overlapped a transient test edit (subsequent target runs passed). Linux CI, the new encryption on/off Jepsen runs, and current-HEAD review of the acceptance changes remain gates.
+
+At `8db0155d632aec73a433cc3478f0a474c6921e8d`, current-head code/security reviews had no findings and all required CI passed, including the complete Ubuntu Go suite and encryption off/on Jepsen. The non-root local Linux full-suite attempt was not a pass: an unchanged SQS FIFO visibility test failed under load, then passed twenty isolated repetitions. The separate Ubuntu CI result does not erase that local failure.
+
+### Outstanding encrypted stress findings
+
+The design remains Partial despite the short successful CI and review results:
+
+- Additional encrypted three-process stress at `8db0155d`, with snapshot interval 20, a preceding append workload, fresh-table reset, and six verified leader changes, returned `valid? false` with `G-single-item-realtime` and `G1c-realtime`. An acknowledged append 37 to key 16 was absent from a later successful read including 46. History `20261022T235149.907+0900` and the per-node logs under `/private/tmp/elastickv-seal-jepsen-FpXSbX/` are retained. This run is not counted as a success.
+- Existing [PR #1286](https://github.com/bootjp/elastickv/pull/1286) addresses an overlapping Raft apply/timestamp problem. A diagnostic build combining its `78cab1ba` with `8db0155d` passed one identical encrypted workload with six verified transfers. This suggests overlap, not a causal proof or a completed stress campaign. The comparison was made in a separate worktree; this acceptance PR does not duplicate that implementation.
+- The combined run's subsequent encrypted restart failed: the startup guard attempted to scan Raft entries `(6, 877]` after compaction and MemoryStorage returned `requested index is unavailable due to compaction`. The unmodified `8db0155d` binary also refuses startup on that retained diagnostic state, without the #1286 patch. Logs are under `/private/tmp/elastickv-seal-jepsen-dscUKN/`. The earlier plaintext restart success is not evidence that encrypted restart is safe.
+
+Before declaring this design complete, coordinate the existing Raft fix without duplication, cover and resolve the encrypted compacted-snapshot startup failure, and rerun the acceptance matrix. Adding tests, receiving a clean review, or changing status alone does not complete acceptance.
diff --git a/jepsen/src/elastickv/dynamodb_seal_workload.clj b/jepsen/src/elastickv/dynamodb_seal_workload.clj
new file mode 100644
index 000000000..e9bcf0987
--- /dev/null
+++ b/jepsen/src/elastickv/dynamodb_seal_workload.clj
@@ -0,0 +1,110 @@
+(ns elastickv.dynamodb-seal-workload
+  "Local DynamoDB append acceptance with recorded, verified Raft elections."
+  (:gen-class)
+  (:require [clojure.java.shell :as shell]
+            [clojure.string :as str]
+            [cognitect.aws.client.api :as aws]
+            [elastickv.cli :as cli]
+            [elastickv.dynamodb-workload :as dynamo]
+            [jepsen.checker :as checker]
+            [jepsen.client :as client]
+            [jepsen.db :as db]
+            [jepsen.generator :as gen]
+            [jepsen.nemesis :as nemesis]))
+
+(defn- admin! [binary address command]
+  (let [result (shell/sh binary address command)]
+    (when-not (zero? (:exit result))
+      (throw (ex-info "raftadmin failed" (assoc result :address address :command command))))
+    (:out result)))
+
+(defn- current-leader [binary addresses]
+  (or (some (fn [address]
+              (when (= "state: LEADER" (str/trim (admin! binary address "state")))
+                address))
+            addresses)
+      (throw (ex-info "no Raft leader" {:addresses addresses}))))
+
+(defn leader-transfer-nemesis [binary addresses]
+  (reify
+    nemesis/Reflection
+    (fs [_] #{:leader-transfer})
+    nemesis/Nemesis
+    (setup! [this _] this)
+    (invoke! [_ _ op]
+      (try
+        (let [before (current-leader binary addresses)
+              _ (admin! binary before "leadership_transfer")
+              deadline (+ (System/nanoTime) 10000000000)
+              after (loop []
+                      (let [leader (try (current-leader binary addresses)
+                                        (catch Exception _ nil))]
+                        (cond
+                          (and leader (not= before leader)) leader
+                          (< (System/nanoTime) deadline) (do (Thread/sleep 50) (recur))
+                          :else (throw (ex-info "leadership did not change" {:before before})))))]
+          (assoc op :value {:from before :to after}))
+        (catch Exception e
+          (assoc op :error (.getMessage e) :value :transfer-failed))))
+    (teardown! [_ _] nil)))
+
+(defn leader-change-checker []
+  (reify checker/Checker
+    (check [_ _ history _]
+      (let [ops (filter #(and (= :nemesis (:process %)) (= :leader-transfer (:f %))) history)
+            failures (filter :error ops)
+            changes (filter #(let [{:keys [from to]} (:value %)]
+                               (and from to (not= from to))) ops)]
+        {:valid? (boolean (and (seq changes) (empty? failures)))
+         :verified-leader-changes (count changes)
+         :failures (vec failures)}))))
+
+(defn fresh-append-db []
+  (reify db/DB
+    (setup! [_ test node]
+      ;; DB setup completes on every node before any client setup starts.
+      ;; Delete once; the existing clients recreate an empty new generation.
+      (when (= node (first (:nodes test)))
+        (let [opened (client/open! (:client test) test node)]
+          (try
+            (let [response (aws/invoke (:ddb opened)
+                                      {:op :DeleteTable :request {:TableName "jepsen_append"}})]
+              (when (and (:cognitect.anomalies/category response)
+                         (not= "ResourceNotFoundException" (:__type response)))
+                (throw (ex-info "cannot reset append table" {:response response}))))
+            (finally (client/close! opened test))))))
+    (teardown! [_ _ _] nil)))
+
+(defn seal-test [opts]
+  (when-not (and (:local opts) (seq (:raftadmin-bin opts)) (seq (:raft-addresses opts)))
+    (throw (ex-info "seal acceptance requires local mode, raftadmin-bin and raft-addresses" {})))
+  (let [base (dynamo/elastickv-dynamodb-test opts)
+        workload (dynamo/dynamodb-append-workload opts)
+        interval (or (:leader-change-interval opts) 2)]
+    (assoc base
+      :name "elastickv-dynamodb-snapshot-seal"
+      :db (fresh-append-db)
+      :nemesis (leader-transfer-nemesis (:raftadmin-bin opts) (:raft-addresses opts))
+      :checker (checker/compose {:append (:checker workload) :leader-changes (leader-change-checker)})
+      :generator (->> (:generator workload)
+                      (gen/nemesis (cycle [(gen/sleep interval) {:type :info :f :leader-transfer}]))
+                      (gen/stagger (/ (double (or (:rate opts) 5))))
+                      (gen/time-limit (or (:time-limit opts) 30))))))
+
+(def seal-cli-opts
+  [[nil "--raftadmin-bin PATH" "Repository raftadmin binary (local control only)."]
+   [nil "--raft-addresses ADDRESSES" "Comma-separated local Raft/gRPC endpoints."
+    :parse-fn #(str/split % #",")]
+   [nil "--leader-change-interval SECONDS" "Seconds between leadership transfers."
+    :default 2 :parse-fn #(Double/parseDouble %)
+    :validate [pos? "must be positive"]]])
+
+(defn -main [& args]
+  (cli/run-workload!
+    args (into (into cli/common-cli-opts dynamo/dynamo-cli-opts) seal-cli-opts)
+    (fn [options]
+      (let [options (cli/parse-common-opts options (:dynamo-ports options))]
+        (assoc options
+          :dynamo-host (:host options)
+          :node->port (cli/ports->node-map (:dynamo-ports options) (:nodes options)))))
+    seal-test))
diff --git a/jepsen/test/elastickv/dynamodb_seal_workload_test.clj b/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
new file mode 100644
index 000000000..3931b56fc
--- /dev/null
+++ b/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
@@ -0,0 +1,66 @@
+(ns elastickv.dynamodb-seal-workload-test
+  (:require [clojure.java.shell :as shell]
+            [clojure.test :refer :all]
+            [cognitect.aws.client.api :as aws]
+            [elastickv.dynamodb-seal-workload :as seal]
+            [jepsen.checker :as checker]
+            [jepsen.client :as client]
+            [jepsen.db :as db]
+            [jepsen.nemesis :as nemesis]))
+
+(deftest requires-recorded-successful-leader-change
+  (let [check #(checker/check (seal/leader-change-checker) {} % {})
+        op {:process :nemesis :type :info :f :leader-transfer}]
+    (is (not (:valid? (check []))))
+    (is (not (:valid? (check [(assoc op :value {:from "a" :to "a"})]))))
+    (is (:valid? (check [(assoc op :value {:from "a" :to "b"})])))
+    (is (false? (:valid? (check [(assoc op :value {:from "a" :to "b"})
+                                (assoc op :error "transfer failed")]))))))
+
+(deftest records-actual-transfer-and-failure
+  (let [leader (atom "a")
+        n (seal/leader-transfer-nemesis "/raftadmin" ["a" "b"])
+        op {:type :info :f :leader-transfer}]
+    (with-redefs [shell/sh (fn [binary address command]
+                            (is (= "/raftadmin" binary))
+                            (case command
+                              "state" {:exit 0 :out (str "state: " (if (= address @leader) "LEADER" "FOLLOWER") "\n")}
+                              "leadership_transfer" (do (reset! leader "b") {:exit 0 :out "ok\n"})))]
+      (is (= {:from "a" :to "b"} (:value (nemesis/invoke! n {} op)))))
+    (with-redefs [shell/sh (fn [& _] {:exit 1 :err "RPC failed"})]
+      (is (= "raftadmin failed" (:error (nemesis/invoke! n {} op)))))))
+
+(deftest acceptance-is-local-and-opt-in
+  (is (thrown? clojure.lang.ExceptionInfo (seal/seal-test {})))
+  (let [test (seal/seal-test {:local true :raftadmin-bin "/raftadmin" :raft-addresses ["a" "b"]})]
+    (is (= "elastickv-dynamodb-snapshot-seal" (:name test)))
+    (is (= #{:leader-transfer} (nemesis/fs (:nemesis test))))))
+
+(deftest resets-the-append-table-once-before-client-setup
+  (let [calls (atom [])
+        closed (atom 0)
+        test {:nodes ["n1" "n2"] :client ::client}]
+    (with-redefs [client/open! (fn [& _] {:ddb ::ddb})
+                  client/close! (fn [& _] (swap! closed inc))
+                  aws/invoke (fn [ddb request]
+                               (is (= ::ddb ddb))
+                               (swap! calls conj request)
+                               {})]
+      (doseq [node (:nodes test)] (db/setup! (seal/fresh-append-db) test node))
+      (is (= [{:op :DeleteTable :request {:TableName "jepsen_append"}}] @calls))
+      (is (= 1 @closed)))))
+
+(deftest reset-only-ignores-missing-table-and-always-closes
+  (doseq [[response fails?] [[{:cognitect.anomalies/category :not-found
+                             :__type "ResourceNotFoundException"} false]
+                           [{:cognitect.anomalies/category :fault
+                             :__type "InternalServerError"} true]]]
+    (let [closed (atom false)]
+      (with-redefs [client/open! (fn [& _] {:ddb ::ddb})
+                    client/close! (fn [& _] (reset! closed true))
+                    aws/invoke (fn [& _] response)]
+        (let [run #(db/setup! (seal/fresh-append-db) {:nodes ["n1"]} "n1")]
+          (if fails?
+            (is (thrown? clojure.lang.ExceptionInfo (run)))
+            (is (nil? (run)))))
+        (is @closed)))))
diff --git a/main.go b/main.go
index 0b144fbc7..5dcbe4f82 100644
--- a/main.go
+++ b/main.go
@@ -1562,6 +1562,11 @@ func validateSQSFifoPartitionMapNoDedicatedTSOGroup(partitionMap map[string]sqsF
 }
 
 func buildLeaderDynamo(groups []groupSpec, dynamoAddr string, raftDynamoMap string) (map[string]string, error) {
+	// An absent HTTP map selects the coordinator's gRPC follower path. A
+	// self-only map cannot resolve a remote leader and rejects every request.
+	if strings.TrimSpace(raftDynamoMap) == "" {
+		return nil, nil
+	}
 	return buildLeaderAddrMap(groups, dynamoAddr, raftDynamoMap, parseRaftDynamoMap)
 }
 
diff --git a/main_dynamodb_follower_test.go b/main_dynamodb_follower_test.go
new file mode 100644
index 000000000..b983093f4
--- /dev/null
+++ b/main_dynamodb_follower_test.go
@@ -0,0 +1,25 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildLeaderDynamoWithoutHTTPMap(t *testing.T) {
+	t.Parallel()
+	groups := []groupSpec{{id: 1, address: "127.0.0.1:50051"}}
+	for _, raw := range []string{"", "  "} {
+		got, err := buildLeaderDynamo(groups, "127.0.0.1:8000", raw)
+		require.NoError(t, err)
+		require.Empty(t, got)
+	}
+	got, err := buildLeaderDynamo(groups, "127.0.0.1:8000", "127.0.0.1:50052=127.0.0.1:8001")
+	require.NoError(t, err)
+	require.Equal(t, map[string]string{
+		"127.0.0.1:50051": "127.0.0.1:8000",
+		"127.0.0.1:50052": "127.0.0.1:8001",
+	}, got)
+	_, err = buildLeaderDynamo(groups, "127.0.0.1:8000", "invalid")
+	require.Error(t, err)
+}