Merge remote-tracking branch 'origin/main' into review/ha-rolling

The production ship path read the database file with no connection held, and three
failure modes that must be loud were silent.

snapshotcodec: wholeFile.produce kept the "hold db's SOLE connection across the
checkpoint AND the file read" guarantee only on the default (nil-checkpoint) path,
but buildDurability wires production through the INJECTED-checkpoint path, which
released it before readFramed. os.ReadFile takes no SQLite lock, so a concurrent
commit -- and the WAL auto-checkpoint it can trigger, which writes pages straight
into the real path -- tears the image mid-read; the fence then admits that corrupt
snapshot at the lease round, acks it, and a successor restores it. Re-acquire and
hold the connection across the read, AFTER the injected checkpoint: the envelope
backend's Checkpoint takes that same connection itself, so acquiring it first
deadlocks at MaxOpenConns(1). A write landing in the gap is simply absent from the
snapshot, which is safe -- it was never acked.

build: multiReplica keyed on CLOUD_PEERS, which the live-membership chart never
sets (a Deployment has no stable peer list, only CLOUD_PEER_SELECTOR), so a
multi-replica deployment whose durable plane failed to construct reported
"single-replica/dev" -- the silent non-durable state the alert exists to prevent,
and the only gate left now that the opt-in flag is gone. Key it on CLOUD_REPLICAS
first. Also report Membership.Start's error: an empty set means no org has a safe
owner and every write fails closed.

membership: org.Membership's refresh loop DISCARDS the source error, so a missing
pods:list RBAC or a selector typo froze the writer set at its last-good value --
empty if the first refresh failed -- and took every per-org write read-only with no
diagnostic at all. Log it, rate-limited.

books: while durability was opt-in, OrgStore.Sync always acked, so discarding the
ack was dead code. It is a live path now: a fenced ship leaves the postings in a
local file the next hydrate overwrites. The commerce ingest is idempotent, but an
OFX/CSV import is not re-derivable, so neither may pass as a clean success.

LLM.md

@@ -163,6 +163,56 @@ package under `clients/<name>` that obeys these seams — nothing more.
   the SOLE driver (blank-imported once, in orgdb.go); subsystems never import a
   SQLite driver themselves. The caller owns its schema/migration and Close.
 
+## Zero-downtime HA for per-org stores (rolling-upgrade safe)
+
+The per-org store path (`cloud.OrgStore` + `internal/org`) is HA over embedded
+SQLite: `ha` decides WHO writes (HRW election + a monotone fencing round), `vfs`
+FencedStore decides HOW state ships (hydrate-on-open + fenced ship to S3), and this
+layer decides WHEN ownership transitions. Three orthogonal lanes; SQLite stays
+embedded underneath.
+
+- **Durability is THE path, capability-detected — no flag.** `buildDurability`
+  probes at boot: no object store reachable (dev / native-Go) → local-only, same
+  code path; a reachable store → `org.ProbeCAS` PROVES its conditional-PUT
+  atomicity (two racing If-None-Match creates + If-Match updates, exactly one winner
+  each) before fencing any tenant data. A store that can't be proven atomic fails
+  SAFE to local-only + a loud alert (never fence where two writers could win one
+  round). Replaced the old `CLOUD_RESEARCH_DURABLE` opt-in — the atomicity gate (H2)
+  is now a self-check the binary runs.
+- **Live membership (no static peer list).** `membership_k8s.go` lists Ready,
+  non-terminating pods by label (`CLOUD_PEER_SELECTOR`) via the K8s API each 2s
+  refresh, so a rolling upgrade's changing pod set is tracked and a draining/dead pod
+  (`DeletionTimestamp` set, or NotReady) leaves the writer election at once. Out of
+  cluster / no selector → static self set (`podWriterEligible` is the ONE ready gate;
+  visor has the twin, the shared `hanzoai/ha/k8s` source folds them).
+- **M3 live re-acquire, no restart.** A store that opened degraded (read-only) is
+  promoted IN PLACE when this replica becomes the org's elected owner:
+  `Durable.PendingPromotion` gates it, `TryClaim` probes the lease (CAS only, no file
+  I/O), then `OrgStore.promote` quiesces the read-only handle and reopens as owner
+  (Hydrate renews + CarryForward-restores under the FRESH handle — the file swap is
+  why the reopen is required). The reopen claims a strictly higher round, fencing the
+  prior owner — never two live writers.
+- **Graceful drain.** SIGTERM → `SetDraining()` → `/readyz` 503 (drain-aware, ops
+  port) → K8s marks NotReady → peers re-elect this pod's orgs to live successors
+  (which hydrate via M3) → the pod stays serving a short grace, then in-flight drains
+  and final state ships (`OrgStore.CloseAll`, ship-before-close). The shard router
+  routes on the live set when the durable plane is on, so a draining pod's orgs go to
+  the ready successor — not to the gone pod. Manifest: readiness → `/readyz` on the
+  metrics port, `terminationGracePeriodSeconds` ≥ ~40s, RBAC pods:list,watch, the
+  downward-API `POD_NAME`/`POD_NAMESPACE`, `CLOUD_PEER_SELECTOR` (all in `helm/cloud`).
+- **Proof.** `internal/org/rollingupgrade_test.go` rolls 3 pods over 8 orgs with
+  continuous writes and asserts zero lost acked writes, zero split-brain (no
+  (org,round) acked by two pods), and continuous availability — across both a pod
+  restart (fresh rehydrate) and an in-place ownership flap (M3, no restart).
+- **Two extensibility seams (for the tiered-storage perf pass).** The fence's
+  `ConditionalStore` is constructed in `buildDurability`, so a KV read/write-through
+  cache (L1 over the S3 L2) wraps it as a one-line decorator. The ship mechanism is a
+  swappable `snapshotCodec` (default `wholeFile`), so WAL-frame delta shipping
+  replaces it without touching the fence/round. `WithCheckpoint` injects the ship
+  checkpoint (`durableCheckpoint`) — the crypto envelope's re-encrypt integration
+  point: on a defer-encryption-to-checkpoint backend it MUST route through the
+  driver's re-encrypting Checkpoint so ship-before-ack reads FRESH ciphertext (P5).
+
 ## The route table has three projections, and the router is the source
 
 `serve.go` composes ONE route table and projects it three ways, all after

build.go

@@ -2,6 +2,7 @@ package cloud
 
 import (
 	"context"
+	"database/sql"
 	"encoding/base64"
 	"fmt"
 	"os"
@@ -10,6 +11,8 @@ import (
 
 	aiobject "github.com/hanzoai/ai/object"
 	"github.com/hanzoai/cloud/cek"
+	sqlitedrv "github.com/hanzoai/sqlite"
+	"github.com/hanzoai/ha"
 	"github.com/hanzoai/cloud/clients/commerceinproc"
 	"github.com/hanzoai/cloud/clients/metering"
 	"github.com/hanzoai/cloud/internal/org"
@@ -108,7 +111,7 @@ func BuildDeps(cfg *Config) Deps {
 	deps.O11y = pick(cfg, logger, "o11y", "O11y", cfg.O11yZAPAddr, clients.O11yRPCAt, clients.DisabledO11y)
 	deps.VFS = pickVFSClient(cfg, logger)
 	deps.MQ = pick(cfg, logger, "mq", "MQ", cfg.MQZAPAddr, clients.MQRPCAt, clients.DisabledMQ)
-	deps.Durable = buildDurability(cfg, logger)
+	deps.Durable, deps.LiveMembers = buildDurability(cfg, logger)
 
 	// Payments and Vault never co-resident. Disabled stub when no
 	// endpoint, otherwise RPC.
@@ -737,6 +740,11 @@ func pickVFSClient(cfg *Config, log luxlog.Logger) VFSClient {
 //     already relies on (see shardrouter.go).
 const durableBucket = "org-db"
 
+// durableProbePrefix namespaces the boot CAS-atomicity probe's throwaway objects
+// (org.ProbeCAS writes one per boot) away from the orgs/ tree. A bucket lifecycle rule
+// may reap ".probe/*"; the objects are tiny and never read after the probe.
+const durableProbePrefix = ".probe/cas-"
+
 // buildDurability constructs the deployment's HA-durability factory, or nil when the
 // deployment has no object store to be durable against (dev/single-node — every
 // OrgStore then stays local-only). It composes the SeaweedFS S3 If-Match
@@ -746,33 +754,37 @@ const durableBucket = "org-db"
 // master. Any construction failure fails SAFE to nil (local-only) rather than crash
 // the boot; an encryption-capable build with no usable cipher is REFUSED — a build
 // that promises encryption never ships plaintext snapshots to the object store.
-func buildDurability(cfg *Config, log luxlog.Logger) *Durability {
+// It returns the durable factory AND the live-members reader for the shard router (the
+// SAME election snapshot), non-nil together only when the plane is active; both nil when
+// local-only (the router then stays on the static ordinal set).
+func buildDurability(cfg *Config, log luxlog.Logger) (*Durability, func() []ha.Member) {
 	// A multi-replica deployment REQUIRES the durable plane: with >1 writer, a per-org
 	// store that is not hydrate-on-open + fenced is the outage this exists to fix.
 	// disabledDurability logs at the severity the replica count warrants, so a
 	// misconfigured prod deployment is never SILENTLY non-durable (Red L2).
-	multiReplica := len(parsePeers(cfg.ShardPeers)) > 1
-
-	// Explicit opt-in: the object-store fence rests on the deployed SeaweedFS enforcing
-	// conditional-PUT (If-Match) atomically, which must be validated against the deployed
-	// version before it fences real tenant data (the takeover-fence staging gate). Until
-	// CLOUD_RESEARCH_DURABLE is set the store runs local-only — the shard router still
-	// pins each org to one writer, so this is not the rolling-deploy outage; only the
-	// cross-restart object-store snapshot waits for the opt-in.
-	if !cfg.ResearchDurable {
-		disabledDurability(log, multiReplica, "CLOUD_RESEARCH_DURABLE not set — HA object-store durability is opt-in pending the SeaweedFS conditional-PUT atomicity gate")
-		return nil
-	}
+	//
+	// CLOUD_REPLICAS is the FIRST source, not CLOUD_PEERS: under live K8s membership a
+	// Deployment has no static peer list (pod names are not stable, so the chart sets only
+	// CLOUD_PEER_SELECTOR), and keying the severity on CLOUD_PEERS alone would report a
+	// 3-replica deployment whose durable plane failed to construct as "single-replica/dev"
+	// — precisely the silent non-durable state this alert exists to prevent.
+	multiReplica := cfg.Replicas > 1 || len(parsePeers(cfg.ShardPeers)) > 1
 
+	// Durability is THE path — there is no operator toggle. It self-detects capability
+	// at boot: no object store reachable (dev / native-Go / no S3 creds) → local-only,
+	// same code path, graceful; a reachable store → PROVE its conditional-PUT atomicity
+	// (org.ProbeCAS) before fencing a single byte of tenant data. A store that cannot be
+	// proven atomic fails SAFE to local-only with a loud alert — never a silent-wrong
+	// fence on a store that could split-brain.
 	admin := s3admin.New()
 	if !admin.Configured() {
 		disabledDurability(log, multiReplica, "no S3 admin creds (S3_ADMIN_* unset)")
-		return nil
+		return nil, nil
 	}
 	client, err := admin.Client()
 	if err != nil {
 		disabledDurability(log, multiReplica, fmt.Sprintf("S3 client construction failed: %v", err))
-		return nil
+		return nil, nil
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	if err := ensureDurableBucket(ctx, admin, client); err != nil {
@@ -781,16 +793,50 @@ func buildDurability(cfg *Config, log luxlog.Logger) *Durability {
 	}
 	cancel()
 
-	// Membership: CLOUD_PEERS (the shard router's set). A single-pod deployment with
-	// no peers is its own sole writer — still hydrate-on-open + fenced ship across a
-	// rolling restart.
+	// Prove the store enforces conditional-PUT atomically BEFORE fencing any tenant data
+	// (the auto-H2 self-check that replaces the old opt-in flag). A store that cannot be
+	// proven atomic fails SAFE to local-only + a loud alert — never a silent fence on a
+	// store that could admit two writers for one round (split-brain). The result is
+	// cached for the process life (deps.Durable is set once), so this probe runs once.
+	// cond is the linearizable register the fence stands on, constructed HERE (not
+	// hard-wired inside the fence) so a cache tier slots in as a decorator: a
+	// read-through/write-through KV-in-front-of-S3 store (github.com/hanzoai/kv-go) can
+	// wrap this one line to serve low-latency hydrate reads while the authoritative CAS
+	// still lands on S3 — the fence reads and CASes through whatever ConditionalStore it
+	// is handed. (Safety note for that tier: a stale cached lease read only costs a claim
+	// retry, never safety — the S3 CAS is authoritative — so a write-through cache is
+	// sound over the SAME cond that backs both the lease and the data ships.)
+	cond := org.NewS3ConditionalStore(client, durableBucket)
+	probeCtx, probeCancel := context.WithTimeout(context.Background(), 15*time.Second)
+	err = org.ProbeCAS(probeCtx, cond, durableProbePrefix)
+	probeCancel()
+	if err != nil {
+		disabledDurability(log, multiReplica, fmt.Sprintf("object-store conditional-PUT atomicity NOT confirmed — %v", err))
+		return nil, nil
+	}
+
+	// Membership: LIVE when in-cluster + CLOUD_PEER_SELECTOR is set (a rolling upgrade's
+	// changing pod set is tracked, a draining/dead pod is never elected an org's owner),
+	// else the STATIC CLOUD_PEERS/self set — capability-detected, no flag (see
+	// membership_k8s.go). A single-pod deployment with no peers is its own sole writer.
+	// The 2s refresh keeps a drained pod out of every peer's election within a bound the
+	// terminationGracePeriod covers, so a rolling handoff loses no request.
 	self := firstNonEmptyStr(strings.TrimSpace(cfg.ShardSelf), hostnameOr("cloud-0"))
 	peers := parsePeers(cfg.ShardPeers)
 	if len(peers) == 0 {
 		peers = []org.Member{{ID: self, Addr: self}}
 	}
-	members := org.NewMembership(self, org.StaticSource(peers...), 5*time.Second)
-	_ = members.Start(context.Background()) // static source: the initial refresh populates Members()
+	src := membershipSource(peers, cfg.PeerSelector, httpPortOf(cfg.ListenAddr), log)
+	members := org.NewMembership(self, src, 2*time.Second)
+	// Start's initial refresh populates Members() before the first request. REPORT its
+	// error: an empty membership is not benign here — the fencer then names no safe owner
+	// (ErrNoMembership), so every per-org store opens read-only and every write fails
+	// closed. Silently discarding this turns a missing pods:list RBAC or a selector typo
+	// into a fleet-wide write outage with no diagnostic at all.
+	if err := members.Start(context.Background()); err != nil {
+		log.Error("initial writer-membership refresh FAILED — until it succeeds no org has a safe owner and every per-org write fails closed",
+			"selector", cfg.PeerSelector, "self", self, "err", err)
+	}
 
 	cipher := durableCipher(cfg, log)
 	if cipher == nil && cek.Encrypting() {
@@ -798,11 +844,60 @@ func buildDurability(cfg *Config, log luxlog.Logger) *Durability {
 		// misconfig, not a dev path: never ship plaintext snapshots AND never silently
 		// drop durability — fail closed and log LOUDLY for the replica count.
 		disabledDurability(log, multiReplica, "encryption-capable build but no durable cipher (would ship plaintext snapshots)")
-		return nil
+		return nil, nil
 	}
 
-	log.Info("durability enabled", "bucket", durableBucket, "self", self, "peers", len(peers), "encrypted", cipher != nil)
-	return org.NewDurability(org.NewS3ConditionalStore(client, durableBucket), members, cipher)
+	log.Info("durability enabled", "bucket", durableBucket, "self", self, "peers", len(peers), "atomic_cas", true, "encrypted", cipher != nil)
+	// members.Members is the live election snapshot; hand it to the shard router so it
+	// routes on the SAME set the fencer elects over — the store-layer owner and the routed
+	// owner never disagree. WithCheckpoint WIRES the ship checkpoint (durableCheckpoint) so
+	// ship-before-ack folds the WAL into the real path before reading it — the crypto
+	// envelope's re-encrypt integration point (P5).
+	return org.NewDurability(cond, members, cipher, org.WithCheckpoint(durableCheckpoint)), members.Members
+}
+
+// durableCheckpoint makes the real on-disk file reflect every committed write before a
+// durable ship reads it — the checkpoint the codec runs before snapshotting (WithCheckpoint).
+// Two steps, correct on every backend:
+//
+//  1. A TRUNCATE checkpoint with the busy fail-closed guard: busy!=0 means a reader held the
+//     WAL so the main file is missing committed frames, and shipping it would silently lose an
+//     acked write. This folds the WAL and — on the WRITE-time-encrypting backends (cgo
+//     libsqlcipher page-level, and plaintext) — leaves the real path already fresh.
+//  2. sqlitedrv.Checkpoint re-encrypts the pure-Go ENVELOPE backend's real path. The envelope
+//     defers encryption to Checkpoint/Close, so after step 1 the real path is STALE ciphertext
+//     until re-encrypted; without this the ship reads stale bytes and loses acked writes on
+//     takeover (the envelope backend landed in hanzoai/sqlite v0.4.0 — every pure-Go and
+//     mislinked-cgo keyed open routes through it). It is a successful no-op on the write-time
+//     backends, so it runs unconditionally.
+//
+// Step 1's connection is released before step 2 so the envelope re-encrypt sees a clean handle.
+func durableCheckpoint(ctx context.Context, db *sql.DB) error {
+	if err := walCheckpointTruncate(ctx, db); err != nil {
+		return err
+	}
+	if err := sqlitedrv.Checkpoint(db); err != nil {
+		return fmt.Errorf("durable checkpoint re-encrypt: %w", err)
+	}
+	return nil
+}
+
+// walCheckpointTruncate folds the WAL into the main file with the busy fail-closed guard,
+// on its own connection (released on return, before the envelope re-encrypt).
+func walCheckpointTruncate(ctx context.Context, db *sql.DB) error {
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return fmt.Errorf("durable checkpoint conn: %w", err)
+	}
+	defer conn.Close()
+	var busy, logFrames, checkpointed int
+	if err := conn.QueryRowContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &logFrames, &checkpointed); err != nil {
+		return fmt.Errorf("durable checkpoint: %w", err)
+	}
+	if busy != 0 {
+		return fmt.Errorf("durable checkpoint did not complete (busy=%d, log=%d, checkpointed=%d) — refusing to ship a snapshot missing committed WAL frames", busy, logFrames, checkpointed)
+	}
+	return nil
 }
 
 // disabledDurability records that the durable plane is OFF, at ERROR when the

cek/cek.go

@@ -109,25 +109,60 @@ func SetMasterKey(k []byte) {
 	}
 }
 
-// resolveMaster resolves the process master key exactly once:
+// EnsureDevKey installs a deterministic development master key when NO key is
+// configured AND the live codec is not linked — i.e. a pure-Go dev/CI build. It
+// lets that build run through the SAME encrypted path as production (per-db DEK,
+// SQLCipher-format envelope) with zero configuration, instead of a divergent
+// plaintext path. It is a NO-OP when a key is already configured (production uses
+// it) or the live codec is linked (a production binary, which MUST supply the real
+// KMS key and fails closed without it). Call once at boot, before the first Open.
+// Returns true when a dev key was installed. Not safe against a concurrent Open —
+// resolveMaster caches on first use, so this must run first.
+func EnsureDevKey() bool {
+	if len(masterOverride) == 32 || strings.TrimSpace(os.Getenv(masterKeyEnv)) != "" {
+		return false // a real key is configured — use it
+	}
+	if sqlitedrv.CodecLinked() {
+		return false // production build — require the real key (resolveMaster fails closed)
+	}
+	SetMasterKey(devKey())
+	return true
+}
+
+// devKey derives the deterministic development master key. It is intentionally
+// well-known — dev/CI data is not secret — and exists only so a pure-Go build
+// encrypts through the production code path rather than diverging to plaintext. It
+// is never reachable on a codec-linked (production) build; EnsureDevKey gates it.
+func devKey() []byte {
+	sum := sha256.Sum256([]byte("hanzo-cloud-dev-cek-master-v1"))
+	return sum[:]
+}
+
+// resolveMaster resolves the process master key exactly once. Every build encrypts
+// a keyed store — the live libsqlcipher codec when it is linked, the pure-Go codec
+// envelope otherwise (EncryptionAvailable is always true) — so a store is either
+// keyed-and-encrypted or it does not open. There is no plaintext-at-rest mode:
 //
-//	(key, nil)   — 32-byte key AND an encryption-capable build → encrypt.
-//	(nil, nil)   — no key AND a non-encrypting (pure-Go) build → dev/CI plaintext.
-//	(nil, error) — key malformed; OR key set on a non-encrypting build; OR NO key
-//	               on an encryption-capable build. The last is the production
-//	               fail-closed: a capable binary never silently ships plaintext.
+//	(key, nil)   — 32-byte key configured → encrypt (live codec or envelope; both
+//	               write ciphertext at rest).
+//	(nil, error) — no key, OR a malformed key. Opening the data plane with no key is
+//	               the fatal case: a build that can encrypt must never ship plaintext.
+//	               Dev/CI supplies a deterministic dev key at boot (SetMasterKey) so
+//	               it runs encrypted with zero config; production supplies the KMS key.
 func resolveMaster() ([]byte, error) {
 	masterOnce.Do(func() {
 		raw := masterOverride
 		if len(raw) == 0 {
 			b64 := strings.TrimSpace(os.Getenv(masterKeyEnv))
 			if b64 == "" {
+				// EncryptionAvailable is always true, so this is always fatal: a build
+				// that can encrypt never opens the data plane unencrypted. Callers that
+				// want a keyless dev run inject a dev key via SetMasterKey before Open.
 				if sqlitedrv.EncryptionAvailable() {
-					masterErr = fmt.Errorf("cek: %s is required on an encryption-capable build; "+
-						"refusing to open the data plane unencrypted (set the KMS master key, "+
-						"or run a pure-Go dev build)", masterKeyEnv)
+					masterErr = fmt.Errorf("cek: %s is required; refusing to open the data plane "+
+						"unencrypted (set the KMS master key, or inject a dev key at boot)", masterKeyEnv)
 				}
-				return // pure-Go dev/CI: plaintext is expected (no codec linked)
+				return
 			}
 			decoded, err := base64.StdEncoding.DecodeString(b64)
 			if err != nil {
@@ -140,20 +175,17 @@ func resolveMaster() ([]byte, error) {
 			masterErr = fmt.Errorf("cek: master key must decode to 32 bytes, got %d", len(raw))
 			return
 		}
-		if !sqlitedrv.EncryptionAvailable() {
-			masterErr = fmt.Errorf("cek: %s is set but this build cannot encrypt (pure-Go sqlite); "+
-				"rebuild CGO_ENABLED=1 linked against libsqlcipher, or unset it for a dev build", masterKeyEnv)
-			return
-		}
+		// A configured key always encrypts: the live codec when linked, the pure-Go
+		// codec envelope otherwise. There is no "key set but cannot encrypt" case.
 		masterKey = raw
 	})
 	return masterKey, masterErr
 }
 
 // Encrypting reports whether cek will encrypt at rest (a valid master key is
-// configured on an encryption-capable build). cloud calls this once at boot for
-// the posture log; a false result on a capable build means resolveMaster errored
-// and the first store Open will fail closed.
+// configured). cloud calls this once at boot for the posture log; a false result
+// means resolveMaster errored (no or invalid key) and the first store Open will
+// fail closed rather than write plaintext.
 func Encrypting() bool {
 	k, err := resolveMaster()
 	return err == nil && len(k) == 32
@@ -168,9 +200,9 @@ func Open(path string) (*sql.DB, error) {
 		return nil, err
 	}
 	if master == nil {
-		// Only reachable on a non-encrypting dev/CI build (a capable build with no
-		// key already errored above). Preserve the prior bare-path behavior.
-		return sql.Open("sqlite", path)
+		// Unreachable: resolveMaster returns a 32-byte key or a non-nil error, never
+		// (nil, nil). Fail closed anyway — cek never opens a store plaintext.
+		return nil, fmt.Errorf("cek: no master key resolved for %q", path)
 	}
 	return openEncrypted(path, master)
 }
@@ -339,6 +371,16 @@ func openExisting(path string, master []byte) (*sql.DB, error) {
 // the encrypted copy reproduces the source schema + per-table content hash +
 // integrity_check, re-opened via the exact keyed path the app uses).
 func migrateThenOpen(path string, master []byte) (*sql.DB, error) {
+	// Converting an existing plaintext database to SQLCipher uses libsqlcipher's
+	// ATTACH ... KEY + sqlcipher_export (see exportPlaintext), which only the live C
+	// codec provides. The pure-Go envelope opens and creates encrypted stores but
+	// cannot run that in-engine conversion; refuse cleanly rather than fail deep in
+	// the export. Legacy-plaintext migration is a production operation, and
+	// production links the live codec; a pure-Go dev/CI run starts from fresh
+	// encrypted stores (createFresh) instead.
+	if !sqlitedrv.CodecLinked() {
+		return nil, fmt.Errorf("cek: converting plaintext database %q to encrypted requires the live libsqlcipher codec; run a libsqlcipher-linked build to migrate it, or remove it to start from a fresh encrypted store", path)
+	}
 	dekPath := path + dekSuffix
 	// Plaintext header ⇒ an earlier attempt did not commit: discard any stale
 	// sidecar/tmp and redo from the plaintext source of truth.

cek/cek_test.go

@@ -46,8 +46,8 @@ func requireCipher(t *testing.T) {
 		}
 		t.Skip(msg)
 	}
-	if !sqlitedrv.EncryptionAvailable() {
-		skipOrFail("sqlite build cannot encrypt (pure-Go); run with CGO + libsqlcipher")
+	if !sqlitedrv.CodecLinked() {
+		skipOrFail("sqlite build lacks the live libsqlcipher codec (pure-Go envelope encrypts, but these tests migrate via sqlcipher_export); run with CGO + libsqlcipher")
 		return
 	}
 	probe := filepath.Join(t.TempDir(), "probe.db")
@@ -317,6 +317,38 @@ func TestMissingKeyFatalOnCapableBuild(t *testing.T) {
 	}
 }
 
+// TestEnsureDevKeyEncryptsOnPureGo proves the zero-config dev posture: a pure-Go
+// build with no configured key installs a deterministic dev key and runs through
+// the SAME encrypted path as production — the store is ciphertext at rest, never
+// plaintext. On a codec-linked (production) build EnsureDevKey is a no-op.
+func TestEnsureDevKeyEncryptsOnPureGo(t *testing.T) {
+	if sqlitedrv.CodecLinked() {
+		t.Skip("codec-linked build: EnsureDevKey is a no-op; production requires the real key")
+	}
+	resetMaster(nil)
+	os.Unsetenv(masterKeyEnv)
+
+	if !EnsureDevKey() {
+		t.Fatal("EnsureDevKey should install a dev key on a pure-Go build with no key")
+	}
+	if EnsureDevKey() {
+		t.Fatal("EnsureDevKey must be idempotent — a no-op once a key is installed")
+	}
+
+	path := filepath.Join(t.TempDir(), "settings.db")
+	db, err := Open(path)
+	if err != nil {
+		t.Fatalf("open with dev key: %v", err)
+	}
+	if _, err := db.Exec(`CREATE TABLE t(x)`); err != nil {
+		t.Fatalf("ddl: %v", err)
+	}
+	_ = db.Close()
+	if isPlaintextHeader(path) {
+		t.Fatal("SECURITY: dev-key store is plaintext at rest")
+	}
+}
+
 // TestContentHashCatchesMutation (RED MED #4): the content hash catches a value
 // change that row-count + schema + integrity_check all miss, AND does NOT
 // false-positive on the benign implicit-rowid renumbering sqlcipher_export does.

clients/books/bank_api.go

@@ -249,7 +249,14 @@ func (s *state) syncDurable(org string, sandbox bool, posted int) {
 	if sandbox {
 		store = s.sandbox
 	}
-	if _, err := store.Sync(org, ""); err != nil {
+	// acked==false = the fence refused the ship (this pod was deposed): the imported
+	// postings live only in a local file the next hydrate overwrites. Unlike syncLedger's
+	// commerce ingest, an OFX/CSV import is NOT re-derivable server-side, so losing it
+	// loses user-supplied data — never let that pass silently.
+	if acked, err := store.Sync(org, ""); err != nil {
 		s.log.Warn("books bank durable sync degraded", "org", org, "sandbox", sandbox, "err", err)
+	} else if !acked {
+		s.log.Error("books bank postings NOT durably shipped — this pod is not the org's elected writer; the postings will be dropped on the next hydrate, re-import to recover",
+			"org", org, "sandbox", sandbox, "posted", posted)
 	}
 }

clients/books/books.go

@@ -155,8 +155,16 @@ func (s *state) syncLedger(ctx context.Context, org string, sandbox bool) (int,
 		if sandbox {
 			store = s.sandbox
 		}
-		if _, serr := store.Sync(org, ""); serr != nil {
+		// acked==false means this pod was deposed mid-request and the fence REFUSED the
+		// ship: the postings are committed only to a local file a successor's hydrate will
+		// overwrite. Report it — this ingest is idempotent, so a re-run recovers, but it
+		// must never look like a clean success. (Since the durable plane became the default
+		// this is a live path, not a no-op.)
+		if acked, serr := store.Sync(org, ""); serr != nil {
 			s.log.Warn("books durable sync degraded", "org", org, "sandbox", sandbox, "err", serr)
+		} else if !acked {
+			s.log.Error("books postings NOT durably shipped — this pod is not the org's elected writer; the postings will be dropped on the next hydrate, re-run the sync",
+				"org", org, "sandbox", sandbox, "posted", posted)
 		}
 	}
 	return posted, nil

clients/provisioning/addon_envtest_test.go

@@ -57,7 +57,7 @@ func TestPatchAddonSecret_RealAPIServer(t *testing.T) {
 	}
 
 	const name = "commerce-addons"
-	const kvURL = "redis://default:pw@kv:6379"
+	const kvURL = "kv://default:pw@kv:6379"
 	const sqlURL = "postgres://admin:pw@sql:5432/db?sslmode=disable"
 
 	// data reads the live Secret and base64-decodes .data into plain strings.

clients/provisioning/addon_test.go

@@ -144,7 +144,7 @@ func TestDedicated_SQLEngineAssemblesDSN(t *testing.T) {
 // TestDedicated_KVEngineAssemblesDSN proves the kv add-on materializes a valkey
 // Datastore CR that loads a per-instance requirepass from a MOUNTED config
 // Secret (the kv-server binary reads no password from env) and returns a
-// redis://default:… DSN — never an "admin" user that would fail AUTH.
+// kv://default:… DSN — never an "admin" user that would fail AUTH.
 func TestDedicated_KVEngineAssemblesDSN(t *testing.T) {
 	orch := newFakeOrch()
 	s := newDedicatedService(t, orch)
@@ -162,7 +162,7 @@ func TestDedicated_KVEngineAssemblesDSN(t *testing.T) {
 	if cr.Port != 6379 || cr.Username != "default" {
 		t.Fatalf("kv endpoint port=%d user=%q, want 6379/default", cr.Port, cr.Username)
 	}
-	if cr.ConnectionString != "redis://default:"+cr.Password+"@"+host+":6379" {
+	if cr.ConnectionString != "kv://default:"+cr.Password+"@"+host+":6379" {
 		t.Fatalf("dsn %q not the expected redis default-user DSN", cr.ConnectionString)
 	}
 

clients/provisioning/billing_test.go

@@ -72,7 +72,7 @@ func newBilledService(t *testing.T, commerceURL string, kinds ...string) (*cloud
 	t.Setenv("CLOUD_KMS_NODES", "")
 	t.Setenv("CLOUD_KMS_PASSPHRASE", "")
 	log := luxlog.New("module", "provbilltest")
-	mp := &mockProv{cs: "redis://u:pw@kv.hanzo.svc:6379/0", host: "kv.hanzo.svc", port: 6379, db: "prefix:"}
+	mp := &mockProv{cs: "kv://u:pw@kv.hanzo.svc:6379/0", host: "kv.hanzo.svc", port: 6379, db: "prefix:"}
 	reg := map[string]Provisioner{}
 	for _, k := range kinds {
 		reg[k] = mp

clients/provisioning/dedicated.go

@@ -217,7 +217,10 @@ var dedicatedEngines = map[string]engine{
 			return map[string]string{"kv.conf": "requirepass " + pw + "\n"}
 		},
 		dsn: func(user, pw, host string, port int, _ string) string {
-			return fmt.Sprintf("redis://%s:%s@%s:%d", user, pw, host, port)
+			// kv:// is the scheme our client parses. The Datastore spec.type stays
+			// "valkey" — that one is the operator's contract (Engine::Valkey), not
+			// ours to rename.
+			return fmt.Sprintf("kv://%s:%s@%s:%d", user, pw, host, port)
 		},
 	},
 	// datastore ("datastore"): ghcr.io/hanzoai/datastore provisions

clients/provisioning/provisioning.go

@@ -9,7 +9,7 @@
 // is naturally tenant-scoped; the assembled DSN is injected as <KIND>_URL into
 // the app instance's addons Secret, switching it off Base onto the backend:
 //
-//	kv        -> Hanzo KV        Datastore type=valkey       redis://…:6379
+//	kv        -> Hanzo KV        Datastore type=valkey       kv://…:6379
 //	sql       -> Hanzo SQL       Datastore type=postgresql   postgres://…:5432
 //	docdb     -> Hanzo DocDB     Datastore type=docdb        mongodb://…:27017
 //	datastore -> Hanzo Datastore Datastore type=datastore    datastore://…:8123

clients/provisioning/provisioning_test.go

@@ -61,7 +61,7 @@ func newTestService(t *testing.T, kinds ...string) (*cloud.Service[state], *mock
 	t.Setenv("CLOUD_KMS_NODES", "")
 	t.Setenv("CLOUD_KMS_PASSPHRASE", "")
 	log := luxlog.New("module", "provtest")
-	mp := &mockProv{cs: "redis://u:pw@kv.hanzo.svc:6379/0", host: "kv.hanzo.svc", port: 6379, db: "prefix:"}
+	mp := &mockProv{cs: "kv://u:pw@kv.hanzo.svc:6379/0", host: "kv.hanzo.svc", port: 6379, db: "prefix:"}
 	reg := map[string]Provisioner{}
 	for _, k := range kinds {
 		reg[k] = mp

config.go

@@ -139,15 +139,15 @@ type Config struct {
 	// black-hole every request by forwarding it away with no shard of its own).
 	ShardSelf string
 
-	// ResearchDurable opts the per-org store into the HA object-store durability plane
-	// (hydrate-on-open + fenced ship to org-db, CLOUD_RESEARCH_DURABLE). It is OFF by
-	// default so a deploy never begins fencing real tenant data on the object store's
-	// conditional-PUT atomicity before that atomicity is validated against the deployed
-	// SeaweedFS version (the takeover-fence staging gate). With it off the store runs
-	// local-only per pod — still single-writer-per-org via the shard router (ha.Owner) —
-	// so the DDL-drift-proof + evidence-preserving record layer is fully active; only the
-	// cross-restart object-store snapshot/fence waits for the opt-in.
-	ResearchDurable bool
+	// PeerSelector is the Kubernetes label selector that names this deployment's writer
+	// pods (CLOUD_PEER_SELECTOR, e.g. "app.kubernetes.io/name=cloud,app.kubernetes.io/
+	// instance=<release>"). When set AND running in-cluster, membership is LIVE — the
+	// binary lists Ready, non-terminating pods matching it via the K8s API, so a rolling
+	// upgrade's changing pod set is tracked and a draining/dead pod is never elected an
+	// org's owner. Empty (dev / native-Go / not in a cluster) falls back to the STATIC
+	// ShardPeers/self set — capability-detected, no on/off flag. Deployment wiring the
+	// chart sets, not a feature toggle.
+	PeerSelector string
 
 	// ListenAddr is the public HTTP listener (default :8080).
 	ListenAddr string
@@ -431,9 +431,9 @@ func LoadConfig() *Config {
 		WriterURL:         strings.TrimRight(getenv("CLOUD_WRITER_URL", ""), "/"),
 		ReaderRetryBudget: getenvDuration("CLOUD_READER_RETRY_BUDGET", 25*time.Second),
 		WriterLease:       getenvBool("CLOUD_WRITER_LEASE"),
-		ResearchDurable:   getenvBool("CLOUD_RESEARCH_DURABLE"),
 		ShardPeers:        getenv("CLOUD_PEERS", ""),
 		ShardSelf:         firstNonEmptyStr(getenv("CLOUD_POD_NAME", ""), getenv("POD_NAME", "")),
+		PeerSelector:      getenv("CLOUD_PEER_SELECTOR", ""),
 		PaymentsZAPAddr:   getenv("CLOUD_PAYMENTS_ZAP_ADDR", ""),
 		VaultZAPAddr:      getenv("CLOUD_VAULT_ZAP_ADDR", ""),
 		// Billing gate (KMS-backed COMMERCE_SERVICE_TOKEN; never plaintext).

deps.go

@@ -11,6 +11,7 @@ package cloud
 
 import (
 	"github.com/hanzoai/cloud/clients/metering"
+	"github.com/hanzoai/ha"
 	luxlog "github.com/luxfi/log"
 
 	"github.com/hanzoai/cloud/audit"
@@ -63,6 +64,15 @@ type Deps struct {
 	// dev/single-node), and every OrgStore is exactly the pre-durability cache.
 	Durable *Durability
 
+	// LiveMembers reads the CURRENT live writer set (the SAME ha.Membership snapshot the
+	// durability fencer elects over). Non-nil ONLY when the durable plane is active — the
+	// shard router then routes on the live set, so a draining/dead pod's orgs go to the
+	// ready successor that hydrates them (M3), not to the gone pod. nil ⇒ the router falls
+	// back to the static CLOUD_PEERS set: without the durable plane a peer cannot serve
+	// another pod's local-only files, so ownership must stay pinned to the ordinal (which
+	// reattaches its PVC across a restart). One field gates the whole live-routing path.
+	LiveMembers func() []ha.Member
+
 	// AIDefaultModel is the served model a subsystem uses when a caller supplies
 	// none (CLOUD_AI_DEFAULT_MODEL, default deepseek-v4-flash). It is the ONE
 	// cloud-side model default, sourced from config so no subsystem hardcodes a

drain.go

@@ -0,0 +1,36 @@
+package cloud
+
+// drain.go is the graceful-drain readiness signal that makes a rolling upgrade lose no
+// request. On SIGTERM the process flips to draining and /readyz returns 503, so Kubernetes
+// marks the pod NotReady: it is removed from Service endpoints AND — via the live
+// membership source's terminating/ready gate (membership_k8s.go) — from every peer's
+// writer election. An org this pod owned is then re-elected to a live successor that
+// hydrates its latest fenced snapshot (M3), BEFORE this pod stops serving. The process
+// stays UP through terminationGracePeriodSeconds to drain in-flight requests and ship
+// final state (OrgStore.CloseAll, ship-before-close), so no acknowledged write is lost.
+//
+// Liveness (/healthz) stays 200 throughout — we want the process alive to finish the
+// drain; only readiness (/readyz) flips, the standard K8s "stop new traffic, keep me
+// running to finish" contract.
+
+import (
+	"sync/atomic"
+	"time"
+)
+
+// draining flips true when a graceful shutdown begins. Read lock-free by the /readyz
+// handler on the ops listener.
+var draining atomic.Bool
+
+// SetDraining marks the process draining (readiness → NotReady). Idempotent.
+func SetDraining() { draining.Store(true) }
+
+// Draining reports whether a graceful shutdown is in progress.
+func Draining() bool { return draining.Load() }
+
+// shardDrainGrace is how long a draining pod stays serving (NotReady) so peers observe it
+// leave the membership and re-elect its orgs to live successors before it tears down. It
+// is a few membership-refresh intervals (membership_k8s.go refreshes at 2s), covered by
+// terminationGracePeriodSeconds. Applied only when sharding is active (multi-pod); a
+// single-pod deployment has no successor to hand off to, so it drains immediately.
+const shardDrainGrace = 6 * time.Second

durablecheckpoint_test.go

@@ -0,0 +1,80 @@
+package cloud
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	sqlitedrv "github.com/hanzoai/sqlite"
+)
+
+// TestDurableCheckpointReEncryptsEnvelope proves durableCheckpoint makes the real
+// on-disk path reflect every committed write on the pure-Go codec ENVELOPE backend —
+// where encryption is deferred to Checkpoint/Close, so a raw wal_checkpoint alone
+// leaves the real path stale and a fenced ship would restore that stale snapshot,
+// silently losing an acked write on takeover (the HIGH-1 defect, PoC-confirmed).
+//
+// The store handle is kept OPEN across the checkpoint and the read: closing it would
+// itself re-encrypt (Close seals), masking whether durableCheckpoint did so. On the
+// write-time backends (cgo libsqlcipher, plaintext) the store persists per-commit and
+// the re-encrypt step is a no-op, so this passes on both lanes.
+func TestDurableCheckpointReEncryptsEnvelope(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "org.db")
+	key := make([]byte, 32)
+	for i := range key {
+		key[i] = byte(i + 1)
+	}
+
+	// Seed the encrypted store with row A and seal it (Close re-encrypts the real path).
+	seed, err := sqlitedrv.OpenDB(path, key)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if _, err := seed.Exec(`CREATE TABLE t(v TEXT)`); err != nil {
+		t.Fatalf("ddl: %v", err)
+	}
+	if _, err := seed.Exec(`INSERT INTO t VALUES('A')`); err != nil {
+		t.Fatalf("insert A: %v", err)
+	}
+	if err := seed.Close(); err != nil {
+		t.Fatalf("seal: %v", err)
+	}
+
+	// The live store handle Durable.Bind lends to Sync. Commit row B — an ACKED write.
+	handle, err := sqlitedrv.OpenDB(path, key)
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer handle.Close()
+	if _, err := handle.Exec(`INSERT INTO t VALUES('B')`); err != nil {
+		t.Fatalf("insert B: %v", err)
+	}
+
+	// The ship checkpoint. After it, the real-path bytes readFramed ships MUST include B.
+	if err := durableCheckpoint(context.Background(), handle); err != nil {
+		t.Fatalf("durableCheckpoint: %v", err)
+	}
+
+	// A fresh open reads the CURRENT real-path ciphertext — exactly what a successor
+	// CarryForward-restores on takeover. It must see both acked writes, not a stale {A}.
+	if got := realPathRows(t, path, key); got != 2 {
+		t.Fatalf("SILENT LOST WRITE: real path has %d row(s) after durableCheckpoint, want 2 "+
+			"(acked write 'B' dropped → lost on takeover)", got)
+	}
+}
+
+// realPathRows opens the store fresh (decrypting the current real-path ciphertext) and
+// returns the row count — the view a successor sees after CarryForward.
+func realPathRows(t *testing.T, path string, key []byte) int {
+	t.Helper()
+	db, err := sqlitedrv.OpenDB(path, key)
+	if err != nil {
+		t.Fatalf("open real path: %v", err)
+	}
+	defer db.Close()
+	var n int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&n); err != nil {
+		t.Fatalf("count: %v", err)
+	}
+	return n
+}

go.mod

@@ -27,7 +27,7 @@ require (
 	github.com/hanzoai/pubsub v1.0.0
 	github.com/hanzoai/s3-go v1.0.0
 	github.com/hanzoai/sign v1.0.0
-	github.com/hanzoai/sqlite v0.3.2
+	github.com/hanzoai/sqlite v0.4.0
 	github.com/hanzoai/stream v1.2.0
 	github.com/lib/pq v1.12.3
 	github.com/luxfi/log v1.5.0
@@ -90,7 +90,7 @@ require (
 	github.com/hanzoai/go-cosyvoice v1.0.0 // indirect
 	github.com/hanzoai/go-openai-realtime v1.0.0 // indirect
 	github.com/hanzoai/go-openai-realtime/contrib/ws-gorilla v1.0.0 // indirect
-	github.com/hanzoai/sqlcipher v0.1.0 // indirect
+	github.com/hanzoai/sqlcipher v0.1.1 // indirect
 	github.com/hanzoai/xorm v1.4.4 // indirect
 	github.com/hanzokv/go/extra/kvcmd/v9 v9.22.0 // indirect
 	github.com/hanzokv/go/extra/kvotel/v9 v9.22.0 // indirect
@@ -116,6 +116,10 @@ require (
 	k8s.io/controller-manager v0.34.0 // indirect
 	k8s.io/kube-aggregator v0.34.0 // indirect
 	k8s.io/kubectl v0.34.0 // indirect
+	modernc.org/libc v1.72.3 // indirect
+	modernc.org/mathutil v1.7.1 // indirect
+	modernc.org/memory v1.11.0 // indirect
+	modernc.org/sqlite v1.51.0 // indirect
 	rsc.io/qr v0.2.0 // indirect
 )
 

go.sum

@@ -1144,10 +1144,10 @@ github.com/hanzoai/sendgrid-go v3.4.2-0.20180724185151-733a05184a8d+incompatible
 github.com/hanzoai/sendgrid-go v3.4.2-0.20180724185151-733a05184a8d+incompatible/go.mod h1:HTKx0R5LWWt7H1eG4pCpNS4PVO6tZIUVVZN/80sHz9o=
 github.com/hanzoai/sign v1.0.0 h1:xqrMzBHp/m7mxAsYgV/1/kT7x5GEXYOXfu9xbiKDeNg=
 github.com/hanzoai/sign v1.0.0/go.mod h1:pL+4Kwh+W2AsMaU0QUF/OJoe9eV/CWHhO07Z5KV2SOg=
-github.com/hanzoai/sqlcipher v0.1.0 h1:V9gKG3ZltN2ZCteDrOnXWfOeEe/YDhhUm9AorQEAuBo=
-github.com/hanzoai/sqlcipher v0.1.0/go.mod h1:F0soUYM1i4sawOZUpRvVnWoUayPbeGVlGq01VXy9Aqg=
-github.com/hanzoai/sqlite v0.3.2 h1:B/TRunlIDZECEypmr6rHeNyWf37YZXvPJHBDEFMKn/g=
-github.com/hanzoai/sqlite v0.3.2/go.mod h1:a3llsefKbu2Iq/0rJ1mlWCaU2t2cXh+aze85x+oW72k=
+github.com/hanzoai/sqlcipher v0.1.1 h1:GARjSiUEa1lwhd1/f87XRaujZBG5s1ZwxrZW2Es/ADI=
+github.com/hanzoai/sqlcipher v0.1.1/go.mod h1:F0soUYM1i4sawOZUpRvVnWoUayPbeGVlGq01VXy9Aqg=
+github.com/hanzoai/sqlite v0.4.0 h1:xlF6w6xzn2VSeAVBKFrZfIq0ISqRSTHz2g3QN3eCcnE=
+github.com/hanzoai/sqlite v0.4.0/go.mod h1:sE5uvENQ+q+rI011AwFuAuJ7DwOoDnywyCQstZ0ekHQ=
 github.com/hanzoai/stream v1.2.0 h1:AVKg/YBgzZ/x8hGjg+bnpldMR34x1FzSp84GVdAPhfI=
 github.com/hanzoai/stream v1.2.0/go.mod h1:mn5cnMQYzdGh5vlslCnh1fdI79qFvbet4MpvY0ME5YU=
 github.com/hanzoai/tasks v1.51.4 h1:qMx80ZfiJPQIDd9lpxxJhDywKYnHOmKxL2IfhZfExgk=
@@ -3154,6 +3154,34 @@ k8s.io/metrics v0.35.3 h1:WonA18pEwrtb7a6XfhFg1ZY1Le0RFkcEw7CFApMTZos=
 k8s.io/metrics v0.35.3/go.mod h1:/O8UBb5QVyAekR2QvL/WWxskpdV1wVSEl4MSLAy4Ql4=
 k8s.io/utils v0.0.0-20260319190234-28399d86e0b5 h1:kBawHLSnx/mYHmRnNUf9d4CpjREbeZuxoSGOX/J+aYM=
 k8s.io/utils v0.0.0-20260319190234-28399d86e0b5/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk=
+modernc.org/cc/v4 v4.28.2 h1:3tQ0lf2ADtoby2EtSP+J7IE2SHwEJdP8ioR59wx7XpY=
+modernc.org/cc/v4 v4.28.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI=
+modernc.org/ccgo/v4 v4.34.0 h1:yRLPFZieg532OT4rp4JFNIVcquwalMX26G95WQDqwCQ=
+modernc.org/ccgo/v4 v4.34.0/go.mod h1:AS5WYMyBakQ+fhsHhtP8mWB82KTGPkNNJDGfGQCe0/A=
+modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM=
+modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU=
+modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
+modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
+modernc.org/gc/v3 v3.1.2 h1:ZtDCnhonXSZexk/AYsegNRV1lJGgaNZJuKjJSWKyEqo=
+modernc.org/gc/v3 v3.1.2/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
+modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
+modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
+modernc.org/libc v1.72.3 h1:ZnDF4tXn4NBXFutMMQC4vtbTFSXhhKzR73fv0beZEAU=
+modernc.org/libc v1.72.3/go.mod h1:dn0dZNnnn1clLyvRxLxYExxiKRZIRENOfqQ8XEeg4Qs=
+modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
+modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
+modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
+modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
+modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg=
+modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
+modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
+modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
+modernc.org/sqlite v1.51.0 h1:aH/MMSoayAIhozZ7uJbVTT9QO/VhzBf0J9tymmmuC/U=
+modernc.org/sqlite v1.51.0/go.mod h1:tcNzv5p84E0skkmJn038y+hWJbLQXQqEnQfeh5r2JLM=
+modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
+modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
+modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
+modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/pdf v0.1.1 h1:k1MczvYDUvJBe93bYd7wrZLLUEcLZAuF824/I4e5Xr4=
 rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=

helm/cloud/templates/_helpers.tpl

@@ -30,3 +30,11 @@ app.kubernetes.io/managed-by: {{ .Release.Service }}
 app.kubernetes.io/name: {{ include "cloud.name" . }}
 app.kubernetes.io/instance: {{ .Release.Name }}
 {{- end }}
+
+{{- define "cloud.serviceAccountName" -}}
+{{- if .Values.serviceAccount.create }}
+{{- default (include "cloud.fullname" .) .Values.serviceAccount.name }}
+{{- else }}
+{{- default "default" .Values.serviceAccount.name }}
+{{- end }}
+{{- end }}

helm/cloud/templates/deployment.yaml

@@ -27,6 +27,10 @@ spec:
       {{- with .Values.imagePullSecrets }}
       imagePullSecrets: {{- toYaml . | nindent 8 }}
       {{- end }}
+      serviceAccountName: {{ include "cloud.serviceAccountName" . }}
+      # Exceed the in-process drain + final ship so a rolling pod hands off cleanly before
+      # SIGKILL (see drain.go / serve.go shutdown).
+      terminationGracePeriodSeconds: {{ .Values.terminationGracePeriodSeconds }}
       securityContext: {{- toYaml .Values.podSecurityContext | nindent 8 }}
       containers:
         - name: cloud
@@ -38,11 +42,32 @@ spec:
             - "--data-dir={{ .Values.hanzo.dataDir }}"
             - "--iam-issuer={{ .Values.hanzo.iamIssuer }}"
             - "--enable={{ .Values.hanzo.enable }}"
+          # preStop belt-and-suspenders: give the endpoint controller a moment to remove
+          # this pod before SIGTERM so no request is dialed to a tearing-down pod. The
+          # in-process drain (drain.go) is the primary mechanism; this only sharpens it.
+          lifecycle:
+            preStop:
+              exec:
+                command: ["/bin/sh", "-c", "sleep 3"]
           env:
             # Mirrors the resolved replica count so Config.Validate can enforce the
             # single-replica contract embedded IAM requires (memory session store).
             - name: CLOUD_REPLICAS
               value: {{ $replicas | quote }}
+            # Stable pod identity (HRW weight) + namespace for the live membership source.
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: POD_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+            # Label selector that names this deployment's writer pods, so the live
+            # membership source (membership_k8s.go) tracks the pod set in-cluster. Empty /
+            # out-of-cluster falls back to the static self set — no flag, works everywhere.
+            - name: CLOUD_PEER_SELECTOR
+              value: "app.kubernetes.io/name={{ include "cloud.name" . }},app.kubernetes.io/instance={{ .Release.Name }}"
           ports:
             - name: http
               containerPort: 8080

helm/cloud/templates/rbac.yaml

@@ -0,0 +1,36 @@
+{{- if .Values.serviceAccount.create }}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ include "cloud.serviceAccountName" . }}
+  labels: {{- include "cloud.labels" . | nindent 4 }}
+{{- end }}
+{{- if .Values.rbac.create }}
+---
+# The live membership source (membership_k8s.go) lists+watches this deployment's own pods
+# to track the writer set for zero-downtime rolling upgrades. Namespaced, read-only, scoped
+# to pods — the least privilege that lets a replica see its peers leave/join.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: {{ include "cloud.fullname" . }}-membership
+  labels: {{- include "cloud.labels" . | nindent 4 }}
+rules:
+  - apiGroups: [""]
+    resources: ["pods"]
+    verbs: ["get", "list", "watch"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: {{ include "cloud.fullname" . }}-membership
+  labels: {{- include "cloud.labels" . | nindent 4 }}
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: {{ include "cloud.fullname" . }}-membership
+subjects:
+  - kind: ServiceAccount
+    name: {{ include "cloud.serviceAccountName" . }}
+    namespace: {{ .Release.Namespace }}
+{{- end }}

helm/cloud/values.yaml

@@ -13,6 +13,23 @@ imagePullSecrets: []
 
 replicaCount: 1
 
+# rbac grants the pod's ServiceAccount pods:list,watch so the LIVE membership source
+# (membership_k8s.go) can track the writer set for zero-downtime rolling upgrades. When
+# the durable plane is active, every replica elects org owners over this live set, so a
+# draining/dead pod is never routed an org's traffic. Disable only for a single-replica
+# deployment that never scales (the source falls back to self, no API access needed).
+rbac:
+  create: true
+serviceAccount:
+  create: true
+  name: ""
+
+# terminationGracePeriodSeconds MUST exceed the in-process drain (shardDrainGrace, ~6s,
+# during which the pod stays serving NotReady so peers re-elect its orgs) PLUS the shutdown
+# window (in-flight drain + final fenced ship, up to ~30s). 60s leaves headroom; a value
+# below ~40s risks SIGKILL before a pod ships its last writes.
+terminationGracePeriodSeconds: 60
+
 # Brand / domain / enabled subsystems mirror the binary's CLI flags
 # (cmd/cloud/main.go).
 hanzo:
@@ -74,9 +91,14 @@ probes:
       port: http
     initialDelaySeconds: 30
     periodSeconds: 30
+  # Readiness targets /readyz on the ops (metrics) listener, which returns 503 the moment
+  # the process begins a graceful shutdown (drain.go) — so Kubernetes marks the pod
+  # NotReady and stops routing to it BEFORE it tears down. Liveness stays on /health (200
+  # throughout the drain) so the pod is never killed mid-drain. Keep readiness on the
+  # metrics port: /readyz there is the ONLY drain-aware endpoint.
   readiness:
     httpGet:
-      path: /health
-      port: http
+      path: /readyz
+      port: metrics
     initialDelaySeconds: 5
-    periodSeconds: 10
+    periodSeconds: 5

internal/org/checkpoint_test.go

@@ -0,0 +1,145 @@
+// Copyright 2025 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package org
+
+// checkpoint_test.go pins the crypto-integration contract: a fenced ship must fold the
+// WAL into the real on-disk file (and, on the pure-Go encryption envelope, re-encrypt it)
+// BEFORE reading the bytes it ships — otherwise it ships STALE state and a takeover reads
+// a lost acked write. It proves the WithCheckpoint seam runs on every Sync before the
+// read, that the shipped snapshot carries the just-committed write, and that a failing
+// checkpoint fails the ship CLOSED (never a stale ship acked).
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/hanzoai/vfs/replica"
+)
+
+// TestProduceHoldsSoleConnAcrossReadWithCheckpoint pins the torn-snapshot guard on the
+// PRODUCTION ship path. os.ReadFile takes no SQLite lock, so if produce reads the file
+// without holding the store's sole connection, a concurrent commit — and the WAL
+// auto-checkpoint it can trigger, which writes pages straight into the real path — tears
+// the image mid-read; the fence then ACKS that corrupt snapshot at the lease round and a
+// successor restores it. The default (nil-checkpoint) path holds the connection across
+// the read; the injected-checkpoint path, which is the one buildDurability wires for
+// production, must do the same.
+//
+// Deterministic: the sole connection (MaxOpenConns(1)) is held by the test, so a produce
+// that acquires a connection for its read CANNOT get one and fails on the ctx deadline.
+// Without the guard produce would happily read the file and return a payload.
+func TestProduceHoldsSoleConnAcrossReadWithCheckpoint(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "research.db")
+	db, err := sql.Open("sqlite", testDSN(path))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	defer db.Close()
+	db.SetMaxOpenConns(1)
+	ensureKV(t, db)
+	putKV(t, db, "k1", "v1")
+
+	// Occupy the SOLE connection, standing in for a concurrent in-flight writer.
+	held, err := db.Conn(context.Background())
+	if err != nil {
+		t.Fatalf("take sole conn: %v", err)
+	}
+	defer held.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	// A no-op checkpoint: this asserts the guard around the READ, independently of what
+	// the injected checkpoint itself does (production's also takes, and releases, a conn).
+	codec := wholeFile{checkpoint: func(context.Context, *sql.DB) error { return nil }}
+	if _, err := codec.produce(ctx, db, path); err == nil {
+		t.Fatal("produce read the database file WITHOUT holding the sole connection — a concurrent writer can tear the snapshot mid-read and the fence would ack the corrupt image")
+	}
+
+	// Control: with the connection free, the same produce succeeds and carries the write.
+	held.Close()
+	payload, err := codec.produce(context.Background(), db, path)
+	if err != nil {
+		t.Fatalf("produce with a free connection must succeed: %v", err)
+	}
+	if len(payload) == 0 {
+		t.Fatal("produce returned an empty payload")
+	}
+}
+
+// TestSyncCheckpointsBeforeShip: with an injected checkpoint (standing in for the
+// envelope's re-encrypting Checkpoint), Sync must invoke it before reading the file, and
+// the durable object must then carry the committed write — i.e. the ship is FRESH.
+func TestSyncCheckpointsBeforeShip(t *testing.T) {
+	ctx := context.Background()
+	cs := newFakeCondStore()
+	const orgID = "acme"
+	dbKey := replica.DBPath(orgID, "", "research")
+
+	var checkpoints atomic.Int64
+	dy := NewDurability(cs, &liveView{self: "solo", set: []Member{{ID: "solo"}}}, nil,
+		WithCheckpoint(func(ctx context.Context, db *sql.DB) error {
+			// Do the real WAL fold (the envelope would ALSO re-encrypt here) so the file
+			// read that follows sees the latest committed page, then record the call.
+			if _, err := db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)"); err != nil {
+				return err
+			}
+			checkpoints.Add(1)
+			return nil
+		}))
+	d := dy.For(orgID, dbKey, filepath.Join(t.TempDir(), "research.db"))
+	if err := d.Hydrate(ctx); err != nil {
+		t.Fatalf("hydrate: %v", err)
+	}
+	db := openBoundDB(t, d)
+
+	putKV(t, db, "k1", "v1")
+	if acked, err := d.Sync(ctx); err != nil || !acked {
+		t.Fatalf("sync: acked=%v err=%v", acked, err)
+	}
+	if checkpoints.Load() == 0 {
+		t.Fatal("Sync must run the injected checkpoint before shipping (envelope re-encrypt point)")
+	}
+	// The shipped durable object carries the committed write — the ship was fresh, not
+	// stale. (On the envelope backend this is exactly what guards against shipping stale
+	// ciphertext that predates k1.)
+	if v, ok := durableValue(t, ctx, cs, dbKey, "k1"); !ok || v != "v1" {
+		t.Fatalf("durable k1 = %q,%v after checkpointed ship, want v1,true", v, ok)
+	}
+}
+
+// TestSyncFailsClosedOnCheckpointError: if the checkpoint (envelope re-encrypt) fails, the
+// ship must NOT proceed — a stale snapshot shipped as complete is a silent lost write. The
+// write is not acknowledged and the durable object is untouched.
+func TestSyncFailsClosedOnCheckpointError(t *testing.T) {
+	ctx := context.Background()
+	cs := newFakeCondStore()
+	const orgID = "acme"
+	dbKey := replica.DBPath(orgID, "", "research")
+
+	dy := NewDurability(cs, &liveView{self: "solo", set: []Member{{ID: "solo"}}}, nil,
+		WithCheckpoint(func(context.Context, *sql.DB) error {
+			return fmt.Errorf("envelope re-encrypt failed")
+		}))
+	d := dy.For(orgID, dbKey, filepath.Join(t.TempDir(), "research.db"))
+	if err := d.Hydrate(ctx); err != nil {
+		t.Fatalf("hydrate: %v", err)
+	}
+	db := openBoundDB(t, d)
+
+	putKV(t, db, "k1", "v1")
+	acked, err := d.Sync(ctx)
+	if acked || err == nil {
+		t.Fatalf("a failed checkpoint must fail the ship closed: acked=%v err=%v", acked, err)
+	}
+	// Fail-closed guarantees no stale ship: Sync errors out of produce() BEFORE any
+	// fenced Put, so the write is neither acknowledged nor shipped — the successor will
+	// never read a snapshot that predates the checkpoint that could not run.
+}

internal/org/condprobe.go

@@ -0,0 +1,134 @@
+// Copyright 2025 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package org
+
+// condprobe.go is the startup self-validation that decides — with NO operator flag —
+// whether it is SAFE to run the durable+fenced plane over a given object store. The
+// fence's whole safety (fence.go, github.com/hanzoai/vfs/replica.FencedStore) rests on
+// the store enforcing conditional-PUT (If-None-Match on create, If-Match on update)
+// ATOMICALLY server-side: two racing conditional writes at the same expected version
+// must resolve to EXACTLY ONE winner. A store that silently admits both (a
+// last-write-wins gateway with no real precondition) would let two elected writers each
+// "win" a lease round and split-brain the org. So the binary PROVES the property at
+// boot against the actual deployed store, once, rather than trusting a human toggle.
+//
+// This is the atomicity gate as a self-check the binary runs — a sane default, not a
+// staging flag: durable when the store is provably atomic, local-only (fail-safe) when
+// it is not, and never probed at all when no store is reachable (native-Go / dev, which
+// is why durability degrades gracefully everywhere). buildDurability runs it exactly
+// once at boot and caches the outcome for the process life (deps.Durable is nil-or-set
+// thereafter), so it is the single "once" the way cek's master-key probe is.
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/hanzoai/vfs/replica"
+)
+
+// ErrNonAtomicStore reports that the object store did not enforce conditional-PUT
+// atomically — the caller MUST fail safe (run local-only, never fence on it), because
+// fencing on a store that admits two winners for one round is split-brain by
+// construction.
+var ErrNonAtomicStore = errors.New("org: object store does not enforce conditional-PUT atomically — refusing to fence (would risk split-brain)")
+
+// probeRacers is the concurrency each race phase fires at the store. A correct
+// compare-and-set admits exactly one; more racers only sharpen the odds of catching a
+// non-atomic gateway, at a fixed, tiny boot cost.
+const probeRacers = 4
+
+// ProbeCAS proves the store enforces compare-and-set atomically, exercising BOTH
+// preconditions the fence relies on, against a fresh throwaway key:
+//
+//	create-race : probeRacers goroutines each PutIfVersion(key, _, "") — If-None-Match
+//	              on a non-existent object — so exactly one may CREATE it.
+//	update-race : read the create winner's version, then probeRacers goroutines each
+//	              PutIfVersion(key, _, thatVersion) — If-Match — so exactly one may
+//	              ADVANCE it.
+//
+// Exactly-one-winner in BOTH phases ⇒ atomic-confirmed (nil). Any other outcome —
+// more than one winner (non-atomic ⇒ ErrNonAtomicStore), zero winners, or a hard store
+// error (unreachable / refused) — is returned for the caller to fail safe on. The probe
+// writes only under the caller's chosen throwaway prefix (a ".probe/" object per boot,
+// reapable by a bucket lifecycle rule), never into the orgs/ tree.
+func ProbeCAS(ctx context.Context, store replica.ConditionalStore, keyPrefix string) error {
+	key, err := probeKey(keyPrefix)
+	if err != nil {
+		return err
+	}
+	// Phase 1 — create-race (If-None-Match): a fresh key, so exactly one racer creates.
+	winners, err := raceCAS(ctx, store, key, "")
+	if err != nil {
+		return fmt.Errorf("org: CAS probe create-race: %w", err)
+	}
+	if winners != 1 {
+		return fmt.Errorf("%w: create-race admitted %d writers (want 1) — If-None-Match not atomic", ErrNonAtomicStore, winners)
+	}
+	// Read the created object's version to condition the update-race on it.
+	_, version, err := store.Get(ctx, key)
+	if err != nil {
+		return fmt.Errorf("org: CAS probe read-back: %w", err)
+	}
+	// Phase 2 — update-race (If-Match at the SAME version): exactly one may advance it.
+	winners, err = raceCAS(ctx, store, key, version)
+	if err != nil {
+		return fmt.Errorf("org: CAS probe update-race: %w", err)
+	}
+	if winners != 1 {
+		return fmt.Errorf("%w: update-race admitted %d writers (want 1) — If-Match not atomic", ErrNonAtomicStore, winners)
+	}
+	return nil
+}
+
+// raceCAS fires probeRacers concurrent PutIfVersion(key, distinct, expectVersion) and
+// counts how many the store admitted. A loser that failed the precondition
+// (replica.ErrConflict) is the correct, expected outcome; any OTHER error is a hard
+// store failure surfaced so the caller fails safe (an unreachable store yields zero
+// winners AND a hard error, so it never reads as "atomic").
+func raceCAS(ctx context.Context, store replica.ConditionalStore, key, expectVersion string) (winners int, hardErr error) {
+	var (
+		wg sync.WaitGroup
+		mu sync.Mutex
+	)
+	wg.Add(probeRacers)
+	for i := 0; i < probeRacers; i++ {
+		go func(i int) {
+			defer wg.Done()
+			_, err := store.PutIfVersion(ctx, key, probePayload(i), expectVersion)
+			mu.Lock()
+			defer mu.Unlock()
+			switch {
+			case err == nil:
+				winners++
+			case errors.Is(err, replica.ErrConflict):
+				// Expected loser: the precondition held and rejected this racer.
+			case hardErr == nil:
+				hardErr = err
+			}
+		}(i)
+	}
+	wg.Wait()
+	return winners, hardErr
+}
+
+// probeKey mints a unique-per-boot throwaway object key under keyPrefix so the
+// create-race always tests a genuine creation (a fixed key would exist on the second
+// boot and make the create-race admit zero winners — a false non-atomic verdict).
+func probeKey(prefix string) (string, error) {
+	var b [16]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "", fmt.Errorf("org: CAS probe key: %w", err)
+	}
+	return prefix + hex.EncodeToString(b[:]), nil
+}
+
+// probePayload is a small distinct value per racer so a store that (wrongly) accepts
+// two writes leaves observably different bytes, not an idempotent no-op.
+func probePayload(i int) []byte { return []byte{'c', 'a', 's', '-', 'p', 'r', 'o', 'b', 'e', byte('0' + i)} }

internal/org/condprobe_test.go

@@ -0,0 +1,79 @@
+// Copyright 2025 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package org
+
+// condprobe_test.go proves the boot atomicity self-check: an atomic store is
+// confirmed, a last-write-wins (non-atomic) store is REFUSED (ErrNonAtomicStore, the
+// caller then runs local-only), and an unreachable store never reads as atomic. This
+// is H2 as a self-validating boot probe rather than a human-set flag.
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/hanzoai/vfs/replica"
+)
+
+// TestProbeCASAtomicStoreConfirmed: over the SAME atomic fakeCondStore the fence trusts
+// (single mutex making PutIfVersion an indivisible compare-and-set), the probe confirms
+// atomicity — exactly one winner in both the create-race and the update-race.
+func TestProbeCASAtomicStoreConfirmed(t *testing.T) {
+	if err := ProbeCAS(context.Background(), newFakeCondStore(), ".probe/cas-"); err != nil {
+		t.Fatalf("atomic store must confirm: %v", err)
+	}
+}
+
+// TestProbeCASNonAtomicRefused: a store whose conditional PUT ignores the precondition
+// (both racers "win") is the split-brain hazard the probe exists to catch — it must
+// return ErrNonAtomicStore so buildDurability fails safe to local-only.
+func TestProbeCASNonAtomicRefused(t *testing.T) {
+	err := ProbeCAS(context.Background(), &nonAtomicStore{objects: map[string][]byte{}}, ".probe/cas-")
+	if !errors.Is(err, ErrNonAtomicStore) {
+		t.Fatalf("non-atomic store must be refused: err=%v, want ErrNonAtomicStore", err)
+	}
+}
+
+// TestProbeCASUnreachableNotAtomic: an unreachable store surfaces a hard error and is
+// NEVER reported atomic (zero winners must not be mistaken for a pass).
+func TestProbeCASUnreachableNotAtomic(t *testing.T) {
+	cs := newFakeCondStore()
+	cs.failAll = errors.New("object store unreachable")
+	if err := ProbeCAS(context.Background(), cs, ".probe/cas-"); err == nil {
+		t.Fatal("unreachable store must not confirm atomicity")
+	}
+}
+
+// nonAtomicStore is a last-write-wins object store: PutIfVersion ignores expectVersion
+// and always succeeds, modelling an S3 gateway with no real conditional-PUT support —
+// the exact store the probe must refuse to fence on.
+type nonAtomicStore struct {
+	mu      sync.Mutex
+	objects map[string][]byte
+	ver     int
+}
+
+func (s *nonAtomicStore) Get(_ context.Context, key string) ([]byte, string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	data, ok := s.objects[key]
+	if !ok {
+		return nil, "", replica.ErrNotFound
+	}
+	return append([]byte(nil), data...), strconv.Itoa(s.ver), nil
+}
+
+func (s *nonAtomicStore) PutIfVersion(_ context.Context, key string, data []byte, _ string) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.ver++
+	s.objects[key] = append([]byte(nil), data...)
+	return strconv.Itoa(s.ver), nil // ignores the precondition — always "wins".
+}
+
+var _ replica.ConditionalStore = (*nonAtomicStore)(nil)

internal/org/durable.go

@@ -45,27 +45,14 @@ package org
 import (
 	"context"
 	"database/sql"
-	"encoding/binary"
 	"errors"
 	"fmt"
-	"os"
 	"sync"
 
 	"github.com/hanzoai/ha"
 	"github.com/hanzoai/vfs/replica"
 )
 
-// dekSuffix is cek's key-sidecar suffix (github.com/hanzoai/cloud/cek): the file
-// holding a database's wrapped page key. A durable snapshot ships it beside the
-// database bytes so a successor can open the encrypted file. Kept as a local
-// constant (a stable on-disk convention) so this package stays dep-free of cek.
-const dekSuffix = ".dek"
-
-// errCorruptFrame is returned when a durable payload does not decode as a
-// (sidecar, database) frame this package wrote — fail closed rather than restore
-// arbitrary bytes over a live database.
-var errCorruptFrame = errors.New("org: durable payload is not a valid (sidecar,db) frame")
-
 // Durability is the per-deployment, org-agnostic durable-store factory: the shared
 // election+fence over ONE object store, plus the optional at-rest envelope. It holds
 // no per-org state — For() mints a Durable per org DB. nil ⇒ durability disabled
@@ -74,6 +61,7 @@ type Durability struct {
 	fencer *CASFencer
 	fenced *replica.FencedStore
 	cipher *Cipher
+	codec  snapshotCodec // how the durable payload is produced/applied (swappable ship)
 }
 
 // NewDurability builds the factory over an atomic-CAS object store (the SeaweedFS S3
@@ -81,14 +69,44 @@ type Durability struct {
 // optional per-org envelope Cipher (nil ⇒ the durable object is stored in the clear;
 // pure-Go dev only). The SAME cond backs both the lease and the data ships, so they
 // share one linearizable register.
-func NewDurability(cond replica.ConditionalStore, view ownerView, cipher *Cipher) *Durability {
+//
+// The cond is an INTERFACE by design (replica.ConditionalStore): a read-through/
+// write-through cache tier (KV in front of S3) wraps it with a one-line decorator at the
+// buildDurability construction site, with no change here — the fence reads and CASes
+// through whatever store it is handed. The ship mechanism is likewise swappable: the
+// default wholeFile codec can be replaced by a WAL-frame delta codec behind snapshotCodec
+// without touching the fence or round. WithCheckpoint injects the envelope's re-encrypting
+// Checkpoint (crypto-integration seam).
+func NewDurability(cond replica.ConditionalStore, view ownerView, cipher *Cipher, opts ...DurabilityOption) *Durability {
+	var o durabilityOpts
+	for _, fn := range opts {
+		fn(&o)
+	}
 	return &Durability{
 		fencer: NewCASFencer(cond, view),
 		fenced: replica.NewFencedStore(cond),
 		cipher: cipher,
+		codec:  wholeFile{checkpoint: o.checkpoint},
 	}
 }
 
+// DurabilityOption configures a Durability.
+type DurabilityOption func(*durabilityOpts)
+
+type durabilityOpts struct {
+	checkpoint func(context.Context, *sql.DB) error
+}
+
+// WithCheckpoint injects the operation that folds the WAL into the real on-disk file and,
+// on the pure-Go encryption ENVELOPE, re-encrypts that real path — so a fenced ship reads
+// FRESH bytes, never stale ciphertext (which would be a lost acked write on takeover). The
+// composition root wires cek's Checkpoint here on the envelope backend; the default (no
+// option) is a raw TRUNCATE checkpoint, correct for the SQLCipher page-level and plaintext
+// backends that encrypt on write. This composes ship-before-ack with encrypt-on-checkpoint.
+func WithCheckpoint(fn func(context.Context, *sql.DB) error) DurabilityOption {
+	return func(o *durabilityOpts) { o.checkpoint = fn }
+}
+
 // For mints the Durable binding for one org DB: orgID is the org SLUG (the HRW
 // election key AND the cipher AAD — the caller passes the SAME slug the on-disk path
 // and the shard router hash use). dbKey is the durable object location
@@ -122,14 +140,12 @@ type Durable struct {
 // unopenable, so a store is always available for reads; writes fail closed until a
 // later open re-acquires.
 //
-// RECOVERY from a degraded open (Red M3): a pod that could not acquire at open stays
-// read-only for that store's cached lifetime — it does NOT re-acquire in place, because
-// a takeover CarryForward restores the durable snapshot OVER the local file, which is
-// unsafe under the live handle the store already handed out (stale reads). Recovery is
-// therefore a FRESH open: a pod restart (a readiness/liveness probe can gate on the
-// degraded log) or the shard router routing the org to a healthy owner. An in-process
-// quiesce-close-reopen on the cached entry is the future enhancement; until then the
-// safe, simple recovery is re-open.
+// RECOVERY from a degraded open (M3): a pod that could not acquire at open stays
+// read-only until this replica becomes the org's elected owner (a membership change),
+// at which point the OrgStore promotes the store IN PLACE — PendingPromotion gates it,
+// TryClaim proves the lease is claimable, then a quiesce-close-reopen swaps in a writer
+// handle and CarryForward-restores the latest snapshot under the FRESH handle (never
+// under the live one — the swap is why the reopen is required). No process restart.
 func (d *Durable) Hydrate(ctx context.Context) error {
 	lease, err := d.dy.fencer.Acquire(ctx, d.orgID)
 	if err != nil {
@@ -170,6 +186,42 @@ func (d *Durable) Owned() bool {
 	return d.owned
 }
 
+// PendingPromotion reports whether this store opened degraded (does not hold the lease)
+// yet this replica is NOW the org's elected owner — a membership change made it the
+// writer, so the store must be promoted (re-acquire + hydrate + reopen) to serve writes.
+// Cheap and I/O-free: a lock plus one HRW over the live member snapshot, evaluated only
+// when the store is not already owned. It is the per-request gate the OrgStore checks on
+// a cache hit; the actual promotion runs (rarely) only when this returns true.
+func (d *Durable) PendingPromotion() bool {
+	d.mu.Lock()
+	owned := d.owned
+	d.mu.Unlock()
+	if owned {
+		return false
+	}
+	return d.dy.fencer.ElectsSelf(d.orgID)
+}
+
+// TryClaim probes whether this replica can hold the org's writer lease right now and, if
+// so, claims it — the promotion gate. It performs ONLY the lease CAS (via the fencer), no
+// local-file I/O, so a failed probe (not the elected owner, or the store unreachable)
+// costs nothing and leaves any live handle untouched. On success the lease object names
+// this replica at a fresh round (fencing any prior owner); the caller then quiesces and
+// reopens, whose Hydrate renews THIS same lease and CarryForward-restores the latest
+// snapshot under the fresh handle. Returns (false, nil) when not the elected owner,
+// (false, err) on a store error, (true, nil) when claimed.
+func (d *Durable) TryClaim(ctx context.Context) (bool, error) {
+	_, err := d.dy.fencer.Acquire(ctx, d.orgID)
+	switch {
+	case err == nil:
+		return true, nil
+	case errors.Is(err, ErrNotOwner), errors.Is(err, ErrNoMembership):
+		return false, nil
+	default:
+		return false, err
+	}
+}
+
 // Sync snapshots the local file and ships it to the durable object fenced at the
 // lease round — the ship-before-ack step. acked is true only if the fenced store
 // admitted our round. A ship rejected as ErrStaleRound returns (false, nil): this
@@ -222,11 +274,9 @@ func (d *Durable) Close(ctx context.Context) error {
 	return nil
 }
 
-// snapshot copies the local file consistently: it takes the store's SOLE connection
-// (MaxOpenConns(1)), TRUNCATE-checkpoints the WAL so the main file holds every
-// committed page, then reads the file bytes while still holding the connection so no
-// writer can fold new frames in mid-read. The <db>.dek key sidecar (present only on
-// an encrypting build) is read too and framed with the database bytes.
+// snapshot produces the durable payload for the bound local database via the swappable
+// codec (default wholeFile: checkpoint + framed file copy). It takes the store's SOLE
+// connection through the codec so the payload is consistent against concurrent writes.
 func (d *Durable) snapshot(ctx context.Context) ([]byte, error) {
 	d.mu.Lock()
 	db := d.db
@@ -234,38 +284,13 @@ func (d *Durable) snapshot(ctx context.Context) ([]byte, error) {
 	if db == nil {
 		return nil, fmt.Errorf("org: durable %s not bound to a db", d.dbKey)
 	}
-	conn, err := db.Conn(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("org: durable snapshot conn %s: %w", d.dbKey, err)
-	}
-	defer conn.Close()
-	// TRUNCATE-checkpoint the WAL so the main file holds every committed page, and
-	// CHECK the result: busy!=0 means the checkpoint could not fold all frames (a
-	// reader held the WAL), so the file on disk is MISSING committed rows. Fail closed
-	// — a partial snapshot shipped as complete is a silent lost write. (busy, logFrames,
-	// checkpointed) is the PRAGMA's row.
-	var busy, logFrames, checkpointed int
-	if err := conn.QueryRowContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &logFrames, &checkpointed); err != nil {
-		return nil, fmt.Errorf("org: durable checkpoint %s: %w", d.dbKey, err)
-	}
-	if busy != 0 {
-		return nil, fmt.Errorf("org: durable checkpoint %s did not complete (busy=%d, log=%d, checkpointed=%d) — snapshot would miss committed WAL frames", d.dbKey, busy, logFrames, checkpointed)
-	}
-	main, err := os.ReadFile(d.dbPath)
-	if err != nil {
-		return nil, fmt.Errorf("org: durable read %s: %w", d.dbPath, err)
-	}
-	sidecar, err := os.ReadFile(d.dbPath + dekSuffix)
-	if err != nil && !os.IsNotExist(err) {
-		return nil, fmt.Errorf("org: durable read sidecar %s: %w", d.dbPath, err)
-	}
-	return frame(sidecar, main), nil
+	return d.dy.codec.produce(ctx, db, d.dbPath)
 }
 
-// restore is the hydrate callback: it opens the sealed durable payload, splits the
-// (sidecar, database) frame, writes the sidecar (so cek can open the encrypted file)
-// and atomically swaps the database bytes into place via replica.RestoreFile. An
-// empty payload (nothing shipped yet) keeps whatever the local file already holds.
+// restore is the hydrate callback: it opens the sealed durable payload (envelope
+// decryption is orthogonal, done HERE around the codec) and hands the plaintext to the
+// codec to apply onto the local file. An empty payload (nothing shipped yet) keeps
+// whatever the local file already holds.
 func (d *Durable) restore(sealed []byte) error {
 	if len(sealed) == 0 {
 		return nil
@@ -278,24 +303,7 @@ func (d *Durable) restore(sealed []byte) error {
 		}
 		payload = pt
 	}
-	sidecar, main, err := unframe(payload)
-	if err != nil {
-		return err
-	}
-	// RestoreFile FIRST: it MkdirAll's the parent and atomically swaps the database
-	// bytes into place. The sidecar is written AFTER, into the now-existing directory,
-	// so both the encrypted file and its key are present before cek opens them. (Writing
-	// the sidecar first would fail on a fresh successor whose orgs/<slug>/ dir does not
-	// exist yet.)
-	if err := replica.RestoreFile(d.dbPath, main); err != nil {
-		return err
-	}
-	if len(sidecar) > 0 {
-		if err := os.WriteFile(d.dbPath+dekSuffix, sidecar, 0o600); err != nil {
-			return fmt.Errorf("org: durable write sidecar %s: %w", d.dbPath, err)
-		}
-	}
-	return nil
+	return d.dy.codec.apply(d.dbPath, payload)
 }
 
 // restoreLatestReadOnly refreshes the local file from the durable object without a
@@ -308,26 +316,3 @@ func (d *Durable) restoreLatestReadOnly(ctx context.Context) {
 	}
 	_ = d.restore(payload)
 }
-
-// frame prepends the sidecar under a uvarint length so a successor can split it from
-// the database bytes. len(sidecar)==0 ⇒ a leading 0 byte, i.e. a plaintext store.
-func frame(sidecar, main []byte) []byte {
-	buf := make([]byte, 0, binary.MaxVarintLen64+len(sidecar)+len(main))
-	var n [binary.MaxVarintLen64]byte
-	m := binary.PutUvarint(n[:], uint64(len(sidecar)))
-	buf = append(buf, n[:m]...)
-	buf = append(buf, sidecar...)
-	return append(buf, main...)
-}
-
-// unframe splits a framed payload back into (sidecar, database). The bound is checked
-// as sl > len(b)-m (a SUBTRACTION, never m+sl) so a maliciously large uvarint length
-// cannot wrap uint64 addition past the guard and panic the slice — it fails closed.
-func unframe(b []byte) (sidecar, main []byte, err error) {
-	sl, m := binary.Uvarint(b)
-	if m <= 0 || sl > uint64(len(b)-m) {
-		return nil, nil, errCorruptFrame
-	}
-	off := m + int(sl)
-	return b[m:off], b[off:], nil
-}

internal/org/fence.go

@@ -102,6 +102,16 @@ func (f *CASFencer) Acquire(ctx context.Context, orgID string) (ha.Lease, error)
 	return f.claim(ctx, orgID, self)
 }
 
+// ElectsSelf reports whether this replica is the HRW-elected writer for orgID under the
+// CURRENT live membership — the election half of Acquire WITHOUT the lease CAS, so it does
+// no I/O. It is the cheap gate a degraded (read-only) store consults to decide whether a
+// membership change has made it the owner and it should attempt promotion. Fail-closed on
+// an empty set (no safe owner ⇒ not self).
+func (f *CASFencer) ElectsSelf(orgID string) bool {
+	members := f.view.Members()
+	return len(members) > 0 && IsOwner(orgID, f.view.Self(), members)
+}
+
 // claim reads the current lease and either renews it (owner == self: keep round)
 // or takes it over (strictly bump the round to recorded+1, via a version-
 // conditioned CAS so two racing claimers cannot both win the same round).

internal/org/promote_test.go

@@ -0,0 +1,212 @@
+// Copyright 2025 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package org
+
+// promote_test.go is the M3 proof: a store that opened DEGRADED (read-only, not the
+// elected owner) is promoted to the writer IN PLACE when a membership change makes this
+// replica the org's owner — re-acquiring the lease, hydrating the predecessor's shipped
+// state, and serving writes, with NO process restart and NO split-brain (the promoted
+// store takes over at a strictly higher round, so the deposed owner is fenced).
+
+import (
+	"context"
+	"database/sql"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/hanzoai/vfs/replica"
+)
+
+// liveView is a MUTABLE membership view — the seam a rolling-upgrade change flows
+// through. swap() replaces the live set so PendingPromotion re-evaluates HRW against the
+// new pod set, exactly as internal/org.Membership's atomic snapshot does in production.
+type liveView struct {
+	self string
+	mu   sync.Mutex
+	set  []Member
+}
+
+func (v *liveView) Self() string { return v.self }
+
+func (v *liveView) Members() []Member {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	return append([]Member(nil), v.set...)
+}
+
+func (v *liveView) swap(ms []Member) {
+	v.mu.Lock()
+	v.set = ms
+	v.mu.Unlock()
+}
+
+// openBoundDB opens the local handle for a Durable (as the real store does: one
+// connection, bound so Sync checkpoints on it) and ensures the kv table.
+func openBoundDB(t *testing.T, d *Durable) *sql.DB {
+	t.Helper()
+	// Create the parent dir before opening (what production's openOrgDB does): a fresh org
+	// whose durable object is empty has nothing to restore, so the orgs/<slug>/ dir does
+	// not exist yet.
+	if err := os.MkdirAll(filepath.Dir(d.dbPath), 0o700); err != nil {
+		t.Fatalf("mkdir %s: %v", d.dbPath, err)
+	}
+	db, err := sql.Open("sqlite", testDSN(d.dbPath))
+	if err != nil {
+		t.Fatalf("open %s: %v", d.dbPath, err)
+	}
+	db.SetMaxOpenConns(1)
+	if err := db.Ping(); err != nil {
+		t.Fatalf("ping %s: %v", d.dbPath, err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	d.Bind(db)
+	ensureKV(t, db)
+	return db
+}
+
+// TestDurablePromotionLiveReacquireNoRestart drives the full M3 transition through the
+// Durable primitives the OrgStore orchestrates (PendingPromotion → TryClaim →
+// quiesce-close → reopen+Hydrate): a non-owner that opened read-only becomes the elected
+// owner after a membership change and takes over live — hydrating the predecessor's k1,
+// serving k2, at a higher round — while the deposed owner is fenced.
+func TestDurablePromotionLiveReacquireNoRestart(t *testing.T) {
+	ctx := context.Background()
+	cs := newFakeCondStore()
+	const orgID = "acme"
+	dbKey := replica.DBPath(orgID, "", "research")
+	set := []Member{{ID: "pod-a"}, {ID: "pod-b"}}
+	ownerID := electedOwnerID(orgID, set)
+	otherID := "pod-a"
+	if otherID == ownerID {
+		otherID = "pod-b"
+	}
+
+	// Owner pod: elected owner under {a,b}. Writes k1, ships.
+	ownerDur := NewDurability(cs, &liveView{self: ownerID, set: set}, nil)
+	owner := ownerDur.For(orgID, dbKey, filepath.Join(t.TempDir(), "research.db"))
+	if err := owner.Hydrate(ctx); err != nil {
+		t.Fatalf("owner hydrate: %v", err)
+	}
+	ownerDB := openBoundDB(t, owner)
+	putKV(t, ownerDB, "k1", "v1")
+	if acked, err := owner.Sync(ctx); err != nil || !acked {
+		t.Fatalf("owner sync k1: acked=%v err=%v", acked, err)
+	}
+
+	// Other pod: opens DEGRADED under the SAME live set (not the elected owner). It
+	// read-only-hydrates k1 but does not own — writes fail closed.
+	otherView := &liveView{self: otherID, set: set}
+	otherDur := NewDurability(cs, otherView, nil)
+	otherPath := filepath.Join(t.TempDir(), "research.db")
+	other := otherDur.For(orgID, dbKey, otherPath)
+	if err := other.Hydrate(ctx); err != nil {
+		t.Fatalf("other hydrate: %v", err)
+	}
+	if other.Owned() {
+		t.Fatal("non-owner must open degraded (read-only), not owned")
+	}
+	if other.PendingPromotion() {
+		t.Fatal("must NOT be pending promotion while another pod is the elected owner")
+	}
+	otherDB := openBoundDB(t, other)
+	if !hasKV(t, otherDB, "k1") {
+		t.Fatal("degraded open should have read-only hydrated the owner's k1")
+	}
+
+	// Rolling upgrade: the owner drains out of the live set. The other pod is now the
+	// HRW owner — so its degraded store becomes pending-promotion.
+	otherView.swap([]Member{{ID: otherID}})
+	if !other.PendingPromotion() {
+		t.Fatal("must be pending promotion once this replica is the elected owner")
+	}
+
+	// M3 promotion, no restart: claim the lease (probe — no file I/O), quiesce the
+	// read-only handle, then reopen as owner (fresh Durable, SAME local path — the
+	// in-place swap the OrgStore performs). Hydrate renews the claimed lease and
+	// CarryForward-restores the latest snapshot under the fresh handle.
+	claimed, err := other.TryClaim(ctx)
+	if err != nil || !claimed {
+		t.Fatalf("TryClaim after election: claimed=%v err=%v, want true/nil", claimed, err)
+	}
+	_ = otherDB.Close() // quiesce before the reopen's CarryForward swaps the file.
+
+	promoted := otherDur.For(orgID, dbKey, otherPath)
+	if err := promoted.Hydrate(ctx); err != nil {
+		t.Fatalf("promoted hydrate: %v", err)
+	}
+	if !promoted.Owned() {
+		t.Fatal("promoted store must hold the writer lease")
+	}
+	if promoted.lease.Round <= owner.lease.Round {
+		t.Fatalf("promoted round %d not strictly > deposed owner %d — fencing token not advanced", promoted.lease.Round, owner.lease.Round)
+	}
+	promotedDB := openBoundDB(t, promoted)
+	if !hasKV(t, promotedDB, "k1") {
+		t.Fatal("promoted store lost the predecessor's k1 (hydrate-on-promote failed)")
+	}
+	putKV(t, promotedDB, "k2", "v2")
+	if acked, err := promoted.Sync(ctx); err != nil || !acked {
+		t.Fatalf("promoted sync k2: acked=%v err=%v", acked, err)
+	}
+
+	// The deposed owner that keeps running ships at its STALE round: fenced, never acked.
+	putKV(t, ownerDB, "k-stale", "should-be-fenced")
+	acked, err := owner.Sync(ctx)
+	if err != nil {
+		t.Fatalf("deposed owner sync unexpected hard error: %v", err)
+	}
+	if acked {
+		t.Fatal("deposed owner ACKED a fenced ship after promotion — split-brain double-write")
+	}
+
+	// Durable state: k1 + k2 present, the deposed writer's k-stale never landed.
+	if v, ok := durableValue(t, ctx, cs, dbKey, "k1"); !ok || v != "v1" {
+		t.Fatalf("durable k1 = %q,%v, want v1,true", v, ok)
+	}
+	if v, ok := durableValue(t, ctx, cs, dbKey, "k2"); !ok || v != "v2" {
+		t.Fatalf("durable k2 = %q,%v, want v2,true", v, ok)
+	}
+	if v, ok := durableValue(t, ctx, cs, dbKey, "k-stale"); ok {
+		t.Fatalf("deposed writer's k-stale leaked into durable state as %q", v)
+	}
+}
+
+// TestPendingPromotionFalseWhenNotElected: a degraded store whose replica is NOT the
+// elected owner must never report pending-promotion (it stays read-only), and an OWNED
+// store never does either (nothing to promote).
+func TestPendingPromotionFalseWhenNotElected(t *testing.T) {
+	ctx := context.Background()
+	cs := newFakeCondStore()
+	const orgID = "acme"
+	dbKey := replica.DBPath(orgID, "", "research")
+	set := []Member{{ID: "pod-a"}, {ID: "pod-b"}}
+	ownerID := electedOwnerID(orgID, set)
+	otherID := "pod-a"
+	if otherID == ownerID {
+		otherID = "pod-b"
+	}
+
+	// Owned store: not pending (already the writer).
+	ownerDur := NewDurability(cs, &liveView{self: ownerID, set: set}, nil)
+	owner := ownerDur.For(orgID, dbKey, filepath.Join(t.TempDir(), "research.db"))
+	if err := owner.Hydrate(ctx); err != nil {
+		t.Fatalf("owner hydrate: %v", err)
+	}
+	if owner.PendingPromotion() {
+		t.Fatal("an owned store is never pending promotion")
+	}
+
+	// Degraded store, still not elected: not pending.
+	other := NewDurability(cs, &liveView{self: otherID, set: set}, nil).For(orgID, dbKey, filepath.Join(t.TempDir(), "research.db"))
+	if err := other.Hydrate(ctx); err != nil {
+		t.Fatalf("other hydrate: %v", err)
+	}
+	if other.PendingPromotion() {
+		t.Fatal("a degraded store whose replica is not the elected owner must not be pending promotion")
+	}
+}

internal/org/rollingupgrade_test.go

@@ -0,0 +1,301 @@
+// Copyright 2025 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package org
+
+// rollingupgrade_test.go is the zero-downtime PROOF: N pods each own a shard of orgs, and
+// we ROLL them — drain a pod, let ownership transition to a live successor, rejoin it —
+// while a client writes continuously to every org (always routing to the org's current
+// elected owner, as the shard router does). It asserts the three properties a true
+// zero-downtime rolling upgrade must hold:
+//
+//	(a) ZERO lost acked writes    — every write the store acknowledged is readable from the
+//	                                durable object after all transitions (the successor
+//	                                hydrated it).
+//	(b) ZERO split-brain          — no (org, round) is ever acknowledged by two different
+//	                                pods; a deposed owner's ship is fenced, never acked.
+//	(c) Continuous availability   — every org is served by SOME ready owner at every step,
+//	                                within a bounded re-route (the write always lands).
+//
+// It exercises BOTH takeover paths: a pod RESTART (fresh rejoin → hydrate from the durable
+// object) and an in-place ownership FLAP with no restart (scale a pod in then out → the
+// deposed store's next write is fenced, and the retry PROMOTES it live via M3). Sequential
+// and deterministic (one client goroutine, membership changed at controlled points), the
+// same fake-CAS + membership harness durable_test.go uses.
+
+import (
+	"context"
+	"database/sql"
+	"path/filepath"
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/hanzoai/vfs/replica"
+)
+
+// sharedSet is the converged live membership every pod reads — the atomic snapshot the
+// production K8s source maintains, here mutated at controlled points to model a roll.
+type sharedSet struct {
+	mu  sync.Mutex
+	set []Member
+}
+
+func (s *sharedSet) get() []Member {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]Member(nil), s.set...)
+}
+
+func (s *sharedSet) put(ids ...string) {
+	ms := make([]Member, len(ids))
+	for i, id := range ids {
+		ms[i] = Member{ID: id}
+	}
+	s.mu.Lock()
+	s.set = ms
+	s.mu.Unlock()
+}
+
+// podView is one pod's membership view: its own self over the shared converged set.
+type podView struct {
+	self   string
+	shared *sharedSet
+}
+
+func (v podView) Self() string      { return v.self }
+func (v podView) Members() []Member { return v.shared.get() }
+
+type rollOrg struct {
+	d  *Durable
+	db *sql.DB // the bound handle, closed on quiesce/reopen
+}
+
+// rollPod is one replica: its own local data dir, a Durability over the shared object
+// store + its own view, and the per-org stores it currently holds open.
+type rollPod struct {
+	id     string
+	dir    string
+	dy     *Durability
+	stores map[string]*rollOrg
+}
+
+type rollHarness struct {
+	t          *testing.T
+	ctx        context.Context
+	cs         replica.ConditionalStore
+	shared     *sharedSet
+	orgs       []string
+	pods       map[string]*rollPod
+	ackedSeq   map[string]int    // org → highest seq the store acknowledged
+	roundOwner map[string]string // "org:round" → the ONE pod that acked at that round
+}
+
+const rerouteAttempts = 6 // bounded re-route: a transition resolves within this many tries.
+
+func newRollHarness(t *testing.T, orgs []string) *rollHarness {
+	return &rollHarness{
+		t: t, ctx: context.Background(), cs: newFakeCondStore(), shared: &sharedSet{},
+		orgs: orgs, pods: map[string]*rollPod{},
+		ackedSeq: map[string]int{}, roundOwner: map[string]string{},
+	}
+}
+
+// startPod (re)creates a pod process: a FRESH local dir + empty store set — modelling a
+// restart, where the process re-hydrates each org from the durable object on open.
+func (h *rollHarness) startPod(id string) {
+	h.pods[id] = &rollPod{
+		id: id, dir: h.t.TempDir(),
+		dy:     NewDurability(h.cs, podView{self: id, shared: h.shared}, nil),
+		stores: map[string]*rollOrg{},
+	}
+}
+
+func (h *rollHarness) stopPod(id string) { delete(h.pods, id) }
+
+func orgDBKey(org string) string { return replica.DBPath(org, "", "research") }
+func (p *rollPod) orgPath(org string) string {
+	return filepath.Join(p.dir, "orgs", org, "research.db")
+}
+
+// ensureOwned opens the org's store on first use, or PROMOTES it in place (M3) when it
+// opened degraded and this pod is now the elected owner — TryClaim, quiesce the read-only
+// handle, reopen as owner (Hydrate renews + CarryForward-restores). Exactly OrgStore.forPath
+// + promote, driven through the same Durable primitives.
+func (h *rollHarness) ensureOwned(p *rollPod, org string) *rollOrg {
+	oo, ok := p.stores[org]
+	if !ok {
+		d := p.dy.For(org, orgDBKey(org), p.orgPath(org))
+		_ = d.Hydrate(h.ctx)
+		oo = &rollOrg{d: d, db: openBoundDB(h.t, d)}
+		p.stores[org] = oo
+		return oo
+	}
+	if !oo.d.Owned() && oo.d.PendingPromotion() {
+		if claimed, _ := oo.d.TryClaim(h.ctx); claimed {
+			_ = oo.db.Close() // quiesce before CarryForward swaps the file
+			d := p.dy.For(org, orgDBKey(org), p.orgPath(org))
+			_ = d.Hydrate(h.ctx)
+			oo = &rollOrg{d: d, db: openBoundDB(h.t, d)}
+			p.stores[org] = oo
+		}
+	}
+	return oo
+}
+
+// write routes seq to the org's CURRENT elected owner and records the acknowledgement,
+// retrying across a bounded re-route while ownership transitions. It FAILS the test if no
+// owner can serve within the budget (availability) or if two pods ack one round (split
+// brain).
+func (h *rollHarness) write(org string, seq int) {
+	for attempt := 0; attempt < rerouteAttempts; attempt++ {
+		owner, ok := Owner(org, h.shared.get())
+		if !ok {
+			h.t.Fatalf("org %s: empty membership — no owner (availability)", org)
+		}
+		p := h.pods[owner.ID]
+		if p == nil {
+			h.t.Fatalf("org %s: elected owner %s is not a live pod", org, owner.ID)
+		}
+		oo := h.ensureOwned(p, org)
+		if !oo.d.Owned() {
+			continue // just opened degraded / promotion pending — retry drives it owned.
+		}
+		putKV(h.t, oo.db, "seq", strconv.Itoa(seq))
+		acked, err := oo.d.Sync(h.ctx)
+		if err != nil {
+			h.t.Fatalf("org %s seq %d on %s: hard sync error: %v", org, seq, owner.ID, err)
+		}
+		if !acked {
+			continue // deposed mid-write (fenced): retry re-routes / promotes.
+		}
+		h.recordAck(org, seq, owner.ID, uint64(oo.d.lease.Round))
+		return
+	}
+	h.t.Fatalf("AVAILABILITY VIOLATED: org %s seq %d not served within %d attempts", org, seq, rerouteAttempts)
+}
+
+// recordAck registers a durable acknowledgement and checks the split-brain invariant: a
+// given (org, round) belongs to exactly ONE pod, because a handoff strictly bumps the
+// round and the fence rejects a deposed writer — so two distinct pods acking one round
+// would mean two live writers.
+func (h *rollHarness) recordAck(org string, seq int, podID string, round uint64) {
+	key := org + ":" + strconv.FormatUint(round, 10)
+	if prev, ok := h.roundOwner[key]; ok && prev != podID {
+		h.t.Fatalf("SPLIT-BRAIN: org %s round %d acked by BOTH %s and %s", org, round, prev, podID)
+	}
+	h.roundOwner[key] = podID
+	if seq > h.ackedSeq[org] {
+		h.ackedSeq[org] = seq
+	}
+}
+
+// writeAll writes the next seq to every org (a full sweep of continuous client traffic).
+func (h *rollHarness) writeAll(seq *int) {
+	for _, org := range h.orgs {
+		*seq++
+		h.write(org, *seq)
+	}
+}
+
+// verifyNoLostWrites opens each org's durable object fresh (the exact path a brand-new
+// successor takes) and asserts it carries at least the highest seq the store ever acked —
+// no acknowledged write was lost across any transition.
+func (h *rollHarness) verifyNoLostWrites() {
+	for _, org := range h.orgs {
+		want := h.ackedSeq[org]
+		if want == 0 {
+			continue
+		}
+		v, ok := durableValue(h.t, h.ctx, h.cs, orgDBKey(org), "seq")
+		if !ok {
+			h.t.Fatalf("org %s: durable object has no state, want acked seq %d (LOST)", org, want)
+		}
+		got, _ := strconv.Atoi(v)
+		if got < want {
+			h.t.Fatalf("org %s: durable seq %d < highest acked %d — LOST ACKED WRITE", org, got, want)
+		}
+	}
+}
+
+// TestRollingUpgradeZeroDowntime is the end-to-end proof over 4 orgs and 3 pods.
+func TestRollingUpgradeZeroDowntime(t *testing.T) {
+	orgs := []string{"acme", "globex", "initech", "umbrella", "hooli", "piedpiper", "wonka", "stark"}
+	h := newRollHarness(t, orgs)
+	for _, id := range []string{"cloud-0", "cloud-1", "cloud-2"} {
+		h.startPod(id)
+	}
+	h.shared.put("cloud-0", "cloud-1", "cloud-2")
+	seq := 0
+
+	// Warm: every org served by its initial owner.
+	h.writeAll(&seq)
+
+	// Phase A — rolling RESTART of each pod: drain it (leaves the set → its orgs re-home to
+	// live successors that hydrate), write continuously, then rejoin it FRESH (a restart:
+	// new local dir → re-hydrate from the durable object). Zero request lost throughout.
+	for _, rolling := range []string{"cloud-0", "cloud-1", "cloud-2"} {
+		// Drain.
+		remaining := without([]string{"cloud-0", "cloud-1", "cloud-2"}, rolling)
+		h.shared.put(remaining...)
+		h.stopPod(rolling)
+		h.writeAll(&seq) // successors take over; every org still served.
+		h.writeAll(&seq)
+
+		// Rejoin (restart): fresh process, rehydrates the orgs HRW hands back to it.
+		h.startPod(rolling)
+		h.shared.put("cloud-0", "cloud-1", "cloud-2")
+		h.writeAll(&seq)
+		h.writeAll(&seq)
+	}
+
+	// Phase B — in-place ownership FLAP, NO restart (exercises M3): scale a 4th pod IN
+	// (steals + fences some orgs from the running 0/1/2), write, then scale it OUT. The
+	// orgs return to their original still-running owners, whose cached stores are now at a
+	// stale round — the next write is fenced and the retry PROMOTES them live (no restart).
+	h.startPod("cloud-3")
+	h.shared.put("cloud-0", "cloud-1", "cloud-2", "cloud-3")
+	// The flap only exercises in-place promotion if cloud-3 actually steals (and fences) an
+	// org from a still-running 0/1/2 — assert it does, so the test never silently degenerates.
+	stolen := 0
+	for _, org := range orgs {
+		if o, _ := Owner(org, h.shared.get()); o.ID == "cloud-3" {
+			stolen++
+		}
+	}
+	if stolen == 0 {
+		t.Fatal("phase B degenerate: cloud-3 stole no org (widen the org set)")
+	}
+	h.writeAll(&seq)
+	h.writeAll(&seq)
+	h.stopPod("cloud-3")
+	h.shared.put("cloud-0", "cloud-1", "cloud-2")
+	h.writeAll(&seq) // returning orgs: fenced-then-promoted in place on the original owners.
+	h.writeAll(&seq)
+
+	// (a) No acked write lost anywhere. ((b) split-brain and (c) availability were asserted
+	// continuously inside write().)
+	h.verifyNoLostWrites()
+
+	// Sanity: the proof actually drove ownership across pods (not a degenerate single-owner
+	// run) — more than one distinct (org,round)→pod binding was recorded.
+	distinct := map[string]bool{}
+	for _, pod := range h.roundOwner {
+		distinct[pod] = true
+	}
+	if len(distinct) < 2 {
+		t.Fatalf("proof did not exercise a handoff: only %d pod(s) ever owned a round", len(distinct))
+	}
+}
+
+func without(all []string, drop string) []string {
+	out := make([]string, 0, len(all))
+	for _, s := range all {
+		if s != drop {
+			out = append(out, s)
+		}
+	}
+	return out
+}

internal/org/snapshotcodec.go

@@ -0,0 +1,177 @@
+// Copyright 2025 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package org
+
+// snapshotcodec.go is the SWAPPABLE ship mechanism — the seam that lets HOW the durable
+// payload is produced/applied change without touching WHEN it ships (the fence, the
+// monotone round, CarryForward). Today the default codec checkpoints the WAL and copies
+// the whole file (framed with the cek key sidecar); a WAL-frame delta codec
+// (github.com/hanzoai/replicate — low-memory streaming) drops in behind this same
+// interface later, and the fence ships whatever bytes produce() returns and hands
+// whatever bytes it admitted to apply(). The codec works on PLAINTEXT payloads; envelope
+// sealing (Cipher) is orthogonal, applied by Sync/restore AROUND the codec — so the two
+// concerns (what bytes represent the DB vs. how they are encrypted at rest) never
+// complect.
+
+import (
+	"context"
+	"database/sql"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"os"
+
+	"github.com/hanzoai/vfs/replica"
+)
+
+// dekSuffix is cek's key-sidecar suffix (github.com/hanzoai/cloud/cek): the file holding
+// a database's wrapped page key. The wholeFile codec ships it beside the database bytes
+// so a successor can open the encrypted file. Kept as a local constant (a stable on-disk
+// convention) so this package stays dep-free of cek.
+const dekSuffix = ".dek"
+
+// errCorruptFrame is returned when a payload does not decode as a (sidecar, database)
+// frame this package wrote — fail closed rather than restore arbitrary bytes over a live
+// database.
+var errCorruptFrame = errors.New("org: durable payload is not a valid (sidecar,db) frame")
+
+// snapshotCodec produces a durable payload from the bound local database and applies a
+// restored payload back onto it. produce reads through db's sole connection so the
+// payload is consistent against concurrent writes; apply reconstructs the local file
+// (and any auxiliary files) atomically. Both operate on plaintext.
+type snapshotCodec interface {
+	// produce returns the durable payload for the database at dbPath, read through db
+	// (the store's single connection). The returned bytes are the fence's ship payload.
+	produce(ctx context.Context, db *sql.DB, dbPath string) ([]byte, error)
+	// apply writes a produced payload back onto the local database at dbPath. An empty
+	// payload (nothing shipped yet) is a no-op that keeps whatever the local file holds.
+	apply(dbPath string, payload []byte) error
+}
+
+// wholeFile is the default codec: fold the WAL into the real on-disk file so it holds
+// every committed page, copy the file bytes, and frame them with the cek key sidecar so a
+// successor can open the encrypted file. It is correct whether the file is encrypted
+// (production, sidecar present) or plaintext (pure-Go dev, no sidecar) — the same path, no
+// per-build branch.
+//
+// checkpoint is the crypto-integration seam. The SQLCipher page-level and plaintext
+// backends encrypt on WRITE, so the default nil path (a raw TRUNCATE checkpoint, fail
+// closed on busy) already leaves the real path holding fresh bytes. The pure-Go
+// encryption ENVELOPE instead defers encryption to Checkpoint/Close — the plaintext lives
+// on tmpfs and the real path is stale ciphertext until re-encrypted — so on that backend
+// the composition root injects cek's Checkpoint here (WithCheckpoint), and produce reads
+// the real path AFTER it, never shipping stale ciphertext (which would be a lost acked
+// write on takeover). A nil checkpoint means the write-time-encrypting default.
+type wholeFile struct {
+	checkpoint func(ctx context.Context, db *sql.DB) error
+}
+
+// produce folds the WAL into the real file, then copies it: the checkpoint runs FIRST so
+// the real path holds every committed page (and, on the envelope backend, fresh
+// ciphertext); the <db>.dek key sidecar (present only on an encrypting build) is read too
+// and framed with the database bytes.
+func (wf wholeFile) produce(ctx context.Context, db *sql.DB, dbPath string) ([]byte, error) {
+	if wf.checkpoint != nil {
+		// Envelope backend: its Checkpoint folds the WAL AND re-encrypts the real path.
+		// It takes db's SOLE connection itself, so it must run BEFORE we hold that
+		// connection (holding it first would deadlock at MaxOpenConns(1)).
+		if err := wf.checkpoint(ctx, db); err != nil {
+			return nil, fmt.Errorf("org: snapshot checkpoint %s: %w", dbPath, err)
+		}
+		// Then hold the SOLE connection across the file read, exactly as the default path
+		// does: os.ReadFile is a raw read that takes no SQLite lock, so a concurrent
+		// writer's commit — and the WAL auto-checkpoint it can trigger, which writes pages
+		// straight into the real path — would tear the image mid-read and ship a CORRUPT
+		// snapshot that the fence then acks at the lease round. A write that lands in the
+		// gap between the checkpoint and this acquire is simply not in the snapshot, which
+		// is safe: it has not been acked, so no acknowledged write is lost.
+		conn, err := db.Conn(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("org: snapshot conn %s: %w", dbPath, err)
+		}
+		defer conn.Close()
+		return readFramed(dbPath)
+	}
+	// Default backend (write-time encryption): hold db's SOLE connection (MaxOpenConns(1))
+	// across the TRUNCATE checkpoint AND the file read so no writer folds new frames
+	// mid-read. busy!=0 means the checkpoint could not fold all frames (a reader held the
+	// WAL) so the file is MISSING committed rows — fail closed (a partial snapshot shipped
+	// as complete is a silent lost write). (busy, logFrames, checkpointed) is the row.
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("org: snapshot conn %s: %w", dbPath, err)
+	}
+	defer conn.Close()
+	var busy, logFrames, checkpointed int
+	if err := conn.QueryRowContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &logFrames, &checkpointed); err != nil {
+		return nil, fmt.Errorf("org: snapshot checkpoint %s: %w", dbPath, err)
+	}
+	if busy != 0 {
+		return nil, fmt.Errorf("org: snapshot checkpoint %s did not complete (busy=%d, log=%d, checkpointed=%d) — snapshot would miss committed WAL frames", dbPath, busy, logFrames, checkpointed)
+	}
+	return readFramed(dbPath)
+}
+
+// readFramed reads the real-path database bytes and the cek key sidecar (absent on a
+// plaintext build) and frames them into one durable payload.
+func readFramed(dbPath string) ([]byte, error) {
+	main, err := os.ReadFile(dbPath)
+	if err != nil {
+		return nil, fmt.Errorf("org: snapshot read %s: %w", dbPath, err)
+	}
+	sidecar, err := os.ReadFile(dbPath + dekSuffix)
+	if err != nil && !os.IsNotExist(err) {
+		return nil, fmt.Errorf("org: snapshot read sidecar %s: %w", dbPath, err)
+	}
+	return frame(sidecar, main), nil
+}
+
+// apply splits the (sidecar, database) frame, atomically swaps the database bytes into
+// place (replica.RestoreFile, which MkdirAll's the parent), then writes the sidecar into
+// the now-existing directory. RestoreFile FIRST so a fresh successor whose orgs/<slug>/
+// dir does not exist yet gets the directory before the sidecar write. An empty payload is
+// a no-op.
+func (wholeFile) apply(dbPath string, payload []byte) error {
+	if len(payload) == 0 {
+		return nil
+	}
+	sidecar, main, err := unframe(payload)
+	if err != nil {
+		return err
+	}
+	if err := replica.RestoreFile(dbPath, main); err != nil {
+		return err
+	}
+	if len(sidecar) > 0 {
+		if err := os.WriteFile(dbPath+dekSuffix, sidecar, 0o600); err != nil {
+			return fmt.Errorf("org: write sidecar %s: %w", dbPath, err)
+		}
+	}
+	return nil
+}
+
+// frame prepends the sidecar under a uvarint length so a successor can split it from the
+// database bytes. len(sidecar)==0 ⇒ a leading 0 byte, i.e. a plaintext store.
+func frame(sidecar, main []byte) []byte {
+	buf := make([]byte, 0, binary.MaxVarintLen64+len(sidecar)+len(main))
+	var n [binary.MaxVarintLen64]byte
+	m := binary.PutUvarint(n[:], uint64(len(sidecar)))
+	buf = append(buf, n[:m]...)
+	buf = append(buf, sidecar...)
+	return append(buf, main...)
+}
+
+// unframe splits a framed payload back into (sidecar, database). The bound is checked as
+// sl > len(b)-m (a SUBTRACTION, never m+sl) so a maliciously large uvarint length cannot
+// wrap uint64 addition past the guard and panic the slice — it fails closed.
+func unframe(b []byte) (sidecar, main []byte, err error) {
+	sl, m := binary.Uvarint(b)
+	if m <= 0 || sl > uint64(len(b)-m) {
+		return nil, nil, errCorruptFrame
+	}
+	off := m + int(sl)
+	return b[m:off], b[off:], nil
+}

membership_k8s.go

@@ -0,0 +1,167 @@
+package cloud
+
+// membership_k8s.go is the LIVE writer-membership source: the discovery mechanism that
+// plugs into internal/org's Source seam (membership.go names "a K8s Endpoints poll
+// (prod)" as exactly this) so the horizontally-scaled cloud tracks its CHANGING pod set
+// instead of a static list. It is the fix for the rolling-upgrade outage: with a static
+// peer set, ha.Owner keeps electing a pod that is draining or already gone, and the
+// shard router forwards an org's requests to a dead pod; a live, READY-gated set drops
+// that pod the moment it starts terminating, so ha.Owner re-elects a live successor and
+// the org stays served.
+//
+// Capability-detected, NO on/off flag (the CTO one-way rule): in-cluster AND a selector
+// configured → live K8s membership; otherwise (dev, native-Go, no selector) → the
+// STATIC ShardPeers/self set. Native-Go always works: no cluster, no problem.
+//
+// It is a bounded LIST poll driven by internal/org.Membership's existing refresh loop
+// (which already retains the last-good set on a transient error and serves the hot-path
+// AmOwner check lock-free from an atomic snapshot). Each poll re-lists, so a dropped
+// connection self-heals on the next tick — no watch state to wedge. The ready-gating
+// predicate (podWriterEligible) is the ONE copy the future github.com/hanzoai/ha/k8s
+// shared source folds together with visor/object/coordinator.go's identical twin.
+
+import (
+	"context"
+	"os"
+	"strings"
+	"sync/atomic"
+
+	"github.com/hanzoai/cloud/internal/org"
+	luxlog "github.com/luxfi/log"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+// membershipSource returns the writer-membership Source for the durability fencer AND
+// the shard router. self and staticPeers are the fallback (self is always a member of
+// the static set); selector is CLOUD_PEER_SELECTOR; port is the http port stamped onto
+// each live peer's Addr so the shard router can dial it. When the process is in a
+// cluster and selector != "", it yields the live Ready pod set; otherwise it yields the
+// static set, so the exact same wiring runs everywhere.
+func membershipSource(staticPeers []org.Member, selector, port string, log luxlog.Logger) org.Source {
+	static := org.StaticSource(staticPeers...)
+	if strings.TrimSpace(selector) == "" {
+		return static // dev / native-Go / no selector: the static set is the whole writer set.
+	}
+	client, ns, err := inClusterPods()
+	if err != nil {
+		// Selector configured but not in a cluster (or the API is unreachable at boot):
+		// fall back to static rather than fail — the deployment still serves, just
+		// without live membership tracking. Logged so an in-cluster misconfig is visible.
+		if log != nil {
+			log.Warn("live K8s membership unavailable — using static peer set", "selector", selector, "err", err)
+		}
+		return static
+	}
+	if log != nil {
+		log.Info("live K8s membership enabled", "namespace", ns, "selector", selector)
+	}
+	var listFailures atomic.Int64
+	return func(ctx context.Context) ([]org.Member, error) {
+		list, err := client.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{LabelSelector: selector})
+		if err != nil {
+			// Fail closed: org.Membership keeps the last-good set, never flapping to
+			// empty (an empty set would strand every org with no owner).
+			//
+			// REPORT it, rate-limited. org.Membership's refresh loop DISCARDS this error,
+			// so without a log here a persistent failure (pods:list RBAC missing, API
+			// unreachable) is completely invisible: membership freezes at its last-good
+			// set — empty if the very first refresh failed — and every per-org write then
+			// fails closed with no clue why. Logged on the 1st failure and every 30th
+			// after (~1/min at the 2s refresh) so a blip stays quiet but an outage does not.
+			if n := listFailures.Add(1); log != nil && (n == 1 || n%30 == 0) {
+				log.Error("live writer-membership pod LIST failed — the writer set is FROZEN at its last-good value; while it is empty every per-org write fails closed",
+					"selector", selector, "namespace", ns, "consecutive_or_total_failures", n, "err", err)
+			}
+			return nil, err
+		}
+		listFailures.Store(0)
+		return readyMembers(list.Items, port), nil
+	}
+}
+
+// inClusterPods builds a typed clientset from the in-cluster service-account, mirroring
+// the access pattern visor/object/coordinator.go already uses. A non-nil error means
+// "not in a cluster" (dev / native-Go), and the caller falls back to static membership.
+func inClusterPods() (kubernetes.Interface, string, error) {
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, "", err
+	}
+	client, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return nil, "", err
+	}
+	return client, podNamespace(), nil
+}
+
+// readyMembers converts a pod list to the writer-eligible member set: each Ready,
+// non-terminating pod becomes a Member{ID: pod name (stable for the pod's life — the HRW
+// weight), Addr: podIP:port (the shard router dials it directly)}. A pod with no IP yet
+// (just scheduled) keeps an empty Addr; it is still HRW-eligible only once Ready, by
+// which point the IP is set.
+func readyMembers(pods []corev1.Pod, port string) []org.Member {
+	out := make([]org.Member, 0, len(pods))
+	for i := range pods {
+		p := &pods[i]
+		if !podWriterEligible(p) {
+			continue
+		}
+		addr := p.Status.PodIP
+		if addr != "" && port != "" {
+			addr = addr + ":" + port
+		}
+		out = append(out, org.Member{ID: p.Name, Addr: addr})
+	}
+	return out
+}
+
+// podWriterEligible reports whether a pod is a live writer candidate: Running and Ready,
+// and NOT terminating. The DeletionTimestamp gate is the fast drain signal — the instant
+// K8s marks a pod for deletion (a rolling upgrade's first step), it leaves the writer set
+// so ha.Owner re-elects a live successor BEFORE the pod stops serving; the Ready gate
+// covers an unready-but-not-terminating pod (starting up, or a failing readiness probe).
+// This is the ONE ready-gating predicate — identical to visor/object/coordinator.go's,
+// the twin the shared github.com/hanzoai/ha/k8s source will fold together.
+func podWriterEligible(p *corev1.Pod) bool {
+	if p.DeletionTimestamp != nil {
+		return false // terminating — drain it from the writer set immediately.
+	}
+	if p.Status.Phase != corev1.PodRunning {
+		return false
+	}
+	for _, c := range p.Status.Conditions {
+		if c.Type == corev1.PodReady {
+			return c.Status == corev1.ConditionTrue
+		}
+	}
+	return false // no Ready condition yet.
+}
+
+// podNamespace resolves this pod's namespace for the peer list: POD_NAMESPACE (Downward
+// API) when set, else the in-cluster service-account namespace file. Empty only outside a
+// cluster, where inClusterPods has already returned the static fallback.
+func podNamespace() string {
+	if ns := strings.TrimSpace(os.Getenv("POD_NAMESPACE")); ns != "" {
+		return ns
+	}
+	if b, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil {
+		if ns := strings.TrimSpace(string(b)); ns != "" {
+			return ns
+		}
+	}
+	return "default"
+}
+
+// httpPortOf extracts the port from a listen address (":8080" → "8080", "0.0.0.0:8080" →
+// "8080") for stamping onto a live peer's Addr. A value with no colon is returned as-is
+// (already a bare port); empty stays empty.
+func httpPortOf(listenAddr string) string {
+	listenAddr = strings.TrimSpace(listenAddr)
+	if i := strings.LastIndex(listenAddr, ":"); i >= 0 {
+		return listenAddr[i+1:]
+	}
+	return listenAddr
+}

membership_k8s_test.go

@@ -0,0 +1,135 @@
+package cloud
+
+// membership_k8s_test.go proves the parts of the live membership source that run without
+// a cluster: the ready-gating predicate (the safety-critical filter that keeps a
+// draining/unready pod out of the writer election), the pod→member conversion, and the
+// capability-detected static fallback. The live LIST against a real API server is a
+// staging-gated integration (a cluster is required), flagged in the report.
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hanzoai/cloud/internal/org"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func readyPod(name, ip string) corev1.Pod {
+	return corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+		Status: corev1.PodStatus{
+			Phase:      corev1.PodRunning,
+			PodIP:      ip,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
+		},
+	}
+}
+
+// TestPodWriterEligible is the drain-safety table: only a Running+Ready+non-terminating
+// pod is a writer candidate. A terminating pod (DeletionTimestamp set — a rolling
+// upgrade's first step) is excluded IMMEDIATELY so ha.Owner re-elects a live successor
+// before the pod stops serving.
+func TestPodWriterEligible(t *testing.T) {
+	terminating := readyPod("cloud-1", "10.0.0.1")
+	now := metav1.Now()
+	terminating.DeletionTimestamp = &now
+
+	notReady := readyPod("cloud-2", "10.0.0.2")
+	notReady.Status.Conditions[0].Status = corev1.ConditionFalse
+
+	pending := readyPod("cloud-3", "10.0.0.3")
+	pending.Status.Phase = corev1.PodPending
+
+	noCond := readyPod("cloud-4", "10.0.0.4")
+	noCond.Status.Conditions = nil
+
+	for _, tc := range []struct {
+		name string
+		pod  corev1.Pod
+		want bool
+	}{
+		{"running-ready", readyPod("cloud-0", "10.0.0.0"), true},
+		{"terminating", terminating, false},
+		{"ready-false", notReady, false},
+		{"pending", pending, false},
+		{"no-ready-condition", noCond, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := podWriterEligible(&tc.pod); got != tc.want {
+				t.Fatalf("podWriterEligible(%s) = %v, want %v", tc.name, got, tc.want)
+			}
+		})
+	}
+}
+
+// TestReadyMembers: a mixed pod list yields only the eligible pods, each as
+// Member{ID: pod name, Addr: podIP:port}. The terminating pod is dropped — it must never
+// appear in the set ha.Owner elects over.
+func TestReadyMembers(t *testing.T) {
+	now := metav1.Now()
+	draining := readyPod("cloud-1", "10.0.0.1")
+	draining.DeletionTimestamp = &now
+
+	members := readyMembers([]corev1.Pod{
+		readyPod("cloud-0", "10.0.0.0"),
+		draining,
+		readyPod("cloud-2", "10.0.0.2"),
+	}, "8080")
+
+	if len(members) != 2 {
+		t.Fatalf("want 2 eligible members, got %d (%+v)", len(members), members)
+	}
+	byID := map[string]string{}
+	for _, m := range members {
+		byID[m.ID] = m.Addr
+	}
+	if byID["cloud-0"] != "10.0.0.0:8080" || byID["cloud-2"] != "10.0.0.2:8080" {
+		t.Fatalf("member addrs wrong: %+v", byID)
+	}
+	if _, drained := byID["cloud-1"]; drained {
+		t.Fatal("terminating pod cloud-1 must not be in the writer set")
+	}
+}
+
+// TestMembershipSourceStaticFallback: with no selector (dev / native-Go) the source IS
+// the static set — the exact wiring runs everywhere, no cluster required.
+func TestMembershipSourceStaticFallback(t *testing.T) {
+	peers := []org.Member{{ID: "cloud-0", Addr: "cloud-0:8080"}, {ID: "cloud-1", Addr: "cloud-1:8080"}}
+	src := membershipSource(peers, "", "8080", nil)
+	got, err := src(context.Background())
+	if err != nil {
+		t.Fatalf("static source: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("static fallback want 2 members, got %d", len(got))
+	}
+}
+
+// TestMembershipSourceSelectorOutOfClusterFallsBack: a selector is set but the process
+// is NOT in a cluster (the test env) — the source must fall back to static rather than
+// fail, so a misconfigured selector never strands the deployment.
+func TestMembershipSourceSelectorOutOfClusterFallsBack(t *testing.T) {
+	peers := []org.Member{{ID: "solo", Addr: "solo:8080"}}
+	src := membershipSource(peers, "app.kubernetes.io/name=cloud", "8080", nil)
+	got, err := src(context.Background())
+	if err != nil {
+		t.Fatalf("out-of-cluster source must not error: %v", err)
+	}
+	if len(got) != 1 || got[0].ID != "solo" {
+		t.Fatalf("out-of-cluster selector must fall back to static self: %+v", got)
+	}
+}
+
+func TestHTTPPortOf(t *testing.T) {
+	for _, tc := range []struct{ in, want string }{
+		{":8080", "8080"},
+		{"0.0.0.0:8080", "8080"},
+		{"8080", "8080"},
+		{"", ""},
+	} {
+		if got := httpPortOf(tc.in); got != tc.want {
+			t.Fatalf("httpPortOf(%q) = %q, want %q", tc.in, got, tc.want)
+		}
+	}
+}

orgdb.go

@@ -262,8 +262,44 @@ func (c *OrgStore[T]) forPath(slug, dbKey, path string) (T, error) {
 	var zero T
 	c.mu.Lock()
 	if st, ok := c.byPath[path]; ok {
+		// Cache hit. If this durable store opened DEGRADED (read-only) and this replica
+		// has since become the org's elected owner — a rolling-upgrade membership change
+		// — promote it IN PLACE (M3): re-acquire the lease, hydrate the latest snapshot,
+		// and swap in a writer handle, with no process restart. The check is I/O-free and
+		// only does real work when the store is unowned AND newly elected (rare).
+		d := c.durables[path]
+		if d == nil || !d.PendingPromotion() {
+			c.mu.Unlock()
+			return st, nil
+		}
+		if inf, promoting := c.inflight[path]; promoting {
+			c.mu.Unlock()
+			<-inf.done
+			return inf.st, inf.err
+		}
+		inf := &openState[T]{done: make(chan struct{})}
+		c.inflight[path] = inf
+		delete(c.byPath, path)
+		delete(c.durables, path)
 		c.mu.Unlock()
-		return st, nil
+
+		st2, d2, err := c.promote(slug, dbKey, path, st, d)
+		inf.st, inf.d, inf.err = st2, d2, err
+		c.mu.Lock()
+		delete(c.inflight, path)
+		if d2 != nil { // a usable store (promoted writer, or the kept read-only one)
+			c.byPath[path] = st2
+			c.durables[path] = d2
+		}
+		c.mu.Unlock()
+		close(inf.done)
+		if err != nil && d2 != nil && c.log != nil {
+			// Reopen degraded again (transient store/membership blip): still serving the
+			// prior read-only state, so log and keep availability rather than surface it.
+			c.log.Warn("org store promotion incomplete — serving prior state", "subsystem", c.subsystem, "org", slug, "err", err)
+			return st2, nil
+		}
+		return st2, err
 	}
 	// Local-only (no Durability): open under c.mu — disk I/O only, unchanged from the
 	// pre-durability cache.
@@ -338,6 +374,32 @@ func (c *OrgStore[T]) openDurable(slug, dbKey, path string) (T, *org.Durable, er
 	return st, d, nil
 }
 
+// promote upgrades a degraded (read-only) store to the writer, in place, when this
+// replica has become the org's elected owner (M3 — no process restart). It PROBES first
+// (TryClaim: only the lease CAS, no file I/O), so a transient membership/store blip that
+// is not yet claimable leaves the read-only handle serving untouched. Only once the lease
+// is claimed does it quiesce — close the read-only handle — and re-open as owner, whose
+// Hydrate renews that same lease and CarryForward-restores the latest snapshot under the
+// FRESH handle (never under the live one — the file swap is exactly why the reopen is
+// required). The fence's monotone round makes this safe against a still-running prior
+// owner: the reopen claims a strictly higher round, so any late ship from the deposed
+// writer is rejected — never two live writers for one org.
+//
+// Returns the store to (re)publish and an error to LOG: (new writer, nil) on success;
+// (the same read-only store, nil-or-err) when not yet claimable; (zero, err) only if the
+// reopen hard-fails after the claim (a local disk error — the entry is then dropped and
+// the failure surfaced).
+func (c *OrgStore[T]) promote(slug, dbKey, path string, old T, oldDur *org.Durable) (T, *org.Durable, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), durableOpTimeout)
+	claimed, err := oldDur.TryClaim(ctx)
+	cancel()
+	if err != nil || !claimed {
+		return old, oldDur, err // not the owner yet / store blip: keep serving read-only.
+	}
+	_ = old.Close() // quiesce: release the read-only handle before CarryForward swaps the file.
+	return c.openDurable(slug, dbKey, path)
+}
+
 // Sync ships the org's local file to its durable object, fenced at the lease round
 // (the ship-before-ack step a durable subsystem calls after a write commits). It
 // returns acked=false when this replica is not the owner or was deposed mid-request

serve.go

@@ -112,7 +112,7 @@ func Serve(specs []MountSpec, enable []string) error {
 	// This is what lifts the deployment off replicas:1 without any shared RWX volume —
 	// per-pod RWO PVC + org→owner routing = one writer per tenant file. See
 	// shardrouter.go.
-	shardRtr := newShardRouter(cfg, deps.Logger)
+	shardRtr := newShardRouter(cfg, deps.Logger, deps.LiveMembers)
 	if shardRtr != nil {
 		deps.Logger.Info("shard routing ENABLED (horizontal writer scale)",
 			"self", shardRtr.self, "peers", shardRtr.peerIDs(),
@@ -130,14 +130,21 @@ func Serve(specs []MountSpec, enable []string) error {
 	// bootstrap via cloud.RegisterTelemetryInstaller (the cycle-free inversion).
 	telemetryShutdown := installTelemetry(context.Background(), "hanzo-cloud")
 
-	// Data-plane encryption posture (cek). On an encryption-capable build a
-	// missing/invalid CLOUD_KMS_MASTER_KEY_REF makes the FIRST store open fail
-	// closed (MountAll aborts) — the same fail-closed stance as the KMS store; we
-	// surface it here so the posture is never silent.
-	if cek.Encrypting() {
+	// Data-plane encryption posture (cek). Every build encrypts a keyed store — the
+	// live libsqlcipher codec in production, the pure-Go codec envelope in dev/CI —
+	// so a store either opens keyed-and-encrypted or fails closed; there is no
+	// plaintext-at-rest mode. EnsureDevKey gives a pure-Go dev/CI build with no
+	// configured key a deterministic dev key so it runs encrypted with zero config;
+	// a production (codec-linked) build with a missing/invalid CLOUD_KMS_MASTER_KEY_REF
+	// makes the FIRST store open fail closed (MountAll aborts). It runs BEFORE the
+	// posture read below, which caches the resolved key. Surfaced so it is never silent.
+	switch {
+	case cek.EnsureDevKey():
+		deps.Logger.Warn("data-plane encryption ACTIVE with a DEV key (pure-Go build, no KMS key configured — dev/CI only)")
+	case cek.Encrypting():
 		deps.Logger.Info("data-plane encryption ACTIVE (SQLCipher at rest, per-db DEK)")
-	} else {
-		deps.Logger.Warn("data-plane encryption OFF (pure-Go dev build, or missing key on a capable build → store opens fail closed)")
+	default:
+		deps.Logger.Warn("data-plane encryption posture: missing/invalid key on a production build → store opens fail closed")
 	}
 
 	// ReadBufferSize raises the fasthttp header ceiling above the 4 KiB fiber
@@ -453,6 +460,16 @@ func Serve(specs []MountSpec, enable []string) error {
 	select {
 	case <-ctx.Done():
 		deps.Logger.Info("shutdown requested")
+		// Graceful drain: go NotReady so peers re-elect this pod's orgs to live successors
+		// (each hydrates the latest fenced snapshot, M3) BEFORE we stop serving, then pause
+		// for that to propagate through the membership refresh. Only when sharding is active
+		// — a single-pod deployment has no successor, so it drains immediately and relies on
+		// the final ship (CloseAll) below. In-flight requests drain in app.ShutdownWithContext.
+		SetDraining()
+		if shardRtr != nil {
+			deps.Logger.Info("draining: NotReady, waiting for peers to re-elect owned orgs", "grace", shardDrainGrace)
+			time.Sleep(shardDrainGrace)
+		}
 	case err := <-listenErr:
 		return fmt.Errorf("listen: %w", err)
 	}
@@ -499,8 +516,18 @@ func healthMux() *http.ServeMux {
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write([]byte(`{"status":"ok"}`))
 	}
-	mux.HandleFunc("/healthz", ok)
-	mux.HandleFunc("/readyz", ok)
+	mux.HandleFunc("/healthz", ok) // liveness: stays 200 while draining (finish the drain).
+	// readiness: 503 once draining so K8s marks the pod NotReady — removed from endpoints
+	// AND from every peer's writer election — before it stops serving (graceful handoff).
+	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
+		if Draining() {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = w.Write([]byte(`{"status":"draining"}`))
+			return
+		}
+		ok(w, r)
+	})
 	mux.HandleFunc("/health", ok)
 	mux.HandleFunc("/metrics", func(w http.ResponseWriter, _ *http.Request) {
 		w.Header().Set("Content-Type", "text/plain; version=0.0.4")

shardrouter.go

@@ -72,7 +72,14 @@ const shardForwardTimeout = 25 * time.Second
 // in which case Middleware is a pass-through — byte-identical to today.
 type shardRouter struct {
 	self  string      // this pod's stable id (StatefulSet ordinal name, e.g. cloud-1)
-	peers []ha.Member // the full, stable writer set (id@addr), identical on every pod
+	peers []ha.Member // static writer set (CLOUD_PEERS) — the fallback + the boot log label
+
+	// members is the LIVE writer set the durable plane supplies (the SAME election
+	// snapshot the fencer uses). Non-nil ⇒ route on it, so a draining/dead pod's orgs go
+	// to the ready successor that hydrates them (M3). nil ⇒ static peers only: without the
+	// durable plane a peer cannot serve another pod's local-only files, so ownership stays
+	// pinned to the ordinal (which reattaches its PVC on restart).
+	members func() []ha.Member
 
 	log luxlog.Logger
 
@@ -80,19 +87,40 @@ type shardRouter struct {
 	clients map[string]*fasthttp.HostClient // addr → pooled streaming client, lazily built
 }
 
-// newShardRouter builds the router from config, or returns nil when sharding is off
-// (CLOUD_PEERS names ≤1 pod, or no self id) so the caller wires no middleware and the
-// single-pod path is unchanged. Config.Validate has already refused to boot a
-// multi-peer set that does not contain self, so a non-nil router always owns a shard.
-func newShardRouter(cfg *Config, log luxlog.Logger) *shardRouter {
+// newShardRouter builds the router, or returns nil when routing is off (single pod) so the
+// caller wires no middleware. Two activation paths: LIVE membership (live != nil — the
+// durable plane is on and tracks the pod set through K8s) activates routing on any stable
+// self id; otherwise the STATIC contract — CLOUD_PEERS names >1 pod AND self is one of
+// them (Config.Validate enforced this) — as before. self is the pod name (CLOUD_POD_NAME /
+// hostname), matching the id the live membership source assigns each pod.
+func newShardRouter(cfg *Config, log luxlog.Logger, live func() []ha.Member) *shardRouter {
 	peers := parsePeers(cfg.ShardPeers)
-	self := strings.TrimSpace(cfg.ShardSelf)
+	self := firstNonEmptyStr(strings.TrimSpace(cfg.ShardSelf), hostnameOr(""))
+	if live != nil {
+		if self == "" {
+			return nil // no stable id ⇒ cannot decide "do I own this org"; single-pod path.
+		}
+		return &shardRouter{self: self, peers: peers, members: live, log: log, clients: map[string]*fasthttp.HostClient{}}
+	}
 	if len(peers) < 2 || self == "" || !peersContain(peers, self) {
 		return nil
 	}
 	return &shardRouter{self: self, peers: peers, log: log, clients: map[string]*fasthttp.HostClient{}}
 }
 
+// set is the writer set to elect over: the LIVE membership when the durable plane provides
+// it (a rolling upgrade's current pods — draining/dead ones already dropped), else the
+// static CLOUD_PEERS set. A momentarily-empty live snapshot (a refresh blip) falls back to
+// peers so no org is ever stranded with no owner.
+func (r *shardRouter) set() []ha.Member {
+	if r.members != nil {
+		if m := r.members(); len(m) > 0 {
+			return m
+		}
+	}
+	return r.peers
+}
+
 // peerIDs returns the member ids for logging.
 func (r *shardRouter) peerIDs() []string {
 	ids := make([]string, len(r.peers))
@@ -122,7 +150,7 @@ func (r *shardRouter) Middleware() zip.Handler {
 		if slug == "" {
 			return c.Continue()
 		}
-		owner, ok := ha.Owner(slug, r.peers)
+		owner, ok := ha.Owner(slug, r.set())
 		if !ok || owner.ID == r.self {
 			return c.Continue()
 		}
@@ -140,11 +168,13 @@ func (r *shardRouter) forward(c *zip.Ctx, owner ha.Member, slug string) error {
 	req := fc.Request()
 	resp := fc.Response()
 
-	// Loop guard / divergence fail-closed. Under the static, identical peer set this
-	// is unreachable (the owner always recomputes owner==self); if it ever fires,
-	// membership diverged and serving locally would touch another shard's files, so
-	// we refuse (421) instead. A client that forges the header only 421s ITS OWN
-	// cross-shard request — never another tenant, never availability for others.
+	// Loop guard / divergence fail-closed. On the static peer set this is unreachable
+	// (the owner always recomputes owner==self). Under LIVE membership two pods can briefly
+	// hold divergent views mid-refresh, so an already-forwarded request may reach a pod that
+	// no longer owns the org; rather than loop or serve another shard's files, we refuse
+	// (421) — a retryable, fail-closed signal that self-heals the instant views converge
+	// (bounded by the 2s refresh). A forged header only 421s the client's OWN cross-shard
+	// request — never another tenant, never others' availability.
 	if len(req.Header.Peek(shardHopHeader)) > 0 {
 		r.log.Error("shard loop guard tripped — membership divergence", "org_slug", slug, "owner", owner.ID, "self", r.self)
 		resp.Reset()

shardrouter_test.go

@@ -311,14 +311,14 @@ func TestNewShardRouter_DisabledCases(t *testing.T) {
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			cfg := &Config{ShardPeers: tc.peers, ShardSelf: tc.self}
-			if r := newShardRouter(cfg, testLog()); r != nil {
+			if r := newShardRouter(cfg, testLog(), nil); r != nil {
 				t.Fatalf("newShardRouter(%q, %q) = non-nil, want nil (sharding off)", tc.peers, tc.self)
 			}
 		})
 	}
 	// Enabled: 3 peers, self in set.
 	cfg := &Config{ShardPeers: "cloud-0@a:8000,cloud-1@b:8000,cloud-2@c:8000", ShardSelf: "cloud-1"}
-	r := newShardRouter(cfg, testLog())
+	r := newShardRouter(cfg, testLog(), nil)
 	if r == nil {
 		t.Fatalf("newShardRouter with valid 3-peer set returned nil")
 	}
@@ -327,6 +327,52 @@ func TestNewShardRouter_DisabledCases(t *testing.T) {
 	}
 }
 
+// TestShardRouterLiveMembership: with a live-members source (durable plane on) the router
+// elects over the CURRENT set, so removing a pod re-homes its orgs to a live successor,
+// and the router activates even with no CLOUD_PEERS (a Deployment's random pod names).
+func TestShardRouterLiveMembership(t *testing.T) {
+	var live []ha.Member
+	src := func() []ha.Member { return live }
+
+	// No CLOUD_PEERS, but a live source + a stable self ⇒ routing is ON.
+	cfg := &Config{ShardPeers: "", ShardSelf: "cloud-1"}
+	r := newShardRouter(cfg, testLog(), src)
+	if r == nil {
+		t.Fatal("live membership must activate routing even without CLOUD_PEERS")
+	}
+
+	// Full set: every org has SOME owner drawn from the live members.
+	live = []ha.Member{{ID: "cloud-0", Addr: "10.0.0.0:8080"}, {ID: "cloud-1", Addr: "10.0.0.1:8080"}, {ID: "cloud-2", Addr: "10.0.0.2:8080"}}
+	owner3, ok := ha.Owner("acme", r.set())
+	if !ok {
+		t.Fatal("owner over the live set must resolve")
+	}
+
+	// Drain the elected owner out of the live set: the org re-homes to a DIFFERENT live pod
+	// (never the drained one) — the routing half of zero-downtime.
+	live = removeMember(live, owner3.ID)
+	owner2, ok := ha.Owner("acme", r.set())
+	if !ok || owner2.ID == owner3.ID {
+		t.Fatalf("after draining %s the org must re-home to a live pod, got %q (ok=%v)", owner3.ID, owner2.ID, ok)
+	}
+
+	// Momentarily-empty live snapshot falls back to the static peers, never stranding.
+	live = nil
+	if got := r.set(); len(got) != len(r.peers) {
+		t.Fatalf("empty live snapshot must fall back to static peers (%d), got %d", len(r.peers), len(got))
+	}
+}
+
+func removeMember(ms []ha.Member, id string) []ha.Member {
+	out := ms[:0:0]
+	for _, m := range ms {
+		if m.ID != id {
+			out = append(out, m)
+		}
+	}
+	return out
+}
+
 func TestParsePeers(t *testing.T) {
 	got := parsePeers("cloud-0@cloud-0.cloud.hanzo.svc:8000, cloud-1@cloud-1.cloud.hanzo.svc:8000 ,cloud-2@c:8000")
 	if len(got) != 3 {