feat(encryption): Stage 6E-2c — dynamic raft envelope wrap on ShardGroup (#922)

## Summary

Stage 6E-2c: install the Stage 6E-2 wrap-on-propose plumbing on every
shard group without yet activating any cipher. Every `ShardGroup` gets
an `atomic.Pointer[RaftPayloadWrapper]` cell; the proposer chain
consults that pointer on every `Propose` / `ProposeAdmin` call;
`SetRaftPayloadWrap` is the hot-swap surface Stage 6E-2d will use when
`EnableRaftEnvelope` commits. The wrap closure factory itself ships in
Stage 6E-2e/2f.

A startup hook reads the sidecar and logs a `WARN` if
`RaftEnvelopeCutoverIndex != 0` — operator-visible refusal of a
deployment that crosses the cutover line under a binary without the
cipher factory wired.

## Pieces

- `dynamicWrappedProposer` (kv/raft_payload_wrapper.go): wraps a
`raftengine.Proposer` with an `atomic.Pointer.Load` on every call.
Mirrors `wrappedProposer`'s Propose / ProposeAdmin routing and wrap
semantics — the wrap layer is NOT a barrier exemption; Propose and
ProposeAdmin apply the same wrap. The cutover marker stays cleartext via
the raw-engine reference at the call site, not at the method level.
- `ShardGroup.raftPayloadWrap` + `SetRaftPayloadWrap` +
`RaftPayloadWrap` (kv/sharded_coordinator.go): the published hot-swap
surface.
- `NewLeaderProxyForShardGroup` + `withDynamicRaftPayloadWrap`
(kv/sharded_coordinator.go, kv/transaction.go): the production wiring
path. `main.go`'s `buildShardGroups` now constructs every shard group's
`LeaderProxy` via this helper so the dynamic wrap chain is universal.
- `noteRaftEnvelopeCutoverStartup` (main_encryption_write_wiring.go):
process-start sidecar read. Silent on the universal pre-cutover case and
on any read error (the Stage 6C-1 startup guard is the authoritative
refuser). Logs `WARN` only when the cutover index is non-zero but the
binary does not (yet) ship the cipher factory.

## Tests (8 new)

- `TestDynamicWrappedProposer_NilPointerReturnsInnerVerbatim` —
defensive degradation for nil `*atomic.Pointer`.
- `TestDynamicWrappedProposer_NilStoredIsPassThrough` — Stage 3 default
(verbatim routing for Propose and ProposeAdmin separately).
- `TestDynamicWrappedProposer_LoadsCurrentWrapEveryCall` — hot-swap
contract (install / clear observed by the next proposal).
- `TestDynamicWrappedProposer_ProposeAdminAppliesCurrentWrap` —
admin-path mirror; confirms `inner.ProposeAdmin` (not `Propose`) is
reached when wrap is active.
- `TestShardGroup_SetRaftPayloadWrap_RoundTrip` — Set/Get round-trip
including the nil-clears semantics.
- `TestNoteRaftEnvelopeCutoverStartup_*` (×4) — empty path / missing
file / zero cutover / non-zero cutover (asserts the hook is read-only).

## Five-lens self-review

- **Data loss**: the dynamic wrap proposer is a pass-through when the
pointer holds nil (the Stage 6E-2c default). Production behaviour is
unchanged today — every proposal still reaches the engine cleartext via
the existing path. No new error swallow, no new fall-through.
- **Concurrency / distributed**: hot-swap uses `atomic.Pointer.Store` /
`Load`; reads are lock-free and observe a fully-published closure
(either nil or non-nil) on every call. Last-writer-wins on concurrent
stores; no torn read.
- **Performance**: one extra `atomic.Pointer.Load` per Propose /
ProposeAdmin on the dynamic path; no allocation, no contention. Wrap
closure invocation cost is paid only when non-nil; matches the existing
`wrappedProposer` profile.
- **Data consistency**: the wrap pointer's lifetime is owned by the
`ShardGroup` that owns the engine; the proposer holds an
`*atomic.Pointer` reference back into `ShardGroup`. No cross-shard
mixing — each shard has its own pointer, so a future per-shard rotation
policy works without changes here.
- **Test coverage**: 8 new tests cover the wrap-proposer mechanics, the
ShardGroup hot-swap surface, and the startup hook input matrix.

## Caller audit (per /loop semantic-change rule)

- `kv.NewLeaderProxyWithEngine` (kept; non-wrap-aware): no behaviour
change. Test fixtures and any shard group that opts out of encryption
continue to use it.
- `kv.NewLeaderProxyForShardGroup` (new; wrap-aware): used by
`main.go`'s `buildShardGroups` for every production shard group. Audit:
the only construction site for production `ShardGroup` instances now
routes through this helper.
- ShardGroup struct literal in `main.go`: switched from one-shot literal
with `Txn` set inline to two-step build (struct literal, then `Txn =
NewLeaderProxyForShardGroup(sg, opts...)`). Necessary because
`NewLeaderProxyForShardGroup` needs the `&sg.raftPayloadWrap` pointer,
which is only addressable after the struct exists.
- `SetRaftPayloadWrap` callers: none in this PR. The surface is
published for Stage 6E-2d's `EnableRaftEnvelope` handler.

## Test plan

- [x] `go build ./...` clean
- [x] `go vet ./...` clean
- [x] `golangci-lint run` clean (pre-commit hook ran)
- [x] `go test -race -run
'DynamicWrappedProposer|ShardGroup_Set|WrappedProposer|NoteRaftEnvelopeCutover'
.` passes
- [x] `go test -race ./kv/` scoped tests pass

Refs: `docs/design/2026_05_31_partial_6e_enable_raft_envelope.md`


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

## Release Notes

* **New Features**
* Introduced runtime-swappable raft payload envelope wrapping with
dynamic wrapper support across all proposal types and shard groups.

* **Chores**
* Added startup validation to detect and prevent incompatible payload
envelope cutover states with clear operator guidance.

* **Tests**
* Expanded test coverage for dynamic payload wrapper behavior, atomic
pointer runtime swapping, integration with shard groups, and startup
safety checks.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

kv/raft_payload_wrapper.go

@@ -2,6 +2,7 @@ package kv
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/bootjp/elastickv/internal/raftengine"
 	"github.com/cockroachdb/errors"
@@ -108,3 +109,95 @@ func (p *wrappedProposer) ProposeAdmin(ctx context.Context, data []byte) (*rafte
 	}
 	return res, nil
 }
+
+// dynamicWrappedProposer is the Stage 6E-2c sibling of wrappedProposer
+// for the case where the wrap closure must be swappable at runtime.
+// On every Propose / ProposeAdmin call it loads an atomic.Pointer to
+// the currently-active RaftPayloadWrapper; a nil pointer means wrap
+// is inactive (payload passes through verbatim).
+//
+// Why a separate type (vs adding a setter to wrappedProposer): the
+// existing wrappedProposer captures wrap at construction time and is
+// used by static call sites that never need to change wrap state.
+// The Stage 6E-2 cutover model needs runtime swap so the
+// EnableRaftEnvelope admin handler (Stage 6E-2d) can publish the
+// active wrap closure the instant the cutover entry commits, without
+// rebuilding the TransactionManager or stalling in-flight proposals.
+// Splitting the two keeps the static-wrap fast path branch-free and
+// keeps the dynamic path's atomic.Pointer.Load() cost confined to
+// the call sites that need it.
+//
+// The pointer is required to be non-nil; pass an atomic.Pointer that
+// stores nil to express "wrap inactive". This is so the call site
+// (typically a ShardGroup) owns the storage and the proposer just
+// reads.
+type dynamicWrappedProposer struct {
+	inner   raftengine.Proposer
+	wrapPtr *atomic.Pointer[RaftPayloadWrapper]
+}
+
+// newDynamicWrappedProposer wires a proposer that consults wrapPtr
+// on every Propose / ProposeAdmin. wrapPtr.Load() == nil keeps the
+// path as a pure pass-through to inner; a non-nil value applies
+// wrap before forwarding.
+//
+// Contract: inner MUST be non-nil. A nil inner is a caller bug
+// (the construction site is responsible for handing in a valid
+// proposer) and the constructor deliberately does not silently
+// degrade — passing nil to Propose / ProposeAdmin would NPE at
+// the first call site, which is the same fail-fast shape as the
+// sibling newWrappedProposer(nil, wrap) and matches CLAUDE.md's
+// "don't validate for scenarios that can't happen" policy at
+// internal boundaries.
+//
+// wrapPtr MAY be nil — that path returns inner verbatim, mirroring
+// newWrappedProposer(inner, nil)'s degraded shape so a wiring site
+// that accidentally drops the pointer downgrades to a no-wrap
+// proposer rather than crashing the engine loop on first use. The
+// caller owns the storage lifetime of the pointer when non-nil.
+func newDynamicWrappedProposer(inner raftengine.Proposer, wrapPtr *atomic.Pointer[RaftPayloadWrapper]) raftengine.Proposer {
+	if wrapPtr == nil {
+		// Defensive degradation: passing a nil pointer would NPE
+		// on the first Propose. Treat as static no-wrap so callers
+		// that pass nil by mistake see the Stage 3 default rather
+		// than crashing the engine loop. This matches
+		// newWrappedProposer(inner, nil) returning inner verbatim.
+		return inner
+	}
+	return &dynamicWrappedProposer{inner: inner, wrapPtr: wrapPtr}
+}
+
+func (p *dynamicWrappedProposer) currentWrap() RaftPayloadWrapper {
+	if loaded := p.wrapPtr.Load(); loaded != nil {
+		return *loaded
+	}
+	return nil
+}
+
+func (p *dynamicWrappedProposer) Propose(ctx context.Context, data []byte) (*raftengine.ProposalResult, error) {
+	wrapped, err := applyRaftPayloadWrap(p.currentWrap(), data)
+	if err != nil {
+		return nil, err
+	}
+	res, err := p.inner.Propose(ctx, wrapped)
+	if err != nil {
+		return nil, errors.Wrap(err, "kv: dynamic wrapped propose")
+	}
+	return res, nil
+}
+
+// ProposeAdmin mirrors Propose's wrap-applies semantics. See
+// wrappedProposer.ProposeAdmin for the design rationale (the wrap
+// layer is NOT a barrier exemption; the EnableRaftEnvelope cutover
+// marker bypasses wrap at the call site, not the method level).
+func (p *dynamicWrappedProposer) ProposeAdmin(ctx context.Context, data []byte) (*raftengine.ProposalResult, error) {
+	wrapped, err := applyRaftPayloadWrap(p.currentWrap(), data)
+	if err != nil {
+		return nil, err
+	}
+	res, err := p.inner.ProposeAdmin(ctx, wrapped)
+	if err != nil {
+		return nil, errors.Wrap(err, "kv: dynamic wrapped propose-admin")
+	}
+	return res, nil
+}

kv/raft_payload_wrapper_test.go

@@ -223,3 +223,166 @@ func TestWrappedProposer_RoundTripWithRealCipher(t *testing.T) {
 		t.Fatalf("round-trip mismatch: got %q, want %q", got, plaintext)
 	}
 }
+
+// TestDynamicWrappedProposer_NilPointerReturnsInnerVerbatim pins the
+// defensive degradation in newDynamicWrappedProposer: if a caller
+// accidentally passes a nil *atomic.Pointer, the wrapper falls back
+// to the inner proposer rather than NPE on the first Propose call.
+// This matches newWrappedProposer(inner, nil)'s shape and keeps a
+// misconfigured wiring from crashing the engine loop.
+func TestDynamicWrappedProposer_NilPointerReturnsInnerVerbatim(t *testing.T) {
+	t.Parallel()
+	inner := &fakeProposer{}
+	got := newDynamicWrappedProposer(inner, nil)
+	if got != raftengine.Proposer(inner) {
+		t.Fatal("nil pointer: newDynamicWrappedProposer should return the inner proposer verbatim, not a wrapper")
+	}
+}
+
+// TestDynamicWrappedProposer_NilStoredIsPassThrough pins the
+// Stage 6E-2c "wrap inactive" default: when the atomic pointer
+// holds nil (no wrap closure installed), payloads reach the inner
+// proposer verbatim and routing distinguishes Propose vs
+// ProposeAdmin correctly.
+func TestDynamicWrappedProposer_NilStoredIsPassThrough(t *testing.T) {
+	t.Parallel()
+	var wrapPtr atomic.Pointer[RaftPayloadWrapper]
+	inner := &fakeProposer{}
+	wp := newDynamicWrappedProposer(inner, &wrapPtr)
+
+	if _, err := wp.Propose(context.Background(), []byte("plain")); err != nil {
+		t.Fatalf("Propose: %v", err)
+	}
+	if got := inner.calls.Load(); got != 1 {
+		t.Fatalf("inner.Propose calls = %d, want 1", got)
+	}
+	if !bytes.Equal(inner.last, []byte("plain")) {
+		t.Fatalf("inner.Propose saw %q, want %q (verbatim)", inner.last, []byte("plain"))
+	}
+
+	if _, err := wp.ProposeAdmin(context.Background(), []byte("admin")); err != nil {
+		t.Fatalf("ProposeAdmin: %v", err)
+	}
+	if got := inner.adminCalls.Load(); got != 1 {
+		t.Fatalf("inner.ProposeAdmin calls = %d, want 1", got)
+	}
+	if !bytes.Equal(inner.adminLast, []byte("admin")) {
+		t.Fatalf("inner.ProposeAdmin saw %q, want %q (verbatim)", inner.adminLast, []byte("admin"))
+	}
+}
+
+// TestDynamicWrappedProposer_LoadsCurrentWrapEveryCall pins the
+// Stage 6E-2c hot-swap contract: the atomic pointer is loaded on
+// every Propose / ProposeAdmin call, so a publish via Store between
+// two proposals is observed by the second without rebuilding the
+// proposer.
+func TestDynamicWrappedProposer_LoadsCurrentWrapEveryCall(t *testing.T) {
+	t.Parallel()
+	var wrapPtr atomic.Pointer[RaftPayloadWrapper]
+	inner := &fakeProposer{}
+	wp := newDynamicWrappedProposer(inner, &wrapPtr)
+
+	// First call: no wrap installed, payload passes through verbatim.
+	if _, err := wp.Propose(context.Background(), []byte("first")); err != nil {
+		t.Fatalf("Propose 1: %v", err)
+	}
+	if !bytes.Equal(inner.last, []byte("first")) {
+		t.Fatalf("first call: inner saw %q, want %q", inner.last, "first")
+	}
+
+	// Install a wrap mid-flight; the next call must see it.
+	var wrap RaftPayloadWrapper = func(p []byte) ([]byte, error) {
+		out := make([]byte, len(p)+1)
+		out[0] = 'W'
+		copy(out[1:], p)
+		return out, nil
+	}
+	wrapPtr.Store(&wrap)
+
+	if _, err := wp.Propose(context.Background(), []byte("second")); err != nil {
+		t.Fatalf("Propose 2: %v", err)
+	}
+	want := append([]byte{'W'}, []byte("second")...)
+	if !bytes.Equal(inner.last, want) {
+		t.Fatalf("second call: inner saw %q, want %q (wrapped)", inner.last, want)
+	}
+
+	// Clear the wrap; the next call must see it gone.
+	wrapPtr.Store(nil)
+
+	if _, err := wp.Propose(context.Background(), []byte("third")); err != nil {
+		t.Fatalf("Propose 3: %v", err)
+	}
+	if !bytes.Equal(inner.last, []byte("third")) {
+		t.Fatalf("third call (post-clear): inner saw %q, want %q (verbatim)", inner.last, "third")
+	}
+}
+
+// TestShardGroup_SetRaftPayloadWrap_RoundTrip pins the
+// Stage 6E-2c hot-swap surface on ShardGroup. Set with a non-nil
+// closure publishes it; Set with nil clears; the round-trip via
+// RaftPayloadWrap returns identity.
+func TestShardGroup_SetRaftPayloadWrap_RoundTrip(t *testing.T) {
+	t.Parallel()
+	g := &ShardGroup{}
+	if got := g.RaftPayloadWrap(); got != nil {
+		t.Fatalf("default: RaftPayloadWrap = %v, want nil", got)
+	}
+	var called atomic.Int32
+	var wrap RaftPayloadWrapper = func(p []byte) ([]byte, error) {
+		called.Add(1)
+		return p, nil
+	}
+	g.SetRaftPayloadWrap(wrap)
+	got := g.RaftPayloadWrap()
+	if got == nil {
+		t.Fatal("after Set: RaftPayloadWrap returned nil")
+	}
+	if _, err := got([]byte("x")); err != nil {
+		t.Fatalf("invoking stored wrap: %v", err)
+	}
+	if got := called.Load(); got != 1 {
+		t.Fatalf("stored wrap call count = %d, want 1", got)
+	}
+	g.SetRaftPayloadWrap(nil)
+	if got := g.RaftPayloadWrap(); got != nil {
+		t.Fatalf("after Set(nil): RaftPayloadWrap = %v, want nil", got)
+	}
+}
+
+// TestDynamicWrappedProposer_ProposeAdminAppliesCurrentWrap is the
+// admin-path mirror of LoadsCurrentWrapEveryCall: ProposeAdmin
+// must also see hot-swapped wrap closures so post-cutover admin
+// entries (RotateDEK, RegisterEncryptionWriter) committed at
+// `index > raftEnvelopeCutoverIndex` carry the AEAD envelope the
+// §6.3 strict-`>` apply hook expects. The cutover marker itself
+// bypasses this proposer at the call site (raw engine reference),
+// not at this layer.
+func TestDynamicWrappedProposer_ProposeAdminAppliesCurrentWrap(t *testing.T) {
+	t.Parallel()
+	var wrapPtr atomic.Pointer[RaftPayloadWrapper]
+	var wrap RaftPayloadWrapper = func(p []byte) ([]byte, error) {
+		out := make([]byte, len(p)+1)
+		out[0] = 'A'
+		copy(out[1:], p)
+		return out, nil
+	}
+	wrapPtr.Store(&wrap)
+
+	inner := &fakeProposer{}
+	wp := newDynamicWrappedProposer(inner, &wrapPtr)
+
+	if _, err := wp.ProposeAdmin(context.Background(), []byte("admin-payload")); err != nil {
+		t.Fatalf("ProposeAdmin: %v", err)
+	}
+	if got := inner.adminCalls.Load(); got != 1 {
+		t.Fatalf("inner.ProposeAdmin calls = %d, want 1", got)
+	}
+	if got := inner.calls.Load(); got != 0 {
+		t.Fatalf("inner.Propose calls = %d, want 0 — admin path must NOT downgrade to non-exempt method", got)
+	}
+	want := append([]byte{'A'}, []byte("admin-payload")...)
+	if !bytes.Equal(inner.adminLast, want) {
+		t.Fatalf("inner.ProposeAdmin saw %q, want %q (wrapped)", inner.adminLast, want)
+	}
+}