From 416c056a6dfa919100dd889c7f4a4850abc3a4b9 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Sun, 19 Jul 2026 21:37:30 +0900
Subject: [PATCH 1/5] tso: complete centralized runtime semantics

---
 kv/tso.go | 45 +++++++++++++++++++++++++++++++++++++--------
 1 file changed, 37 insertions(+), 8 deletions(-)

diff --git a/kv/tso.go b/kv/tso.go
index c92840db1..2031122d1 100644
--- a/kv/tso.go
+++ b/kv/tso.go
@@ -199,24 +199,53 @@ func NextTimestampAfterThrough(ctx context.Context, coord Coordinator, startTS u
 	return nextTimestampAfterObserved(ctx, coord, clock, startTS, label)
 }
 
-// TimestampAllocatorThrough returns the allocator configured behind a
-// coordinator or coordinator decorator.
+// TimestampAllocatorThrough returns the currently active allocator behind a
+// coordinator or coordinator decorator. A runtime allocator in legacy mode is
+// intentionally hidden so normal write paths use the coordinator HLC fallback.
 func TimestampAllocatorThrough(coord Coordinator) (TimestampAllocator, bool) {
 	if provider, ok := coord.(TimestampAllocatorProvider); ok {
-		alloc := provider.TimestampAllocator()
-		return alloc, alloc != nil
+		return resolveTimestampAllocator(provider.TimestampAllocator())
 	}
 	switch c := coord.(type) {
 	case *Coordinate:
-		if c != nil && c.tsAllocator != nil {
-			return c.tsAllocator, true
+		if c != nil {
+			return resolveTimestampAllocator(c.tsAllocator)
 		}
 		return nil, false
 	case *ShardedCoordinator:
-		if c != nil && c.tsAllocator != nil {
-			return c.tsAllocator, true
+		if c != nil {
+			return resolveTimestampAllocator(c.tsAllocator)
 		}
 		return nil, false
+	default:
+		alloc, ok := coord.(TimestampAllocator)
+		if !ok {
+			return nil, false
+		}
+		return resolveTimestampAllocator(alloc)
+	}
+}
+
+// ConfiguredTimestampAllocatorThrough returns the configured allocator without
+// resolving runtime-mode decorators. It is used by long-lived internal servers
+// that must keep a DynamicTimestampAllocator reference across later mode-file
+// reloads while still falling back to HLC when that allocator reports legacy.
+func ConfiguredTimestampAllocatorThrough(coord Coordinator) (TimestampAllocator, bool) {
+	if provider, ok := coord.(TimestampAllocatorProvider); ok {
+		alloc := provider.TimestampAllocator()
+		return alloc, alloc != nil
+	}
+	switch c := coord.(type) {
+	case *Coordinate:
+		if c == nil || c.tsAllocator == nil {
+			return nil, false
+		}
+		return c.tsAllocator, true
+	case *ShardedCoordinator:
+		if c == nil || c.tsAllocator == nil {
+			return nil, false
+		}
+		return c.tsAllocator, true
 	default:
 		alloc, ok := coord.(TimestampAllocator)
 		return alloc, ok

From 9b9b7cf1e73cdaf2eb9175cbde8c0fb65ad39dde Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Sun, 19 Jul 2026 23:24:30 +0900
Subject: [PATCH 2/5] tso: add runtime operations and observability

---
 Makefile                                      |  47 +-
 adapter/internal.go                           |  28 +-
 adapter/internal_test.go                      |  52 ++
 docs/centralized_tso_operations.md            | 115 +++++
 ...2026_04_16_implemented_centralized_tso.md} | 100 ++--
 .../2026_05_28_implemented_tla_safety_spec.md |  15 +-
 .../2026_06_23_proposed_scaling_roadmap.md    |  44 +-
 kv/coordinator.go                             |   6 +-
 kv/keyviz_label.go                            |   2 +-
 kv/sharded_coordinator.go                     |  14 +-
 kv/tso.go                                     |  15 +
 kv/tso_fanout_benchmark_test.go               | 140 ++++++
 kv/tso_raft.go                                | 129 ++++-
 kv/tso_runtime.go                             | 453 ++++++++++++++++++
 kv/tso_runtime_test.go                        | 407 ++++++++++++++++
 main.go                                       | 195 +++++---
 main_tso_routing_test.go                      |  58 ++-
 monitoring/prometheus/rules/tso-alerts.yml    | 103 ++++
 monitoring/registry.go                        |  13 +
 monitoring/tso.go                             | 146 ++++++
 monitoring/tso_test.go                        |  48 ++
 21 files changed, 1940 insertions(+), 190 deletions(-)
 create mode 100644 docs/centralized_tso_operations.md
 rename docs/design/{2026_04_16_partial_centralized_tso.md => 2026_04_16_implemented_centralized_tso.md} (90%)
 create mode 100644 kv/tso_fanout_benchmark_test.go
 create mode 100644 kv/tso_runtime.go
 create mode 100644 kv/tso_runtime_test.go
 create mode 100644 monitoring/prometheus/rules/tso-alerts.yml
 create mode 100644 monitoring/tso.go
 create mode 100644 monitoring/tso_test.go

diff --git a/Makefile b/Makefile
index 9dc641ba6..c7e3a8f70 100644
--- a/Makefile
+++ b/Makefile
@@ -29,35 +29,40 @@ TLA_LIB      := ../lib
 
 .PHONY: tla-check tla-tools
 
-tla-tools: $(TLA_JAR)
-
-$(TLA_JAR):
+tla-tools:
 	@mkdir -p $(dir $(TLA_JAR))
-	@echo "Downloading tla2tools.jar $(TLA_VERSION)..."
-	@curl -fsSL -o $(TLA_JAR).tmp $(TLA_URL)
-	@# Prefer sha256sum (GNU coreutils, universal on Linux); fall back to
-	@# shasum -a 256 (default on macOS).  Either yields the same hex
-	@# digest in the first whitespace-delimited field.
-	@if command -v sha256sum >/dev/null 2>&1; then \
-		actual=$$(sha256sum $(TLA_JAR).tmp | awk '{print $$1}'); \
-	elif command -v shasum >/dev/null 2>&1; then \
-		actual=$$(shasum -a 256 $(TLA_JAR).tmp | awk '{print $$1}'); \
-	else \
-		echo "ERROR: neither sha256sum nor shasum is available."; \
-		rm -f $(TLA_JAR).tmp; \
-		exit 1; \
+	@set -eu; \
+	sha256_file() { \
+		if command -v sha256sum >/dev/null 2>&1; then \
+			sha256sum "$$1" | awk '{print $$1}'; \
+		elif command -v shasum >/dev/null 2>&1; then \
+			shasum -a 256 "$$1" | awk '{print $$1}'; \
+		else \
+			echo "ERROR: neither sha256sum nor shasum is available." >&2; \
+			return 1; \
+		fi; \
+	}; \
+	actual=""; \
+	if [ -f "$(TLA_JAR)" ]; then actual=$$(sha256_file "$(TLA_JAR)"); fi; \
+	if [ "$$actual" = "$(TLA_SHA256)" ]; then \
+		echo "tla2tools.jar ready at $(TLA_JAR) (SHA-256 $(TLA_SHA256))"; \
+		exit 0; \
 	fi; \
+	echo "Downloading tla2tools.jar $(TLA_VERSION)..."; \
+	trap 'rm -f "$(TLA_JAR).tmp"' EXIT HUP INT TERM; \
+	curl -fsSL -o "$(TLA_JAR).tmp" "$(TLA_URL)"; \
+	actual=$$(sha256_file "$(TLA_JAR).tmp"); \
 	if [ "$$actual" != "$(TLA_SHA256)" ]; then \
 		echo "ERROR: tla2tools.jar SHA-256 mismatch."; \
 		echo "  expected: $(TLA_SHA256)"; \
 		echo "  actual:   $$actual"; \
-		rm -f $(TLA_JAR).tmp; \
 		exit 1; \
-	fi
-	@mv $(TLA_JAR).tmp $(TLA_JAR)
-	@echo "tla2tools.jar ready at $(TLA_JAR) (SHA-256 $(TLA_SHA256))"
+	fi; \
+	mv "$(TLA_JAR).tmp" "$(TLA_JAR)"; \
+	trap - EXIT HUP INT TERM; \
+	echo "tla2tools.jar ready at $(TLA_JAR) (SHA-256 $(TLA_SHA256))"
 
-tla-check: $(TLA_JAR)
+tla-check: tla-tools
 	@# Per-module orchestration lives in scripts/tla-check.sh so adding
 	@# M3..M5 only needs an entry in that script's TLA_MODULES array
 	@# and a `case` line for the gap-invariant string.  The script does
diff --git a/adapter/internal.go b/adapter/internal.go
index bf4279e31..c78bde7ea 100644
--- a/adapter/internal.go
+++ b/adapter/internal.go
@@ -101,11 +101,17 @@ func (i *Internal) stampTimestamps(ctx context.Context, req *pb.ForwardRequest)
 func (i *Internal) nextTimestamp(ctx context.Context, label string) (uint64, error) {
 	if i.tsAllocator != nil {
 		ts, err := i.tsAllocator.Next(ctx)
-		if err != nil {
+		if err == nil {
+			return ts, nil
+		}
+		if !errors.Is(err, kv.ErrTSOAllocatorRequired) {
 			return 0, errors.Wrap(err, label)
 		}
-		return ts, nil
 	}
+	return i.nextLegacyTimestamp(label)
+}
+
+func (i *Internal) nextLegacyTimestamp(label string) (uint64, error) {
 	if i.clock == nil {
 		return 1, nil
 	}
@@ -121,15 +127,23 @@ func (i *Internal) nextTimestampAfter(ctx context.Context, min uint64, label str
 		return 0, errors.Wrap(kv.ErrTxnCommitTSRequired, label)
 	}
 	if i.tsAllocator != nil {
-		return i.nextTimestampAfterFromAllocator(ctx, min, label)
+		ts, err := i.nextTimestampAfterFromAllocator(ctx, min, label)
+		if err == nil {
+			return ts, nil
+		}
+		if !errors.Is(err, kv.ErrTSOAllocatorRequired) {
+			return 0, err
+		}
 	}
+	return i.nextLegacyTimestampAfter(min, label)
+}
+
+func (i *Internal) nextLegacyTimestampAfter(min uint64, label string) (uint64, error) {
 	if i.clock == nil {
 		return min + 1, nil
 	}
-	if i.clock != nil {
-		i.clock.Observe(min)
-	}
-	ts, err := i.nextTimestamp(ctx, label)
+	i.clock.Observe(min)
+	ts, err := i.nextLegacyTimestamp(label)
 	if err != nil {
 		return 0, err
 	}
diff --git a/adapter/internal_test.go b/adapter/internal_test.go
index fcabc5356..a10fbb775 100644
--- a/adapter/internal_test.go
+++ b/adapter/internal_test.go
@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/binary"
 	"testing"
+	"time"
 
 	"github.com/bootjp/elastickv/kv"
 	pb "github.com/bootjp/elastickv/proto"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -83,6 +85,35 @@ func TestFillForwardedTxnCommitTS_AssignsCommitTS(t *testing.T) {
 	require.Equal(t, meta.CommitTS, commitTS)
 }
 
+func TestFillForwardedTxnCommitTS_FallsBackWhenRuntimeAllocatorLegacy(t *testing.T) {
+	t.Parallel()
+
+	clock := kv.NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	i := &Internal{
+		clock:       clock,
+		tsAllocator: internalLegacyRuntimeAllocator{},
+	}
+	startTS := uint64(10)
+	reqs := []*pb.Request{
+		{
+			IsTxn: true,
+			Phase: pb.Phase_COMMIT,
+			Mutations: []*pb.Mutation{
+				{
+					Op:    pb.Op_PUT,
+					Key:   []byte(kv.TxnMetaPrefix),
+					Value: kv.EncodeTxnMeta(kv.TxnMeta{PrimaryKey: []byte("k"), CommitTS: 0}),
+				},
+			},
+		},
+	}
+
+	commitTS, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, startTS)
+	require.NoError(t, err)
+	require.Greater(t, commitTS, startTS)
+}
+
 func TestFillForwardedTxnCommitTS_PreservesExistingCommitTS(t *testing.T) {
 	t.Parallel()
 
@@ -175,6 +206,21 @@ func TestFillForwardedTxnCommitTS_StampsCommitTSValueOffset(t *testing.T) {
 	require.Zero(t, reqs[0].Mutations[1].CommitTsValueOffset)
 }
 
+func TestStampRawTimestamps_FallsBackWhenRuntimeAllocatorLegacy(t *testing.T) {
+	t.Parallel()
+
+	clock := kv.NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	i := &Internal{
+		clock:       clock,
+		tsAllocator: internalLegacyRuntimeAllocator{},
+	}
+	reqs := []*pb.Request{{Mutations: []*pb.Mutation{{Op: pb.Op_PUT, Key: []byte("k"), Value: []byte("v")}}}}
+
+	require.NoError(t, i.stampRawTimestamps(context.Background(), reqs))
+	require.NotZero(t, reqs[0].Ts)
+}
+
 func TestFillForwardedTxnCommitTS_PrepareAllowsAlreadyStampedOffsets(t *testing.T) {
 	t.Parallel()
 
@@ -241,3 +287,9 @@ func TestStampTxnTimestamps_UsesSingleTxnStartTS(t *testing.T) {
 	require.Greater(t, meta.CommitTS, uint64(9))
 	require.Equal(t, meta.CommitTS, commitTS)
 }
+
+type internalLegacyRuntimeAllocator struct{}
+
+func (internalLegacyRuntimeAllocator) Next(context.Context) (uint64, error) {
+	return 0, errors.WithStack(kv.ErrTSOAllocatorRequired)
+}
diff --git a/docs/centralized_tso_operations.md b/docs/centralized_tso_operations.md
new file mode 100644
index 000000000..c30bb8f8c
--- /dev/null
+++ b/docs/centralized_tso_operations.md
@@ -0,0 +1,115 @@
+# Centralized TSO Operations
+
+This runbook covers runtime mode changes, production signals, and the rollout
+gates for the dedicated group-0 timestamp oracle. The implementation design is
+in `docs/design/2026_04_16_implemented_centralized_tso.md`.
+
+## Runtime configuration
+
+Start every node with the same atomically replaceable mode file:
+
+```text
+--tsoModeFile=/etc/elastickv/tso-mode
+--tsoModeReloadInterval=5s
+--tsoBatchSize=256
+```
+
+The file contains exactly one mode: `legacy`, `shadow`, `cutover`, or
+`phase-d`. `--tsoModeFile` cannot be combined with `--tsoEnabled`,
+`--tsoShadowEnabled`, or `--tsoPhaseDEnabled`; those startup-only flags remain
+for backward compatibility. Runtime mode reload requires the dedicated group 0.
+
+Replace the file atomically so a poll cannot observe a truncated value:
+
+```sh
+printf '%s\n' shadow > /etc/elastickv/tso-mode.tmp
+mv /etc/elastickv/tso-mode.tmp /etc/elastickv/tso-mode
+```
+
+An unreadable file, unknown value, backward transition, or skipped phase is
+rejected without changing the live allocator. Runtime transitions must be
+adjacent and one-way:
+
+```text
+legacy -> shadow -> cutover -> phase-d
+```
+
+The durable group-0 markers override stale local configuration. Once cutover
+or Phase D is committed, a node cannot return to legacy or shadow issuance even
+if its mode file moves backward. An allocation already in flight during a
+reload may finish under the preceding process mode, but cutover-side requests
+still use group 0 and the durable markers never move backward.
+
+## Rollout gates
+
+1. Deploy the binary and group 0 while every mode file remains `legacy`.
+2. Change every node to `shadow`. Confirm all nodes expose
+   `elastickv_tso_mode{mode="shadow"} == 1`.
+3. Hold shadow mode until `legacy_overlap` remains zero for a full 15-minute
+   window, allocation errors remain below 1 percent, and allocation p99 remains
+   below 50 ms.
+4. Change nodes to `cutover`. The first production reservation commits the
+   one-way cutover marker. A node still in shadow observes that marker and
+   returns the dedicated TSO value.
+5. Confirm every binary supports Phase D and every node is on the dedicated
+   path. Then change nodes to `phase-d`. The first Phase-D reservation commits
+   its floor marker before returning a new window.
+
+After the cutover marker commits, rollback means restoring service around the
+dedicated TSO path; it never means returning to legacy issuance. Phase D has the
+same one-way rule and additionally keeps data-group HLC renewal retired.
+
+## Metrics and alerts
+
+| Signal | Meaning | Gate or threshold |
+|---|---|---|
+| `elastickv_tso_request_duration_seconds` | Local/remote reserve and validation attempt latency, split by outcome | Warning at reserve p99 > 50 ms for 10m; critical at > 200 ms for 5m |
+| `elastickv_tso_shadow_comparisons_total` | Accepted candidates, overlap discards, cutover bypasses, and errors | `legacy_overlap` must be zero for 15m before cutover |
+| `elastickv_tso_shadow_divergence_timestamps` | Absolute candidate-to-floor distance for accepted and discarded shadow comparisons | Investigate sustained growth before cutover |
+| `elastickv_tso_mode` | One-hot process-local mode | More than one mode for 15m warns about a stalled rollout |
+| `elastickv_tso_mode_reload_total` | Applied and rejected reloads plus file read/parse failures | Any failure in 10m warns |
+| `elastickv_tso_durable_state` | Applied `cutover` and `phase_d` markers | Phase D without cutover is critical |
+
+The checked rules are in `monitoring/prometheus/rules/tso-alerts.yml`. Allocation
+errors warn above 1 percent for 10 minutes and become critical above 10 percent
+for 5 minutes. The expressions require at least 0.1 reserve attempts/second so
+an idle cluster does not alert on an empty denominator.
+
+## Write-fanout benchmark
+
+Run the modeled remote-TSO benchmark with:
+
+```sh
+go test ./kv -run '^$' -bench '^BenchmarkTSOWriteFanout$' -benchmem -benchtime=1s -count=3
+```
+
+The harness uses 16 concurrent write coordinators sharing a `BatchAllocator`.
+Each operation stamps 1, 3, or 8 fan-out legs, and each refill pays a modeled
+1 ms remote-leader plus Raft-commit delay. The following medians were measured
+on Go 1.26.5, darwin/arm64, Apple M1 Max:
+
+| Fan-out | Batch | Wall time/op | p99 | Refills/op | Timestamp writes/s |
+|---:|---:|---:|---:|---:|---:|
+| 1 | 1 | 1.33 ms | 1,249 ms | 1.000 | 753 |
+| 1 | 64 | 24.81 us | 1.323 ms | 0.01563 | 40,309 |
+| 1 | 256 | 8.92 us | 0.000292 ms | 0.00391 | 112,081 |
+| 3 | 1 | 5.70 ms | 1,233 ms | 3.000 | 526 |
+| 3 | 64 | 105.35 us | 3.212 ms | 0.04690 | 28,478 |
+| 3 | 256 | 17.30 us | 1.272 ms | 0.01172 | 173,453 |
+| 8 | 1 | 10.43 ms | 1,104 ms | 8.000 | 767 |
+| 8 | 64 | 157.49 us | 1.316 ms | 0.1251 | 50,796 |
+| 8 | 256 | 39.43 us | 1.302 ms | 0.03127 | 202,870 |
+
+Batch size 1 is intentionally retained as the unamortized baseline and shows
+severe waiter tails under contention. The production default of 256 keeps the
+modeled fan-out p99 below 4 ms in this harness, including the refill-contention
+tail at fan-out 8, and below the 50 ms warning threshold. These are controlled
+local measurements, not a substitute for observing the production histograms
+during rollout.
+
+## Intentional non-goals
+
+This closure does not automate cluster-wide phase orchestration, choose a
+dedicated subset of TSO members, or change the HLC ceiling formula. Operators
+still advance the shared mode file deployment-by-deployment, and the existing
+group membership and clock-floor decisions remain independent future work.
diff --git a/docs/design/2026_04_16_partial_centralized_tso.md b/docs/design/2026_04_16_implemented_centralized_tso.md
similarity index 90%
rename from docs/design/2026_04_16_partial_centralized_tso.md
rename to docs/design/2026_04_16_implemented_centralized_tso.md
index 979f4dec5..6e3c0c60e 100644
--- a/docs/design/2026_04_16_partial_centralized_tso.md
+++ b/docs/design/2026_04_16_implemented_centralized_tso.md
@@ -1,13 +1,16 @@
 # Centralized Timestamp Oracle (TSO) Design
 
-- Status: Partial — M1-M7 are implemented, including the dedicated group-0
-  FSM, leader-routed durable windows, strict term bootstrap, serialized shadow
-  migration, one-way rolling cutover, durable Phase-D retirement, and
-  cross-shard SSI timestamp validation. Runtime config reload and production
-  latency/alerting work remain open.
-- Author: bootjp
-- Date: 2026-04-16
-- Updated: 2026-07-19
+Status: Implemented
+Author: bootjp
+Date: 2026-04-16
+Updated: 2026-07-19
+
+M1-M8 implement the central subsystem: the dedicated group-0 FSM,
+leader-routed durable windows, strict term bootstrap, serialized shadow
+migration, one-way cutover and Phase-D retirement, validated cross-shard
+timestamps, one-way runtime mode reload, production metrics and alerts, and
+write-fanout benchmark evidence. The topology and clock-policy extensions in
+Section 9 are intentional non-goals and do not leave central scope incomplete.
 
 ---
 
@@ -137,11 +140,20 @@ Implemented:
 19. `BatchAllocator` validates cached candidates once Phase D is required. A
     candidate at/below the Phase-D floor invalidates the entire local window and
     forces a refill above the marker before any timestamp is returned.
-
-Remaining:
-
-1. Runtime config reload for the mode switch; current flags are startup-only.
-2. Production benchmark, divergence metrics, and alert thresholds.
+20. `--tsoModeFile` polls an atomically replaced `legacy`, `shadow`, `cutover`,
+    or `phase-d` mode. Live transitions must be adjacent and one-way. Invalid
+    files, skipped phases, and process-local rollbacks leave the active allocator
+    unchanged. Applied group-0 markers override a stale local mode immediately
+    on allocator resolution, before the next poll can run.
+21. The production registry exports reserve/validation latency by local or
+    remote path and outcome, shadow comparison/divergence, process mode, every
+    reload outcome, and durable marker state. Checked Prometheus rules define
+    warning and critical latency/error thresholds, persistent reload failure,
+    mode divergence, and the 15-minute zero-overlap cutover gate.
+22. `BenchmarkTSOWriteFanout` models 16 concurrent coordinators, 1/3/8-way
+    writes, a 1 ms remote TSO commit, and batch sizes 1/64/256. Three-run
+    evidence supports the existing default batch size 256 and is recorded in
+    `docs/centralized_tso_operations.md`.
 
 ### 1.1 Original Limitation
 
@@ -748,8 +760,10 @@ approach enables a live cutover.
 
 ### 7.3 Phase C — Durable TSO Cutover
 
-- The startup flag `--tsoEnabled` switches coordinator issuance to the
-  leader-routed allocator. Runtime config reload remains future work.
+- The backward-compatible startup flag `--tsoEnabled`, or runtime mode file
+  value `cutover`, switches coordinator issuance to the leader-routed allocator.
+  A live transition is accepted only after `shadow`; restart recovery may start
+  directly in cutover when the durable marker already exists.
 - The first production refill commits the group-0 cutover marker before
   committing and returning its timestamp window.
 - The marker is encoded in a versioned TSO-specific envelope whose leading
@@ -765,12 +779,13 @@ approach enables a live cutover.
 
 - Roll every member to a binary that understands the Phase-D entry and
   `ValidateTimestamp` RPC. Run all members on the dedicated TSO path before
-  enabling `--tsoPhaseDEnabled`; an older member would correctly halt on the
-  unknown control entry, so mixed-version activation is prohibited.
-- The first allocation with `--tsoPhaseDEnabled` commits cutover (if needed),
-  then the Phase-D marker and its pre-Phase-D floor, then a new allocation
-  window strictly above that floor. The switch is one-way and survives restart
-  through the TSO V4 snapshot.
+  enabling `--tsoPhaseDEnabled` or reloading `phase-d`; an older member would
+  correctly halt on the unknown control entry, so mixed-version activation is
+  prohibited.
+- The first allocation with Phase D requested commits cutover (if needed), then
+  the Phase-D marker and its pre-Phase-D floor, then a new allocation window
+  strictly above that floor. The switch is one-way and survives restart through
+  the TSO V4 snapshot.
 - `ShardedCoordinator` dynamically stops data-group ceiling proposals and
   fails closed instead of issuing from its legacy HLC. It continues group-0
   renewal only. Shadow mode observes durable cutover and directly uses the
@@ -812,6 +827,29 @@ candidate is returned. Once the cutover marker applies, shadow callers stop
 returning legacy candidates. Independently, every new TSO leader term fences
 its first window above the strict maximum committed data timestamp.
 
+### 7.6 Runtime Reload and Operational Gates
+
+`DynamicTimestampAllocator` publishes the process-selected allocator through
+an atomic pointer. Before resolving that pointer, it checks the consensus-owned
+cutover and Phase-D markers. An applied marker selects the batch/routed TSO path
+and enables matching remote confirmation immediately, so a stale `legacy` or
+`shadow` file cannot mint a legacy timestamp while waiting for the next poll.
+A request that resolved its preceding mode before local marker apply may finish,
+but every subsequent resolution observes the one-way durable override.
+
+`TSORuntimeController` accepts only adjacent one-way live transitions:
+
+```text
+legacy -> shadow -> cutover -> phase-d
+```
+
+The initial process mode may be later in the sequence for restart recovery.
+No-op polls refresh the durable-state gauges, and every failed poll increments
+its bounded reload-result counter even when duplicate log messages are
+suppressed. Exact rollout gates, alert expressions, atomic file replacement,
+and benchmark evidence are maintained in
+`docs/centralized_tso_operations.md`.
+
 ---
 
 ## 8. Milestones
@@ -825,20 +863,24 @@ its first window above the strict maximum committed data timestamp.
 | M5 — shipped | Preserve the default-group `LocalTSOAllocator` compatibility bridge when group 0 is absent; route coordinator-owned timestamp call sites through the allocator abstraction. | Medium |
 | M6 — shipped | Run the dedicated group-0 FSM, fence each new TSO leader term above all authoritative data-group commit floors, redirect follower requests to the TSO leader over gRPC, synchronously serialize fail-closed shadow issuance, and commit the one-way rolling cutover marker before production windows. | Low |
 | M7 — shipped | Commit the durable Phase-D floor marker, preserve V3 snapshots until activation, retire data-shard HLC renewal and legacy/shadow issuance after cutover, activate before read validation while preserving exact applied snapshots through timestamp-bound per-dispatch vouchers, invalidate pre-Phase-D batch windows, and validate unvouched caller-supplied cross-shard SSI timestamps at the group-0 leader from activation onward. | Low |
+| M8 — shipped | Atomically reload one-way runtime modes, make durable markers override stale local mode before allocation, expose production latency/divergence/state metrics, install checked alert thresholds, and validate the default batch size with concurrent write-fanout evidence. | Low |
 
 ---
 
-## 9. Open Questions
+## 9. Resolved Questions and Future Extensions
 
-1. **TSO RTT when TSO leader ≠ write leader:** What batch size minimises tail
-   latency? Needs benchmarking against realistic write fan-out.
+1. **TSO RTT when TSO leader differs from the write leader (resolved):** The
+   concurrent benchmark covers 1/3/8 fan-out legs with a modeled 1 ms remote
+   TSO commit. Batch size 1 exposes serialized refill tails, while default 256
+   keeps the recorded p99 below the 50 ms warning threshold.
 
-2. **TSO group membership:** Should all cluster nodes join the TSO group, or
-   should a dedicated subset (e.g. 3 out of N) be used to reduce Raft traffic?
+2. **TSO group membership (intentional future extension):** Choosing all nodes
+   or a dedicated subset is a topology policy outside the central timestamp
+   subsystem. The implemented routing and fail-closed term fence support either.
 
-3. **Clock floor semantics:** `max(now, ceiling)` vs. `ceiling + 1` — the
-   stricter form (`ceiling + 1`) guarantees no overlap even if wall clocks
-   drift, at the cost of one extra millisecond per renewal window.
+3. **Clock floor semantics (intentional future extension):** The implementation
+   retains the existing floor formula. Changing to `ceiling + 1` is an
+   independent HLC policy decision, not unfinished centralized-TSO scope.
 
 4. **Non-leader TSO requests (resolved):** Followers redirect to the current
    group-0 leader through `Distribution.GetTimestamp`. Timestamp allocation is
diff --git a/docs/design/2026_05_28_implemented_tla_safety_spec.md b/docs/design/2026_05_28_implemented_tla_safety_spec.md
index cd94bb5d0..265e50f3d 100644
--- a/docs/design/2026_05_28_implemented_tla_safety_spec.md
+++ b/docs/design/2026_05_28_implemented_tla_safety_spec.md
@@ -661,12 +661,11 @@ does not keep this document in `partial`.
    mitigation, the doc lifecycle (Section 8.2) ties promotion to specs
    landing, so a stale `partial` status is a visible signal.
 
-5. **Choice of TSO model.** The HLC spec models the current
-   per-shard-leader ceiling. The centralized TSO proposal
-   (`2026_04_16_partial_centralized_tso.md`) would change that. The two
-   docs are independent; if/when centralized TSO lands, `HLC.tla` (or a
-   sibling `TSO.tla`) is updated as part of that PR. We do **not** block
-   this proposal on the TSO decision.
+5. **Choice of TSO model.** The HLC spec continues to model the compatibility
+   per-shard-leader ceiling, while the implemented centralized TSO
+   (`2026_04_16_implemented_centralized_tso.md`) owns the production ordering
+   path. A sibling `TSO.tla` remains an independent future extension; runtime
+   transition safety is pinned by the TSO FSM, routing, reload, and race tests.
 
 6. **TLC tool licensing and binary distribution.** `tla2tools.jar` is
    MIT-licensed but not in any package manager we already depend on.
@@ -716,8 +715,8 @@ does not keep this document in `partial`.
 - `distribution/engine.go`, `distribution/catalog.go`,
   `distribution/watcher.go` — route catalog.
 - `docs/architecture_overview.md` — system-level diagrams.
-- `docs/design/2026_04_16_partial_centralized_tso.md` — the TSO proposal
-  that this spec is independent of (Section 9, risk 5).
+- `docs/design/2026_04_16_implemented_centralized_tso.md` — the implemented TSO
+  design that this spec is independent of (Section 9, risk 5).
 - Diego Ongaro's Raft TLA+ specification — reference for the abstract
   `Raft.tla` interface.
 - CockroachDB and TiDB MVCC / HLC TLA+ models — public prior art for
diff --git a/docs/design/2026_06_23_proposed_scaling_roadmap.md b/docs/design/2026_06_23_proposed_scaling_roadmap.md
index fe74eda05..4d92db147 100644
--- a/docs/design/2026_06_23_proposed_scaling_roadmap.md
+++ b/docs/design/2026_06_23_proposed_scaling_roadmap.md
@@ -202,38 +202,18 @@ memory each group's private cache/memtable pins.
 
 ### (d) Cross-group transactions at scale
 
-- **Per-group HLC vs centralized TSO.** Today the ceiling is renewed only on
-  the default group (§1.5); the TSO doc
-  (`docs/design/2026_04_16_partial_centralized_tso.md`) has shipped the
-  near-term fix (renew on all led groups, in parallel — its M1) and still
-  leaves the dedicated TSO Raft group with a batch allocator open.
-  **Assessment of whether TSO is still needed once groups multiply:** the
-  near-term per-group fix (TSO doc §6) was *necessary regardless* — it is the
-  minimum correctness fix the moment a node leads a non-default group it is
-  member of, and it has landed before multi-node multi-group bootstrap (b)
-  makes that topology reachable. The *full* dedicated-TSO component (TSO doc
-  M6–M7) is a larger component, but the need for a shared ordering source is
-  not a throughput/amortization question once cross-node cross-group
-  transactions are possible: with the per-group fix in place, every node's
-  timestamps are self-monotonic, but **global** monotonicity across coordinator
-  nodes is still not guaranteed without a shared oracle (TSO doc §6
-  "Guarantee"). Cross-group transactions (`kv/transaction.go`,
-  `kv/txn_codec.go`) whose timestamps can be allocated by different ingress /
-  coordinator nodes need a single ordering source for OCC commit-ts
-  comparability, regardless of where the participating groups' current leaders
-  live. `LeaderProxy.Commit` / `Internal.Forward` preserve non-zero timestamps,
-  so leader co-location does not prove single-clock allocation. (The shared-snapshot
-  invariant — every operation in one txn reading at the *same* `startTS` — is
-  already upheld: `nextStartTS` allocates one `startTS` for the whole txn and
-  propagates it via `reqs.StartTS` to every participating group; the gap is the
-  cross-*coordinator* comparability of the per-txn `commitTS`, not the per-txn
-  `startTS`. See OQ-1.) So: per-group renewal fix is in-scope-soon and
-  load-bearing for one node leading multiple groups; before enabling any
-  cross-group transaction mode in which more than one coordinator node can issue
-  `startTS` / `commitTS`, the roadmap must either pull the dedicated TSO group
-  forward or land a narrower single-oracle bridge that pins cross-group
-  timestamp allocation to one designated leader. Until such txns are enabled,
-  the per-group fix plus the shared-`*HLC` property remains adequate.
+- **Per-group HLC vs centralized TSO.** The centralized TSO design
+  (`docs/design/2026_04_16_implemented_centralized_tso.md`) has shipped M1-M8:
+  all-led-group compatibility renewal, the dedicated group-0 FSM, leader-routed
+  durable windows, one-way cutover and Phase D, runtime mode reload, and
+  operational gates. The shared ordering source is available for cross-node
+  cross-group transactions. Deployments that remain in `legacy` still have
+  only per-node monotonicity; before enabling multiple coordinator nodes for
+  cross-group issuance, they must complete the documented `shadow -> cutover`
+  sequence. `LeaderProxy.Commit` / `Internal.Forward` preserve non-zero
+  timestamps, and one `startTS` remains shared by every transaction participant.
+  Phase D additionally validates a caller-supplied cross-shard `StartTS` at the
+  group-0 leader before commit allocation.
 
 ### (e) Cluster size / membership
 
diff --git a/kv/coordinator.go b/kv/coordinator.go
index af91a09fd..eda72056f 100644
--- a/kv/coordinator.go
+++ b/kv/coordinator.go
@@ -1012,11 +1012,11 @@ func (c *Coordinate) allocateTimestampAfter(ctx context.Context, label string, m
 	if min == ^uint64(0) {
 		return 0, errors.Wrap(ErrTxnCommitTSRequired, label)
 	}
-	if c.tsAllocator != nil {
+	if allocator, ok := resolveTimestampAllocator(c.tsAllocator); ok {
 		if min > 0 {
-			return nextTimestampAfterFromAllocator(ctx, c.tsAllocator, min, label)
+			return nextTimestampAfterFromAllocator(ctx, allocator, min, label)
 		}
-		return nextTimestampFromAllocator(ctx, c.tsAllocator, label)
+		return nextTimestampFromAllocator(ctx, allocator, label)
 	}
 	if c.clock == nil {
 		return 0, errors.Wrap(ErrTSOClockNil, label)
diff --git a/kv/keyviz_label.go b/kv/keyviz_label.go
index ba40e30d5..09a3629c3 100644
--- a/kv/keyviz_label.go
+++ b/kv/keyviz_label.go
@@ -91,7 +91,7 @@ func (c keyVizLabeledCoordinator) RaftLeaderForKey(key []byte) string {
 func (c keyVizLabeledCoordinator) Clock() *HLC { return c.inner.Clock() }
 
 func (c keyVizLabeledCoordinator) TimestampAllocator() TimestampAllocator {
-	alloc, _ := TimestampAllocatorThrough(c.inner)
+	alloc, _ := ConfiguredTimestampAllocatorThrough(c.inner)
 	return alloc
 }
 
diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go
index 3705fcf4c..4a54f3b3e 100644
--- a/kv/sharded_coordinator.go
+++ b/kv/sharded_coordinator.go
@@ -1629,11 +1629,11 @@ func (c *ShardedCoordinator) allocateTimestampAfter(ctx context.Context, label s
 	if min == ^uint64(0) {
 		return 0, errors.Wrap(ErrTxnCommitTSRequired, label)
 	}
-	if c.tsAllocator != nil {
+	if allocator, ok := resolveTimestampAllocator(c.tsAllocator); ok {
 		if min > 0 {
-			return nextTimestampAfterFromAllocator(ctx, c.tsAllocator, min, label)
+			return nextTimestampAfterFromAllocator(ctx, allocator, min, label)
 		}
-		return nextTimestampFromAllocator(ctx, c.tsAllocator, label)
+		return nextTimestampFromAllocator(ctx, allocator, label)
 	}
 	if c.tsoCutoverState != nil && c.tsoCutoverState.PhaseDActive() {
 		return 0, errors.Wrap(ErrTSOAllocatorRequired,
@@ -1663,7 +1663,8 @@ func (c *ShardedCoordinator) NextAfter(ctx context.Context, min uint64) (uint64,
 }
 
 func (c *ShardedCoordinator) nextTxnTSAfter(ctx context.Context, startTS uint64) (uint64, error) {
-	if c.clock == nil && c.tsAllocator == nil {
+	_, allocatorConfigured := resolveTimestampAllocator(c.tsAllocator)
+	if c.clock == nil && !allocatorConfigured {
 		nextTS := startTS + 1
 		if nextTS == 0 {
 			return 0, nil
@@ -1725,7 +1726,8 @@ func (c *ShardedCoordinator) nextStartTS(ctx context.Context, elems []*Elem[OP])
 	if c.clock != nil && maxTS > 0 {
 		c.clock.Observe(maxTS)
 	}
-	if c.clock == nil && c.tsAllocator == nil {
+	_, allocatorConfigured := resolveTimestampAllocator(c.tsAllocator)
+	if c.clock == nil && !allocatorConfigured {
 		return maxTS + 1, nil
 	}
 	ts, err := c.allocateTimestampAfter(ctx, "allocate sharded startTS", maxTS)
@@ -2195,7 +2197,7 @@ func (c *ShardedCoordinator) rawLogs(ctx context.Context, reqs *OperationGroup[O
 }
 
 func (c *ShardedCoordinator) rawLogTimestamp(ctx context.Context) (uint64, error) {
-	if c.tsAllocator != nil {
+	if _, ok := resolveTimestampAllocator(c.tsAllocator); ok {
 		return 0, nil
 	}
 	return c.allocateTimestamp(ctx, "allocate sharded raw log ts")
diff --git a/kv/tso.go b/kv/tso.go
index 2031122d1..0f0be8181 100644
--- a/kv/tso.go
+++ b/kv/tso.go
@@ -252,6 +252,21 @@ func ConfiguredTimestampAllocatorThrough(coord Coordinator) (TimestampAllocator,
 	}
 }
 
+type currentTimestampAllocatorProvider interface {
+	currentTimestampAllocator() TimestampAllocator
+}
+
+func resolveTimestampAllocator(alloc TimestampAllocator) (TimestampAllocator, bool) {
+	for range 4 {
+		provider, ok := alloc.(currentTimestampAllocatorProvider)
+		if !ok {
+			return alloc, alloc != nil
+		}
+		alloc = provider.currentTimestampAllocator()
+	}
+	return alloc, alloc != nil
+}
+
 func coordinatorTimestampAllocator(coord Coordinator) (TimestampAllocator, bool) {
 	return TimestampAllocatorThrough(coord)
 }
diff --git a/kv/tso_fanout_benchmark_test.go b/kv/tso_fanout_benchmark_test.go
new file mode 100644
index 000000000..00e759c7a
--- /dev/null
+++ b/kv/tso_fanout_benchmark_test.go
@@ -0,0 +1,140 @@
+package kv
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// BenchmarkTSOWriteFanout models 16 concurrent write coordinators sharing one
+// process-local BatchAllocator. Each logical operation stamps every fan-out
+// leg, while a refill pays a 1 ms remote leader plus Raft commit delay.
+func BenchmarkTSOWriteFanout(b *testing.B) {
+	const (
+		writers = 16
+		tsoRTT  = time.Millisecond
+	)
+	for _, fanout := range []int{1, 3, 8} {
+		for _, batchSize := range []int{1, 64, 256} {
+			name := fmt.Sprintf("writers=%d/fanout=%d/batch=%d/rtt=1ms", writers, fanout, batchSize)
+			b.Run(name, func(b *testing.B) {
+				runTSOWriteFanoutBenchmark(b, writers, fanout, batchSize, tsoRTT)
+			})
+		}
+	}
+}
+
+func runTSOWriteFanoutBenchmark(b *testing.B, writers, fanout, batchSize int, rtt time.Duration) {
+	backend := &benchmarkTSOAllocator{rtt: rtt}
+	allocator, err := NewBatchAllocator(backend, batchSize)
+	if err != nil {
+		b.Fatal(err)
+	}
+	latencies := make([]int64, b.N)
+	var next atomic.Uint64
+	var firstErr atomic.Pointer[benchmarkTSOError]
+	ctx := context.Background()
+	b.ResetTimer()
+	started := time.Now()
+	var wg sync.WaitGroup
+	for range writers {
+		wg.Add(1)
+		go runTSOWriteFanoutWorker(ctx, allocator, fanout, latencies, &next, &firstErr, &wg)
+	}
+	wg.Wait()
+	elapsed := time.Since(started)
+	b.StopTimer()
+	reportTSOWriteFanoutMetrics(b, fanout, latencies, backend, elapsed, firstErr.Load())
+}
+
+func runTSOWriteFanoutWorker(
+	ctx context.Context,
+	allocator TimestampAllocator,
+	fanout int,
+	latencies []int64,
+	next *atomic.Uint64,
+	firstErr *atomic.Pointer[benchmarkTSOError],
+	wg *sync.WaitGroup,
+) {
+	defer wg.Done()
+	for {
+		index := next.Add(1) - 1
+		if index >= uint64(len(latencies)) { //nolint:gosec // benchmark sizes are non-negative.
+			return
+		}
+		opStarted := time.Now()
+		for range fanout {
+			if _, err := allocator.Next(ctx); err != nil {
+				firstErr.CompareAndSwap(nil, &benchmarkTSOError{err: err})
+				return
+			}
+		}
+		latencies[index] = time.Since(opStarted).Nanoseconds()
+	}
+}
+
+func reportTSOWriteFanoutMetrics(
+	b *testing.B,
+	fanout int,
+	latencies []int64,
+	backend *benchmarkTSOAllocator,
+	elapsed time.Duration,
+	recorded *benchmarkTSOError,
+) {
+	if recorded != nil {
+		b.Fatal(recorded.err)
+	}
+	sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })
+	b.ReportMetric(percentileDuration(latencies, 50).Seconds()*1000, "p50-ms")
+	b.ReportMetric(percentileDuration(latencies, 99).Seconds()*1000, "p99-ms")
+	b.ReportMetric(float64(backend.refills.Load())/float64(b.N), "refills/op")
+	b.ReportMetric(float64(b.N*fanout)/elapsed.Seconds(), "writes/s")
+}
+
+func percentileDuration(sorted []int64, percentile int) time.Duration {
+	if len(sorted) == 0 {
+		return 0
+	}
+	index := (len(sorted)*percentile + 99) / 100
+	if index > 0 {
+		index--
+	}
+	return time.Duration(sorted[index])
+}
+
+type benchmarkTSOError struct {
+	err error
+}
+
+type benchmarkTSOAllocator struct {
+	rtt     time.Duration
+	next    atomic.Uint64
+	refills atomic.Uint64
+}
+
+func (a *benchmarkTSOAllocator) Next(ctx context.Context) (uint64, error) {
+	return a.NextBatch(ctx, 1)
+}
+
+func (a *benchmarkTSOAllocator) NextBatch(ctx context.Context, n int) (uint64, error) {
+	timer := time.NewTimer(a.rtt)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	case <-timer.C:
+	}
+	a.refills.Add(1)
+	end := a.next.Add(uint64(n))    //nolint:gosec // benchmark batch sizes are positive.
+	return end - uint64(n) + 1, nil //nolint:gosec // benchmark batch sizes are positive.
+}
+
+func (a *benchmarkTSOAllocator) IsLeader() bool { return false }
+
+func (a *benchmarkTSOAllocator) RunLeaseRenewal(ctx context.Context) {
+	<-ctx.Done()
+}
diff --git a/kv/tso_raft.go b/kv/tso_raft.go
index 3975f5733..63f06d637 100644
--- a/kv/tso_raft.go
+++ b/kv/tso_raft.go
@@ -5,6 +5,7 @@ import (
 	stderrors "errors"
 	"log/slog"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/bootjp/elastickv/internal/raftengine"
@@ -389,29 +390,36 @@ type LeaderRoutedTSOAllocator struct {
 	remoteRequest  tsoRemoteRequest
 	retryBudget    time.Duration
 	retryInterval  time.Duration
-	activate       bool
-	activatePhaseD bool
+	activation     atomic.Uint32
 	clock          *HLC
 	remoteValidate tsoRemoteValidation
+	observer       TSOObserver
 }
 
+const (
+	tsoActivationInactive uint32 = iota
+	tsoActivationCutover
+	tsoActivationPhaseD
+)
+
 type LeaderRoutedTSOAllocatorOption func(*LeaderRoutedTSOAllocator)
 
 func WithTSOCutoverActivation() LeaderRoutedTSOAllocatorOption {
-	return func(a *LeaderRoutedTSOAllocator) { a.activate = true }
+	return func(a *LeaderRoutedTSOAllocator) { a.setActivation(true, false) }
 }
 
 func WithTSOPhaseDActivation() LeaderRoutedTSOAllocatorOption {
-	return func(a *LeaderRoutedTSOAllocator) {
-		a.activate = true
-		a.activatePhaseD = true
-	}
+	return func(a *LeaderRoutedTSOAllocator) { a.setActivation(true, true) }
 }
 
 func WithTSORoutedClock(clock *HLC) LeaderRoutedTSOAllocatorOption {
 	return func(a *LeaderRoutedTSOAllocator) { a.clock = clock }
 }
 
+func WithTSOObserver(observer TSOObserver) LeaderRoutedTSOAllocatorOption {
+	return func(a *LeaderRoutedTSOAllocator) { a.observer = observer }
+}
+
 func NewLeaderRoutedTSOAllocator(
 	local TSOAllocator,
 	leader raftengine.LeaderView,
@@ -459,7 +467,8 @@ func (a *LeaderRoutedTSOAllocator) NextBatchAfter(ctx context.Context, n int, mi
 }
 
 func (a *LeaderRoutedTSOAllocator) nextBatchAfter(ctx context.Context, n int, min uint64) (uint64, error) {
-	reservation, err := a.nextReservation(ctx, n, min, a.activate, a.activatePhaseD, a.activate)
+	activate, activatePhaseD := a.activationState()
+	reservation, err := a.nextReservation(ctx, n, min, activate, activatePhaseD, activate)
 	if err != nil {
 		return 0, err
 	}
@@ -524,19 +533,28 @@ func (a *LeaderRoutedTSOAllocator) tryBatch(
 ) (TSOReservation, error) {
 	var empty TSOReservation
 	if a.local.IsLeader() {
-		return a.tryLocalBatch(ctx, n, min, activate, activatePhaseD, requireReservation)
+		started := time.Now()
+		reservation, err := a.tryLocalBatch(ctx, n, min, activate, activatePhaseD, requireReservation)
+		a.observeRequest("reserve", "local", err, time.Since(started))
+		return reservation, err
 	}
 	addr := leaderAddrFromEngine(a.leader)
 	if addr == "" {
-		return empty, errors.WithStack(ErrLeaderNotFound)
+		err := errors.WithStack(ErrLeaderNotFound)
+		a.observeRequest("reserve", "remote", err, 0)
+		return empty, err
 	}
+	started := time.Now()
 	reservation, err := a.remoteRequest(ctx, addr, n, min, activate, activatePhaseD)
 	if err != nil {
+		a.observeRequest("reserve", "remote", err, time.Since(started))
 		return empty, err
 	}
 	if err := validateTSOReservation(reservation, n, min, requireReservation); err != nil {
+		a.observeRequest("reserve", "remote", err, time.Since(started))
 		return empty, err
 	}
+	a.observeRequest("reserve", "remote", nil, time.Since(started))
 	return reservation, nil
 }
 
@@ -629,13 +647,18 @@ func (a *LeaderRoutedTSOAllocator) ValidateDurableTimestamp(ctx context.Context,
 			if !ok {
 				return errors.WithStack(ErrTSOProtocolUnsupported)
 			}
+			started := time.Now()
 			lastErr = errors.WithStack(validator.ValidateDurableTimestamp(ctx, timestamp))
+			a.observeRequest("validate", "local", lastErr, time.Since(started))
 		} else {
 			addr := leaderAddrFromEngine(a.leader)
 			if addr == "" {
 				lastErr = errors.WithStack(ErrLeaderNotFound)
+				a.observeRequest("validate", "remote", lastErr, 0)
 			} else {
+				started := time.Now()
 				lastErr = a.remoteValidate(ctx, addr, timestamp)
+				a.observeRequest("validate", "remote", lastErr, time.Since(started))
 			}
 		}
 		if lastErr == nil {
@@ -679,7 +702,8 @@ func (a *LeaderRoutedTSOAllocator) PhaseDActive() bool {
 }
 
 func (a *LeaderRoutedTSOAllocator) PhaseDRequired() bool {
-	return a != nil && (a.activatePhaseD || a.PhaseDActive())
+	_, activatePhaseD := a.activationState()
+	return a != nil && (activatePhaseD || a.PhaseDActive())
 }
 
 func (a *LeaderRoutedTSOAllocator) observeReservation(reservation TSOReservation) {
@@ -688,6 +712,65 @@ func (a *LeaderRoutedTSOAllocator) observeReservation(reservation TSOReservation
 	}
 	end := reservation.Base + uint64(reservation.Count) - 1 //nolint:gosec // validated reservation.
 	a.clock.Observe(end)
+	if a.observer != nil {
+		a.observer.ObserveTSODurableState(reservation.CutoverActive, reservation.PhaseDActive)
+	}
+}
+
+// setActivation atomically publishes the durable markers future reservations
+// must request. In-flight reservations may complete under the prior mode; all
+// such modes still use group 0, and committed markers remain one-way.
+func (a *LeaderRoutedTSOAllocator) setActivation(cutover, phaseD bool) {
+	if a == nil {
+		return
+	}
+	state := tsoActivationInactive
+	if cutover || phaseD {
+		state = tsoActivationCutover
+	}
+	if phaseD {
+		state = tsoActivationPhaseD
+	}
+	a.activation.Store(state)
+}
+
+// promoteActivation advances the requested durable marker without allowing a
+// concurrent observer to lower an already-requested Phase D activation.
+func (a *LeaderRoutedTSOAllocator) promoteActivation(cutover, phaseD bool) {
+	if a == nil {
+		return
+	}
+	target := tsoActivationInactive
+	if cutover || phaseD {
+		target = tsoActivationCutover
+	}
+	if phaseD {
+		target = tsoActivationPhaseD
+	}
+	for current := a.activation.Load(); current < target; current = a.activation.Load() {
+		if a.activation.CompareAndSwap(current, target) {
+			return
+		}
+	}
+}
+
+func (a *LeaderRoutedTSOAllocator) activationState() (bool, bool) {
+	if a == nil {
+		return false, false
+	}
+	state := a.activation.Load()
+	return state >= tsoActivationCutover, state >= tsoActivationPhaseD
+}
+
+func (a *LeaderRoutedTSOAllocator) observeRequest(operation, path string, err error, duration time.Duration) {
+	if a == nil || a.observer == nil {
+		return
+	}
+	outcome := "success"
+	if err != nil {
+		outcome = "error"
+	}
+	a.observer.ObserveTSORequest(operation, path, outcome, duration)
 }
 
 func validateRoutedTSOWindow(base uint64, n int, min uint64) error {
@@ -794,10 +877,11 @@ func nonNilTSOContext(ctx context.Context) context.Context {
 // and retried; once the durable cutover marker is active, the allocator returns
 // the reserved TSO timestamp directly so rolling restarts cannot mix sources.
 type ShadowTimestampAllocator struct {
-	legacy  *HLC
-	shadow  TSOShadowReservationAllocator
-	cutover tsoCutoverState
-	log     *slog.Logger
+	legacy   *HLC
+	shadow   TSOShadowReservationAllocator
+	cutover  tsoCutoverState
+	log      *slog.Logger
+	observer TSOObserver
 }
 
 type ShadowTimestampAllocatorOption func(*ShadowTimestampAllocator)
@@ -806,6 +890,10 @@ func WithTSOShadowCutoverState(state tsoCutoverState) ShadowTimestampAllocatorOp
 	return func(a *ShadowTimestampAllocator) { a.cutover = state }
 }
 
+func WithTSOShadowObserver(observer TSOObserver) ShadowTimestampAllocatorOption {
+	return func(a *ShadowTimestampAllocator) { a.observer = observer }
+}
+
 func NewShadowTimestampAllocator(
 	legacy *HLC,
 	shadow TSOShadowReservationAllocator,
@@ -848,6 +936,7 @@ func (a *ShadowTimestampAllocator) NextAfter(ctx context.Context, min uint64) (u
 func (a *ShadowTimestampAllocator) nextAfter(ctx context.Context, min uint64) (uint64, error) {
 	ctx = nonNilTSOContext(ctx)
 	if a.cutover != nil && a.cutover.CutoverActive() {
+		a.observeShadowComparison("cutover_active", 0)
 		return a.nextDedicatedAfter(ctx, min)
 	}
 	a.legacy.Observe(min)
@@ -861,6 +950,7 @@ func (a *ShadowTimestampAllocator) nextAfter(ctx context.Context, min uint64) (u
 		}
 		reservation, err := a.shadow.ValidateShadowTimestamp(ctx, legacyTS)
 		if err != nil {
+			a.observeShadowComparison("error", 0)
 			a.log.ErrorContext(ctx, "tso shadow allocation failed",
 				slog.Uint64("legacy_ts", legacyTS),
 				slog.Any("err", err),
@@ -869,11 +959,14 @@ func (a *ShadowTimestampAllocator) nextAfter(ctx context.Context, min uint64) (u
 		}
 		a.legacy.Observe(reservation.Base)
 		if reservation.CutoverActive {
+			a.observeShadowComparison("cutover_active", 0)
 			return reservation.Base, nil
 		}
 		if legacyTS > reservation.PreviousAllocationFloor {
+			a.observeShadowComparison("legacy_ahead", legacyTS-reservation.PreviousAllocationFloor)
 			return legacyTS, nil
 		}
+		a.observeShadowComparison("legacy_overlap", reservation.PreviousAllocationFloor-legacyTS)
 		a.log.WarnContext(ctx, "tso shadow discarded overlapping legacy timestamp",
 			slog.Uint64("legacy_ts", legacyTS),
 			slog.Uint64("previous_tso_floor", reservation.PreviousAllocationFloor),
@@ -911,3 +1004,9 @@ func (a *ShadowTimestampAllocator) PhaseDRequired() bool {
 func (a *ShadowTimestampAllocator) Close() error {
 	return nil
 }
+
+func (a *ShadowTimestampAllocator) observeShadowComparison(result string, divergence uint64) {
+	if a != nil && a.observer != nil {
+		a.observer.ObserveTSOShadowComparison(result, divergence)
+	}
+}
diff --git a/kv/tso_runtime.go b/kv/tso_runtime.go
new file mode 100644
index 000000000..fb341b724
--- /dev/null
+++ b/kv/tso_runtime.go
@@ -0,0 +1,453 @@
+package kv
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/cockroachdb/errors"
+)
+
+var (
+	ErrInvalidTSOMode          = errors.New("tso: invalid runtime mode")
+	ErrTSOModeRollback         = errors.New("tso: runtime mode rollback is prohibited")
+	ErrUnsafeTSOModeTransition = errors.New("tso: runtime mode transition skips a required phase")
+)
+
+// TSOMode is the process-local timestamp issuance mode. Its ordering is part
+// of the migration contract: runtime reload may only move to the next value.
+type TSOMode uint32
+
+const (
+	TSOModeLegacy TSOMode = iota
+	TSOModeShadow
+	TSOModeCutover
+	TSOModePhaseD
+)
+
+func ParseTSOMode(raw string) (TSOMode, error) {
+	switch strings.ToLower(strings.TrimSpace(raw)) {
+	case "legacy":
+		return TSOModeLegacy, nil
+	case "shadow":
+		return TSOModeShadow, nil
+	case "cutover":
+		return TSOModeCutover, nil
+	case "phase-d", "phase_d", "phased":
+		return TSOModePhaseD, nil
+	default:
+		return TSOModeLegacy, errors.Wrapf(ErrInvalidTSOMode, "%q", strings.TrimSpace(raw))
+	}
+}
+
+func (m TSOMode) String() string {
+	switch m {
+	case TSOModeLegacy:
+		return "legacy"
+	case TSOModeShadow:
+		return "shadow"
+	case TSOModeCutover:
+		return "cutover"
+	case TSOModePhaseD:
+		return "phase-d"
+	default:
+		return "invalid"
+	}
+}
+
+func validTSOMode(mode TSOMode) bool {
+	return mode <= TSOModePhaseD
+}
+
+// TSOObserver is the bounded-cardinality operational surface for production
+// allocation latency, shadow divergence, durable state, and mode reloads.
+type TSOObserver interface {
+	ObserveTSORequest(operation, path, outcome string, duration time.Duration)
+	ObserveTSOShadowComparison(result string, divergence uint64)
+	ObserveTSOMode(mode string)
+	ObserveTSOModeReload(result string)
+	ObserveTSODurableState(cutoverActive, phaseDActive bool)
+}
+
+type timestampAllocatorSlot struct {
+	allocator TimestampAllocator
+}
+
+// DynamicTimestampAllocator atomically publishes an optional allocator. A nil
+// current allocator deliberately means "use the coordinator's legacy HLC
+// path"; callers must resolve it through TimestampAllocatorThrough rather than
+// calling Next directly.
+type DynamicTimestampAllocator struct {
+	current          atomic.Pointer[timestampAllocatorSlot]
+	durableState     TSORuntimeState
+	durableAllocator TimestampAllocator
+	routed           *LeaderRoutedTSOAllocator
+}
+
+func NewDynamicTimestampAllocator(initial TimestampAllocator) *DynamicTimestampAllocator {
+	d := &DynamicTimestampAllocator{}
+	d.store(initial)
+	return d
+}
+
+func (d *DynamicTimestampAllocator) store(allocator TimestampAllocator) {
+	if d == nil {
+		return
+	}
+	if allocator == nil {
+		d.current.Store(nil)
+		return
+	}
+	d.current.Store(&timestampAllocatorSlot{allocator: allocator})
+}
+
+func (d *DynamicTimestampAllocator) currentTimestampAllocator() TimestampAllocator {
+	if d == nil {
+		return nil
+	}
+	if allocator := d.durableTimestampAllocator(); allocator != nil {
+		return allocator
+	}
+	slot := d.current.Load()
+	if slot == nil {
+		return nil
+	}
+	return slot.allocator
+}
+
+// configureDurableOverride installs immutable references before the dynamic
+// allocator is published. An applied one-way marker must override a stale mode
+// file immediately rather than waiting for the next reload poll.
+func (d *DynamicTimestampAllocator) configureDurableOverride(
+	state TSORuntimeState,
+	allocator TimestampAllocator,
+	routed *LeaderRoutedTSOAllocator,
+) {
+	d.durableState = state
+	d.durableAllocator = allocator
+	d.routed = routed
+}
+
+func (d *DynamicTimestampAllocator) durableTimestampAllocator() TimestampAllocator {
+	if d.durableState == nil || d.durableAllocator == nil {
+		return nil
+	}
+	phaseD := d.durableState.PhaseDActive()
+	if !phaseD && !d.durableState.CutoverActive() {
+		return nil
+	}
+	d.routed.promoteActivation(true, phaseD)
+	return d.durableAllocator
+}
+
+func (d *DynamicTimestampAllocator) Next(ctx context.Context) (uint64, error) {
+	allocator := d.currentTimestampAllocator()
+	if allocator == nil {
+		return 0, errors.WithStack(ErrTSOAllocatorRequired)
+	}
+	timestamp, err := allocator.Next(ctx)
+	return timestamp, errors.Wrap(err, "dynamic tso next")
+}
+
+func (d *DynamicTimestampAllocator) NextAfter(ctx context.Context, min uint64) (uint64, error) {
+	allocator := d.currentTimestampAllocator()
+	if allocator == nil {
+		return 0, errors.WithStack(ErrTSOAllocatorRequired)
+	}
+	if after, ok := allocator.(TimestampAfterAllocator); ok {
+		timestamp, err := after.NextAfter(ctx, min)
+		return timestamp, errors.Wrap(err, "dynamic tso next after")
+	}
+	return nextTimestampAfterFromAllocator(ctx, allocator, min, "dynamic tso next after")
+}
+
+func (d *DynamicTimestampAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error {
+	allocator := d.currentTimestampAllocator()
+	if allocator == nil {
+		return errors.WithStack(ErrTSOAllocatorRequired)
+	}
+	validator, ok := allocator.(DurableTimestampValidator)
+	if !ok {
+		return errors.WithStack(ErrTSOProtocolUnsupported)
+	}
+	return errors.WithStack(validator.ValidateDurableTimestamp(ctx, timestamp))
+}
+
+func (d *DynamicTimestampAllocator) PhaseDActive() bool {
+	state, ok := d.currentTimestampAllocator().(TSOPhaseDState)
+	return ok && state.PhaseDActive()
+}
+
+func (d *DynamicTimestampAllocator) PhaseDRequired() bool {
+	state, ok := d.currentTimestampAllocator().(TSOPhaseDState)
+	return ok && state.PhaseDRequired()
+}
+
+func (d *DynamicTimestampAllocator) Invalidate() {
+	invalidateTimestampWindow(d.currentTimestampAllocator())
+}
+
+// TSORuntimeState is the consensus-owned one-way migration state.
+type TSORuntimeState interface {
+	CutoverActive() bool
+	PhaseDActive() bool
+}
+
+type TSORuntimeControllerConfig struct {
+	Clock       *HLC
+	Routed      *LeaderRoutedTSOAllocator
+	State       TSORuntimeState
+	BatchSize   int
+	InitialMode TSOMode
+	Logger      *slog.Logger
+	Observer    TSOObserver
+}
+
+// TSORuntimeController owns the process-local mode while treating group-0's
+// durable markers as authoritative. InitialMode may start at any phase for a
+// restart, but ApplyMode requires adjacent, one-way runtime transitions.
+type TSORuntimeController struct {
+	dynamic  *DynamicTimestampAllocator
+	shadow   *ShadowTimestampAllocator
+	batch    *BatchAllocator
+	routed   *LeaderRoutedTSOAllocator
+	state    TSORuntimeState
+	observer TSOObserver
+
+	mu   sync.Mutex
+	mode atomic.Uint32
+}
+
+func NewTSORuntimeController(cfg TSORuntimeControllerConfig) (*TSORuntimeController, error) {
+	if cfg.Clock == nil {
+		return nil, errors.WithStack(ErrTSOClockNil)
+	}
+	if cfg.Routed == nil {
+		return nil, errors.WithStack(ErrTSOAllocatorRequired)
+	}
+	if cfg.State == nil {
+		return nil, errors.WithStack(ErrTSOStateRequired)
+	}
+	if !validTSOMode(cfg.InitialMode) {
+		return nil, errors.Wrapf(ErrInvalidTSOMode, "%d", cfg.InitialMode)
+	}
+	if cfg.Logger == nil {
+		cfg.Logger = slog.Default()
+	}
+	shadow, err := NewShadowTimestampAllocator(
+		cfg.Clock,
+		cfg.Routed,
+		cfg.Logger,
+		WithTSOShadowCutoverState(cfg.State),
+		WithTSOShadowObserver(cfg.Observer),
+	)
+	if err != nil {
+		return nil, errors.Wrap(err, "configure runtime tso shadow allocator")
+	}
+	batch, err := NewBatchAllocator(cfg.Routed, cfg.BatchSize)
+	if err != nil {
+		return nil, errors.Wrap(err, "configure runtime tso batch allocator")
+	}
+	dynamic := NewDynamicTimestampAllocator(nil)
+	dynamic.configureDurableOverride(cfg.State, batch, cfg.Routed)
+	controller := &TSORuntimeController{
+		dynamic:  dynamic,
+		shadow:   shadow,
+		batch:    batch,
+		routed:   cfg.Routed,
+		state:    cfg.State,
+		observer: cfg.Observer,
+	}
+	controller.installMode(controller.effectiveMode(cfg.InitialMode))
+	return controller, nil
+}
+
+func (c *TSORuntimeController) Allocator() *DynamicTimestampAllocator {
+	if c == nil {
+		return nil
+	}
+	return c.dynamic
+}
+
+func (c *TSORuntimeController) CurrentMode() TSOMode {
+	if c == nil {
+		return TSOModeLegacy
+	}
+	return TSOMode(c.mode.Load())
+}
+
+func (c *TSORuntimeController) EffectiveMode(requested TSOMode) TSOMode {
+	if c == nil {
+		return requested
+	}
+	return c.effectiveMode(requested)
+}
+
+func (c *TSORuntimeController) effectiveMode(requested TSOMode) TSOMode {
+	if c.state != nil && c.state.PhaseDActive() {
+		return TSOModePhaseD
+	}
+	if c.state != nil && c.state.CutoverActive() && requested < TSOModeCutover {
+		return TSOModeCutover
+	}
+	return requested
+}
+
+func (c *TSORuntimeController) ApplyMode(requested TSOMode) error {
+	if c == nil {
+		return errors.WithStack(ErrTSOAllocatorRequired)
+	}
+	if !validTSOMode(requested) {
+		c.observeReload("rejected")
+		return errors.Wrapf(ErrInvalidTSOMode, "%d", requested)
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	target := c.effectiveMode(requested)
+	durableFloor := c.effectiveMode(TSOModeLegacy)
+	current := c.CurrentMode()
+	if target < current {
+		c.observeReload("rejected")
+		return errors.Wrapf(ErrTSOModeRollback, "%s -> %s", current, target)
+	}
+	if target == current {
+		c.observeDurableState()
+		return nil
+	}
+	if target > current+1 && target > durableFloor+1 {
+		c.observeReload("rejected")
+		return errors.Wrapf(ErrUnsafeTSOModeTransition, "%s -> %s", current, target)
+	}
+	c.installMode(target)
+	c.observeReload("applied")
+	return nil
+}
+
+func (c *TSORuntimeController) installMode(mode TSOMode) {
+	switch mode {
+	case TSOModeLegacy:
+		c.routed.setActivation(false, false)
+		c.dynamic.store(nil)
+	case TSOModeShadow:
+		c.routed.setActivation(false, false)
+		c.dynamic.store(c.shadow)
+	case TSOModeCutover:
+		c.routed.setActivation(true, false)
+		c.batch.Invalidate()
+		c.dynamic.store(c.batch)
+	case TSOModePhaseD:
+		c.routed.setActivation(true, true)
+		c.batch.Invalidate()
+		c.dynamic.store(c.batch)
+	}
+	c.mode.Store(uint32(mode))
+	if c.observer != nil {
+		c.observer.ObserveTSOMode(mode.String())
+	}
+	c.observeDurableState()
+}
+
+func (c *TSORuntimeController) observeReload(result string) {
+	if c != nil && c.observer != nil {
+		c.observer.ObserveTSOModeReload(result)
+	}
+}
+
+func (c *TSORuntimeController) observeDurableState() {
+	if c != nil && c.observer != nil && c.state != nil {
+		c.observer.ObserveTSODurableState(c.state.CutoverActive(), c.state.PhaseDActive())
+	}
+}
+
+func ReadTSOModeFile(path string) (TSOMode, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return TSOModeLegacy, errors.Wrap(err, "read tso mode file")
+	}
+	mode, err := ParseTSOMode(string(raw))
+	return mode, errors.Wrap(err, "parse tso mode file")
+}
+
+// RunTSOModeFileReload polls an atomically replaceable plain-text mode file.
+// Invalid reads or transitions leave the current allocator untouched.
+func RunTSOModeFileReload(
+	ctx context.Context,
+	path string,
+	interval time.Duration,
+	controller *TSORuntimeController,
+	logger *slog.Logger,
+) error {
+	if controller == nil {
+		return errors.WithStack(ErrTSOAllocatorRequired)
+	}
+	if strings.TrimSpace(path) == "" {
+		return nil
+	}
+	if interval <= 0 {
+		return errors.New("tso mode reload interval must be positive")
+	}
+	if logger == nil {
+		logger = slog.Default()
+	}
+	ctx = nonNilTSOContext(ctx)
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	lastError := ""
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			err := reloadTSOModeFile(path, controller)
+			lastError = reportTSOModeReloadError(ctx, path, lastError, err, controller, logger)
+		}
+	}
+}
+
+func reloadTSOModeFile(path string, controller *TSORuntimeController) error {
+	mode, err := ReadTSOModeFile(path)
+	if err != nil {
+		return err
+	}
+	return errors.WithStack(controller.ApplyMode(mode))
+}
+
+func reportTSOModeReloadError(
+	ctx context.Context,
+	path string,
+	lastError string,
+	err error,
+	controller *TSORuntimeController,
+	logger *slog.Logger,
+) string {
+	if err == nil {
+		return ""
+	}
+	if !errors.Is(err, ErrTSOModeRollback) && !errors.Is(err, ErrUnsafeTSOModeTransition) {
+		controller.observeReload(classifyTSOModeReloadError(err))
+	}
+	if lastError == err.Error() {
+		return lastError
+	}
+	logger.ErrorContext(ctx, "tso mode reload rejected",
+		slog.String("path", path),
+		slog.String("current_mode", controller.CurrentMode().String()),
+		slog.Any("err", err),
+	)
+	return err.Error()
+}
+
+func classifyTSOModeReloadError(err error) string {
+	switch {
+	case errors.Is(err, ErrInvalidTSOMode):
+		return "parse_error"
+	case errors.Is(err, ErrTSOModeRollback), errors.Is(err, ErrUnsafeTSOModeTransition):
+		return "rejected"
+	default:
+		return "read_error"
+	}
+}
diff --git a/kv/tso_runtime_test.go b/kv/tso_runtime_test.go
new file mode 100644
index 000000000..3ed8b39ab
--- /dev/null
+++ b/kv/tso_runtime_test.go
@@ -0,0 +1,407 @@
+package kv
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/bootjp/elastickv/distribution"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseTSOMode(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		raw  string
+		want TSOMode
+	}{
+		{raw: "legacy\n", want: TSOModeLegacy},
+		{raw: "SHADOW", want: TSOModeShadow},
+		{raw: "cutover", want: TSOModeCutover},
+		{raw: "phase-d", want: TSOModePhaseD},
+		{raw: "phase_d", want: TSOModePhaseD},
+	} {
+		mode, err := ParseTSOMode(tc.raw)
+		require.NoError(t, err)
+		require.Equal(t, tc.want, mode)
+	}
+	_, err := ParseTSOMode("disabled")
+	require.ErrorIs(t, err, ErrInvalidTSOMode)
+}
+
+func TestDynamicTimestampAllocatorPreservesLegacyFallbackUntilActivated(t *testing.T) {
+	t.Parallel()
+	clock := NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	dynamic := NewDynamicTimestampAllocator(nil)
+	coord := NewShardedCoordinator(distribution.NewEngine(), nil, 1, clock, nil).
+		WithTSOAllocator(dynamic)
+
+	allocator, ok := TimestampAllocatorThrough(coord)
+	require.False(t, ok)
+	require.Nil(t, allocator)
+	legacyTS, err := NextTimestampThrough(context.Background(), coord, "legacy dynamic fallback")
+	require.NoError(t, err)
+
+	dedicated := &phaseDTestAllocator{next: legacyTS + 100}
+	dynamic.store(dedicated)
+	allocator, ok = TimestampAllocatorThrough(coord)
+	require.True(t, ok)
+	require.Same(t, dedicated, allocator)
+	dedicatedTS, err := NextTimestampThrough(context.Background(), coord, "active dynamic allocator")
+	require.NoError(t, err)
+	require.Equal(t, legacyTS+100, dedicatedTS)
+}
+
+func TestDynamicTimestampAllocatorRejectsValidationWithoutAllocator(t *testing.T) {
+	t.Parallel()
+	dynamic := NewDynamicTimestampAllocator(nil)
+	require.ErrorIs(t,
+		dynamic.ValidateDurableTimestamp(context.Background(), 1),
+		ErrTSOAllocatorRequired,
+	)
+}
+
+func TestTSORuntimeControllerEnforcesAdjacentOneWayTransitions(t *testing.T) {
+	t.Parallel()
+	controller, state := newTestTSORuntimeController(t)
+	require.Equal(t, TSOModeLegacy, controller.CurrentMode())
+	require.Nil(t, controller.Allocator().currentTimestampAllocator())
+
+	err := controller.ApplyMode(TSOModeCutover)
+	require.ErrorIs(t, err, ErrUnsafeTSOModeTransition)
+	require.Equal(t, TSOModeLegacy, controller.CurrentMode())
+
+	require.NoError(t, controller.ApplyMode(TSOModeShadow))
+	require.Equal(t, TSOModeShadow, controller.CurrentMode())
+	require.ErrorIs(t, controller.ApplyMode(TSOModeLegacy), ErrTSOModeRollback)
+
+	require.NoError(t, controller.ApplyMode(TSOModeCutover))
+	_, err = controller.Allocator().Next(context.Background())
+	require.NoError(t, err)
+	require.True(t, state.CutoverActive())
+
+	require.NoError(t, controller.ApplyMode(TSOModePhaseD))
+	controller.Allocator().Invalidate()
+	_, err = controller.Allocator().Next(context.Background())
+	require.NoError(t, err)
+	require.True(t, state.PhaseDActive())
+	require.NoError(t, controller.ApplyMode(TSOModeCutover))
+	require.Equal(t, TSOModePhaseD, controller.CurrentMode())
+}
+
+func TestTSORuntimeControllerDurableStateOverridesStaleMode(t *testing.T) {
+	t.Parallel()
+	state := &runtimeTSOState{}
+	state.cutover.Store(true)
+	controller, _ := newTestTSORuntimeControllerWithState(t, state)
+	require.Equal(t, TSOModeCutover, controller.CurrentMode())
+	require.NotNil(t, controller.Allocator().currentTimestampAllocator())
+	require.NoError(t, controller.ApplyMode(TSOModeShadow))
+	require.Equal(t, TSOModeCutover, controller.CurrentMode())
+
+	state.phaseD.Store(true)
+	require.NoError(t, controller.ApplyMode(TSOModeLegacy))
+	require.Equal(t, TSOModePhaseD, controller.CurrentMode())
+}
+
+func TestTSORuntimeControllerDurableStateMaySkipLocalPhases(t *testing.T) {
+	t.Parallel()
+	state := &runtimeTSOState{}
+	controller, _ := newTestTSORuntimeControllerWithState(t, state)
+
+	state.cutover.Store(true)
+	require.NoError(t, controller.ApplyMode(TSOModeLegacy))
+	require.Equal(t, TSOModeCutover, controller.CurrentMode())
+
+	state.phaseD.Store(true)
+	require.NoError(t, controller.ApplyMode(TSOModeLegacy))
+	require.Equal(t, TSOModePhaseD, controller.CurrentMode())
+}
+
+func TestTSORuntimeControllerAllowsPhaseDAfterDurableCutoverOverride(t *testing.T) {
+	t.Parallel()
+	state := &runtimeTSOState{}
+	controller, _ := newTestTSORuntimeControllerWithState(t, state)
+	require.Equal(t, TSOModeLegacy, controller.CurrentMode())
+
+	state.cutover.Store(true)
+	require.NoError(t, controller.ApplyMode(TSOModePhaseD))
+	require.Equal(t, TSOModePhaseD, controller.CurrentMode())
+}
+
+func TestTSORuntimeControllerDurableCutoverOverridesLegacyBeforeReload(t *testing.T) {
+	t.Parallel()
+	controller, state := newTestTSORuntimeController(t)
+	require.Equal(t, TSOModeLegacy, controller.CurrentMode())
+
+	state.cutover.Store(true)
+	coord := NewShardedCoordinator(distribution.NewEngine(), nil, 1, controller.shadow.legacy, nil).
+		WithTSOAllocator(controller.Allocator())
+	allocator, ok := TimestampAllocatorThrough(coord)
+	require.True(t, ok)
+	require.Same(t, controller.batch, allocator)
+	activateCutover, activatePhaseD := controller.routed.activationState()
+	require.True(t, activateCutover)
+	require.False(t, activatePhaseD)
+
+	_, err := NextTimestampThrough(context.Background(), coord, "durable cutover before reload")
+	require.NoError(t, err)
+	require.Equal(t, TSOModeLegacy, controller.CurrentMode(), "mode poll has not run")
+}
+
+func TestDurableCutoverObservationDoesNotLowerPhaseDActivation(t *testing.T) {
+	t.Parallel()
+	controller, state := newTestTSORuntimeController(t)
+	controller.routed.setActivation(true, true)
+	state.cutover.Store(true)
+
+	require.Same(t, controller.batch, controller.Allocator().currentTimestampAllocator())
+	activateCutover, activatePhaseD := controller.routed.activationState()
+	require.True(t, activateCutover)
+	require.True(t, activatePhaseD)
+}
+
+func TestReloadTSOModeFileRefreshesDurableStateWithoutModeChange(t *testing.T) {
+	t.Parallel()
+	state := &runtimeTSOState{}
+	observer := &recordingRuntimeTSOObserver{}
+	controller, _ := newTestTSORuntimeControllerWithObserver(t, TSOModeCutover, state, observer)
+	path := filepath.Join(t.TempDir(), "tso-mode")
+	writeTSOModeFile(t, path, "cutover\n")
+	observer.durableCalls = 0
+
+	state.cutover.Store(true)
+	require.NoError(t, reloadTSOModeFile(path, controller))
+	require.Equal(t, 1, observer.durableCalls)
+	require.True(t, observer.cutoverActive)
+	require.False(t, observer.phaseDActive)
+}
+
+func TestReportTSOModeReloadErrorCountsEveryFailedPoll(t *testing.T) {
+	t.Parallel()
+	observer := &recordingRuntimeTSOObserver{}
+	controller, _ := newTestTSORuntimeControllerWithObserver(
+		t,
+		TSOModeLegacy,
+		&runtimeTSOState{},
+		observer,
+	)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	err := errors.Wrap(ErrInvalidTSOMode, "parse tso mode file")
+
+	lastError := reportTSOModeReloadError(context.Background(), "mode", "", err, controller, logger)
+	reportTSOModeReloadError(context.Background(), "mode", lastError, err, controller, logger)
+	require.Equal(t, 2, observer.reloads["parse_error"])
+}
+
+func TestRunTSOModeFileReloadAppliesSafeSequence(t *testing.T) {
+	t.Parallel()
+	controller, _ := newTestTSORuntimeController(t)
+	path := filepath.Join(t.TempDir(), "tso-mode")
+	writeTSOModeFile(t, path, "legacy\n")
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- RunTSOModeFileReload(ctx, path, time.Millisecond, controller, slog.Default())
+	}()
+	t.Cleanup(func() {
+		cancel()
+		require.NoError(t, <-done)
+	})
+
+	writeTSOModeFile(t, path, "shadow\n")
+	require.Eventually(t, func() bool {
+		return controller.CurrentMode() == TSOModeShadow
+	}, time.Second, time.Millisecond)
+
+	writeTSOModeFile(t, path, "phase-d\n")
+	require.Never(t, func() bool {
+		return controller.CurrentMode() == TSOModePhaseD
+	}, 20*time.Millisecond, time.Millisecond)
+	require.Equal(t, TSOModeShadow, controller.CurrentMode())
+
+	writeTSOModeFile(t, path, "cutover\n")
+	require.Eventually(t, func() bool {
+		return controller.CurrentMode() == TSOModeCutover
+	}, time.Second, time.Millisecond)
+	writeTSOModeFile(t, path, "phase-d\n")
+	require.Eventually(t, func() bool {
+		return controller.CurrentMode() == TSOModePhaseD
+	}, time.Second, time.Millisecond)
+
+	writeTSOModeFile(t, path, "legacy\n")
+	require.Never(t, func() bool {
+		return controller.CurrentMode() != TSOModePhaseD
+	}, 20*time.Millisecond, time.Millisecond)
+}
+
+func writeTSOModeFile(t *testing.T, path, contents string) {
+	t.Helper()
+	tmp := path + ".tmp"
+	require.NoError(t, os.WriteFile(tmp, []byte(contents), 0o600))
+	require.NoError(t, os.Rename(tmp, path))
+}
+
+func TestTSORuntimeControllerConcurrentAllocationsDuringReload(t *testing.T) {
+	controller, _ := newTestTSORuntimeController(t)
+	clock := NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	coord := NewShardedCoordinator(distribution.NewEngine(), nil, 1, clock, nil).
+		WithTSOAllocator(controller.Allocator())
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 32)
+	for range 16 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range 200 {
+				if _, err := NextTimestampThrough(ctx, coord, "runtime reload race"); err != nil {
+					errs <- err
+					return
+				}
+			}
+		}()
+	}
+	require.NoError(t, controller.ApplyMode(TSOModeShadow))
+	require.NoError(t, controller.ApplyMode(TSOModeCutover))
+	require.NoError(t, controller.ApplyMode(TSOModePhaseD))
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+}
+
+func newTestTSORuntimeController(t *testing.T) (*TSORuntimeController, *runtimeTSOState) {
+	t.Helper()
+	return newTestTSORuntimeControllerWithState(t, &runtimeTSOState{})
+}
+
+func newTestTSORuntimeControllerWithState(
+	t *testing.T,
+	state *runtimeTSOState,
+) (*TSORuntimeController, *runtimeTSOState) {
+	return newTestTSORuntimeControllerWithObserver(t, TSOModeLegacy, state, nil)
+}
+
+func newTestTSORuntimeControllerWithObserver(
+	t *testing.T,
+	initial TSOMode,
+	state *runtimeTSOState,
+	observer TSOObserver,
+) (*TSORuntimeController, *runtimeTSOState) {
+	t.Helper()
+	clock := NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	local := &runtimeTSOAllocator{state: state}
+	routed, err := NewLeaderRoutedTSOAllocator(local, &recordingTSOEngine{}, WithTSORoutedClock(clock))
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, routed.Close()) })
+	controller, err := NewTSORuntimeController(TSORuntimeControllerConfig{
+		Clock:       clock,
+		Routed:      routed,
+		State:       state,
+		BatchSize:   8,
+		InitialMode: initial,
+		Observer:    observer,
+	})
+	require.NoError(t, err)
+	return controller, state
+}
+
+type recordingRuntimeTSOObserver struct {
+	reloads       map[string]int
+	durableCalls  int
+	cutoverActive bool
+	phaseDActive  bool
+}
+
+func (o *recordingRuntimeTSOObserver) ObserveTSORequest(string, string, string, time.Duration) {}
+func (o *recordingRuntimeTSOObserver) ObserveTSOShadowComparison(string, uint64)               {}
+func (o *recordingRuntimeTSOObserver) ObserveTSOMode(string)                                   {}
+
+func (o *recordingRuntimeTSOObserver) ObserveTSOModeReload(result string) {
+	if o.reloads == nil {
+		o.reloads = make(map[string]int)
+	}
+	o.reloads[result]++
+}
+
+func (o *recordingRuntimeTSOObserver) ObserveTSODurableState(cutoverActive, phaseDActive bool) {
+	o.durableCalls++
+	o.cutoverActive = cutoverActive
+	o.phaseDActive = phaseDActive
+}
+
+type runtimeTSOState struct {
+	cutover atomic.Bool
+	phaseD  atomic.Bool
+}
+
+func (s *runtimeTSOState) CutoverActive() bool { return s.cutover.Load() }
+func (s *runtimeTSOState) PhaseDActive() bool  { return s.phaseD.Load() }
+
+type runtimeTSOAllocator struct {
+	state *runtimeTSOState
+	next  atomic.Uint64
+	mu    sync.Mutex
+}
+
+func (a *runtimeTSOAllocator) Next(ctx context.Context) (uint64, error) {
+	return a.NextBatch(ctx, 1)
+}
+
+func (a *runtimeTSOAllocator) NextBatch(ctx context.Context, n int) (uint64, error) {
+	reservation, err := a.ReserveBatchAfter(ctx, n, 0, false, false)
+	return reservation.Base, err
+}
+
+func (a *runtimeTSOAllocator) NextBatchAfter(ctx context.Context, n int, min uint64) (uint64, error) {
+	reservation, err := a.ReserveBatchAfter(ctx, n, min, false, false)
+	return reservation.Base, err
+}
+
+func (a *runtimeTSOAllocator) ReserveBatchAfter(
+	_ context.Context,
+	n int,
+	min uint64,
+	activateCutover bool,
+	activatePhaseD bool,
+) (TSOReservation, error) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if activateCutover || activatePhaseD {
+		a.state.cutover.Store(true)
+	}
+	if activatePhaseD {
+		a.state.phaseD.Store(true)
+	}
+	previous := a.next.Load()
+	base := max(previous+1, min+1)
+	end := base + uint64(n) - 1 //nolint:gosec // test uses positive bounded n.
+	a.next.Store(end)
+	return TSOReservation{
+		Base:                    base,
+		Count:                   n,
+		PreviousAllocationFloor: previous,
+		CutoverActive:           a.state.CutoverActive(),
+		PhaseDActive:            a.state.PhaseDActive(),
+	}, nil
+}
+
+func (a *runtimeTSOAllocator) ValidateDurableTimestamp(context.Context, uint64) error { return nil }
+func (a *runtimeTSOAllocator) PhaseDActive() bool                                     { return a.state.PhaseDActive() }
+func (a *runtimeTSOAllocator) PhaseDRequired() bool                                   { return a.state.PhaseDActive() }
+func (a *runtimeTSOAllocator) IsLeader() bool                                         { return true }
+func (a *runtimeTSOAllocator) RunLeaseRenewal(ctx context.Context)                    { <-ctx.Done() }
diff --git a/main.go b/main.go
index 5f515a91e..af302f1ff 100644
--- a/main.go
+++ b/main.go
@@ -55,6 +55,7 @@ const (
 	etcdMaxSizePerMsg     = 1 << 20
 	etcdMaxInflightMsg    = 1024
 	defaultTSOBatchSize   = 256
+	defaultTSOReload      = 5 * time.Second
 
 	defaultFilesystemRootMode              = 0o755
 	defaultFilesystemPlacementScanInterval = 30 * time.Second
@@ -128,6 +129,8 @@ var (
 	tsoShadowEnabled                = flag.Bool("tsoShadowEnabled", false, "Serialize legacy HLC issuance through the dedicated TSO; fail closed on TSO errors and switch to TSO values after cutover")
 	tsoPhaseDEnabled                = flag.Bool("tsoPhaseDEnabled", false, "Close the centralized-TSO compatibility window after every group-0 member understands the M7 marker")
 	tsoBatchSize                    = flag.Int("tsoBatchSize", defaultTSOBatchSize, "Timestamp batch size used by TSO cutover and shadow validation")
+	tsoModeFile                     = flag.String("tsoModeFile", "", "Path to a runtime-reloadable TSO mode file containing legacy, shadow, cutover, or phase-d")
+	tsoModeReloadInterval           = flag.Duration("tsoModeReloadInterval", defaultTSOReload, "Polling interval for tsoModeFile")
 	leaderBalance                   = flag.Bool("leaderBalance", false, "Enable automatic count-based Raft-group leader balancing on the default-group leader")
 	leaderBalanceInterval           = flag.Duration("leaderBalanceInterval", defaultLeaderBalanceInterval, "Interval between leader-balance scheduler evaluations")
 	leaderBalanceGroupCooldown      = flag.Duration("leaderBalanceGroupCooldown", defaultLeaderBalanceGroupCooldown, "Minimum time before the scheduler can move the same raft group again")
@@ -515,7 +518,12 @@ func run() error {
 		WithKeyVizLabelsEnabled(*keyvizLabelsEnabled).
 		WithAllShardGroups(dataGroupIDs(cfg.groups)...).
 		WithPartitionResolver(buildSQSPartitionResolver(cfg.sqsFifoPartitionMap))
-	tsoWiring, err := configureCoordinatorTSO(coordinate, shardGroups, shardStore)
+	tsoWiring, err := configureCoordinatorTSOWithObserver(
+		coordinate,
+		shardGroups,
+		shardStore,
+		metricsRegistry.TSOObserver(),
+	)
 	if err != nil {
 		return err
 	}
@@ -535,6 +543,7 @@ func run() error {
 		sqsAdvertisesHTFIFO(), slog.Default())
 	cleanup.Add(leadershipRefusalDeregister)
 	eg, runCtx := errgroup.WithContext(ctx)
+	startTSOModeReloader(runCtx, eg, tsoWiring)
 	startRaftEngineLifecycleWatchers(runCtx, eg, runtimes)
 	// setupDistributionCatalog + the Stage 7a process-start registration
 	// gate are bundled so run() has a single startup-fault path: a
@@ -622,6 +631,21 @@ func run() error {
 	return nil
 }
 
+func startTSOModeReloader(ctx context.Context, eg *errgroup.Group, wiring coordinatorTSOWiring) {
+	if eg == nil || wiring.runtimeController == nil || strings.TrimSpace(*tsoModeFile) == "" {
+		return
+	}
+	eg.Go(func() error {
+		return kv.RunTSOModeFileReload(
+			ctx,
+			*tsoModeFile,
+			*tsoModeReloadInterval,
+			wiring.runtimeController,
+			slog.Default(),
+		)
+	})
+}
+
 func startRaftEngineLifecycleWatchers(ctx context.Context, eg *errgroup.Group, runtimes []*raftGroupRuntime) {
 	for _, rt := range runtimes {
 		if rt == nil {
@@ -2060,17 +2084,12 @@ func configurationContainsMember(configuration raftengine.Configuration, localID
 }
 
 type coordinatorTSOWiring struct {
-	serverAllocator kv.TSOAllocator
-	routedAllocator *kv.LeaderRoutedTSOAllocator
-	shadowAllocator *kv.ShadowTimestampAllocator
+	serverAllocator   kv.TSOAllocator
+	routedAllocator   *kv.LeaderRoutedTSOAllocator
+	runtimeController *kv.TSORuntimeController
 }
 
 func (w coordinatorTSOWiring) Close() error {
-	if w.shadowAllocator != nil {
-		if err := w.shadowAllocator.Close(); err != nil {
-			return errors.Wrap(err, "close TSO shadow validator")
-		}
-	}
 	if w.routedAllocator == nil {
 		return nil
 	}
@@ -2087,38 +2106,98 @@ func configureCoordinatorTSO(
 	coordinate *kv.ShardedCoordinator,
 	shardGroups map[uint64]*kv.ShardGroup,
 	floorProviders ...kv.TSOCutoverFloorProvider,
+) (coordinatorTSOWiring, error) {
+	var floorProvider kv.TSOCutoverFloorProvider
+	if len(floorProviders) > 0 {
+		floorProvider = floorProviders[0]
+	}
+	return configureCoordinatorTSOWithObserver(coordinate, shardGroups, floorProvider, nil)
+}
+
+func configureCoordinatorTSOWithObserver(
+	coordinate *kv.ShardedCoordinator,
+	shardGroups map[uint64]*kv.ShardGroup,
+	floorProvider kv.TSOCutoverFloorProvider,
+	observer kv.TSOObserver,
 ) (coordinatorTSOWiring, error) {
 	var wiring coordinatorTSOWiring
 	if coordinate == nil {
 		return wiring, errors.Wrap(kv.ErrTSOCoordinatorNil, "configure tso allocator")
 	}
-	if *tsoEnabled && *tsoShadowEnabled {
-		return wiring, errors.New("--tsoEnabled and --tsoShadowEnabled are mutually exclusive")
-	}
-	if *tsoPhaseDEnabled && !*tsoEnabled {
-		return wiring, errors.New("--tsoPhaseDEnabled requires --tsoEnabled")
+	mode, err := configuredTSOMode()
+	if err != nil {
+		return wiring, err
 	}
 
 	tsoGroup, dedicated := shardGroups[dedicatedTSORaftGroupID]
 	if !dedicated {
-		return configureLegacyCoordinatorTSO(coordinate)
+		return configureLegacyCoordinatorTSO(coordinate, mode)
 	}
-	var floorProvider kv.TSOCutoverFloorProvider
-	if len(floorProviders) > 0 {
-		floorProvider = floorProviders[0]
+	return configureDedicatedCoordinatorTSO(coordinate, tsoGroup, floorProvider, mode, observer)
+}
+
+func configuredTSOMode() (kv.TSOMode, error) {
+	if err := validateTSOModeFlags(); err != nil {
+		return kv.TSOModeLegacy, err
+	}
+	if strings.TrimSpace(*tsoModeFile) != "" {
+		mode, err := kv.ReadTSOModeFile(*tsoModeFile)
+		return mode, errors.Wrap(err, "load initial tso runtime mode")
+	}
+	return tsoModeFromLegacyFlags(), nil
+}
+
+func validateTSOModeFlags() error {
+	if err := validateLegacyTSOModeFlags(); err != nil {
+		return err
+	}
+	if strings.TrimSpace(*tsoModeFile) == "" {
+		return nil
+	}
+	if *tsoEnabled || *tsoShadowEnabled || *tsoPhaseDEnabled {
+		return errors.New("--tsoModeFile cannot be combined with tsoEnabled, tsoShadowEnabled, or tsoPhaseDEnabled")
+	}
+	if *tsoModeReloadInterval <= 0 {
+		return errors.New("--tsoModeReloadInterval must be positive")
 	}
-	return configureDedicatedCoordinatorTSO(coordinate, tsoGroup, floorProvider)
+	return nil
 }
 
-func configureLegacyCoordinatorTSO(coordinate *kv.ShardedCoordinator) (coordinatorTSOWiring, error) {
+func validateLegacyTSOModeFlags() error {
+	if *tsoEnabled && *tsoShadowEnabled {
+		return errors.New("--tsoEnabled and --tsoShadowEnabled are mutually exclusive")
+	}
+	if *tsoPhaseDEnabled && !*tsoEnabled {
+		return errors.New("--tsoPhaseDEnabled requires --tsoEnabled")
+	}
+	return nil
+}
+
+func tsoModeFromLegacyFlags() kv.TSOMode {
+	switch {
+	case *tsoPhaseDEnabled:
+		return kv.TSOModePhaseD
+	case *tsoEnabled:
+		return kv.TSOModeCutover
+	case *tsoShadowEnabled:
+		return kv.TSOModeShadow
+	default:
+		return kv.TSOModeLegacy
+	}
+}
+
+func configureLegacyCoordinatorTSO(coordinate *kv.ShardedCoordinator, mode kv.TSOMode) (coordinatorTSOWiring, error) {
 	var wiring coordinatorTSOWiring
-	if *tsoPhaseDEnabled {
+	if strings.TrimSpace(*tsoModeFile) != "" {
+		return wiring, errors.Wrap(kv.ErrTSOGroupRequired, "configure tso mode reload")
+	}
+	if mode == kv.TSOModePhaseD {
 		return wiring, errors.Wrap(kv.ErrTSOGroupRequired, "configure tso phase D")
 	}
-	if *tsoShadowEnabled {
+	if mode == kv.TSOModeShadow {
 		return wiring, errors.Wrap(kv.ErrTSOGroupRequired, "configure tso shadow allocator")
 	}
-	if !*tsoEnabled {
+	if mode == kv.TSOModeLegacy {
 		return wiring, nil
 	}
 	// Preserve the pre-group-0 bridge for deployments that enable TSO
@@ -2139,6 +2218,8 @@ func configureDedicatedCoordinatorTSO(
 	coordinate *kv.ShardedCoordinator,
 	tsoGroup *kv.ShardGroup,
 	floorProvider kv.TSOCutoverFloorProvider,
+	mode kv.TSOMode,
+	observer kv.TSOObserver,
 ) (coordinatorTSOWiring, error) {
 	var wiring coordinatorTSOWiring
 	local, err := kv.NewRaftTSOAllocator(
@@ -2152,64 +2233,46 @@ func configureDedicatedCoordinatorTSO(
 	wiring.serverAllocator = local
 	coordinate.WithTimestampGroup(dedicatedTSORaftGroupID)
 	coordinate.WithTSOCutoverState(tsoGroup.TSOState)
-	if !dedicatedTSOAllocatorRequired(tsoGroup.TSOState) {
+	if !dedicatedTSOAllocatorRequired(tsoGroup.TSOState, mode) {
 		return wiring, nil
 	}
 
-	routedOpts := dedicatedTSORoutingOptions(coordinate.Clock())
+	routedOpts := dedicatedTSORoutingOptions(coordinate.Clock(), observer)
 	routed, err := kv.NewLeaderRoutedTSOAllocator(local, tsoGroup.Engine, routedOpts...)
 	if err != nil {
 		return wiring, errors.Wrap(err, "configure tso leader routing")
 	}
 	wiring.routedAllocator = routed
-	return installDedicatedCoordinatorAllocator(coordinate, tsoGroup.TSOState, wiring, routed)
+	controller, err := kv.NewTSORuntimeController(kv.TSORuntimeControllerConfig{
+		Clock:       coordinate.Clock(),
+		Routed:      routed,
+		State:       tsoGroup.TSOState,
+		BatchSize:   *tsoBatchSize,
+		InitialMode: mode,
+		Logger:      slog.Default(),
+		Observer:    observer,
+	})
+	if err != nil {
+		_ = wiring.Close()
+		return coordinatorTSOWiring{}, errors.Wrap(err, "configure tso runtime controller")
+	}
+	wiring.runtimeController = controller
+	coordinate.WithTSOAllocator(controller.Allocator())
+	return wiring, nil
 }
 
-func dedicatedTSOAllocatorRequired(state *kv.TSOStateMachine) bool {
-	return *tsoEnabled || *tsoShadowEnabled || (state != nil && state.CutoverActive())
+func dedicatedTSOAllocatorRequired(state *kv.TSOStateMachine, mode kv.TSOMode) bool {
+	return strings.TrimSpace(*tsoModeFile) != "" || mode != kv.TSOModeLegacy || (state != nil && state.CutoverActive())
 }
 
-func dedicatedTSORoutingOptions(clock *kv.HLC) []kv.LeaderRoutedTSOAllocatorOption {
+func dedicatedTSORoutingOptions(clock *kv.HLC, observer kv.TSOObserver) []kv.LeaderRoutedTSOAllocatorOption {
 	opts := []kv.LeaderRoutedTSOAllocatorOption{kv.WithTSORoutedClock(clock)}
-	if *tsoEnabled {
-		opts = append(opts, kv.WithTSOCutoverActivation())
-	}
-	if *tsoPhaseDEnabled {
-		opts = append(opts, kv.WithTSOPhaseDActivation())
+	if observer != nil {
+		opts = append(opts, kv.WithTSOObserver(observer))
 	}
 	return opts
 }
 
-func installDedicatedCoordinatorAllocator(
-	coordinate *kv.ShardedCoordinator,
-	state *kv.TSOStateMachine,
-	wiring coordinatorTSOWiring,
-	routed *kv.LeaderRoutedTSOAllocator,
-) (coordinatorTSOWiring, error) {
-	if *tsoShadowEnabled && (state == nil || !state.CutoverActive()) {
-		shadow, err := kv.NewShadowTimestampAllocator(
-			coordinate.Clock(),
-			routed,
-			slog.Default(),
-			kv.WithTSOShadowCutoverState(state),
-		)
-		if err != nil {
-			_ = wiring.Close()
-			return coordinatorTSOWiring{}, errors.Wrap(err, "configure tso shadow allocator")
-		}
-		wiring.shadowAllocator = shadow
-		coordinate.WithTSOAllocator(shadow)
-		return wiring, nil
-	}
-	batch, err := kv.NewBatchAllocator(routed, *tsoBatchSize)
-	if err != nil {
-		_ = wiring.Close()
-		return coordinatorTSOWiring{}, errors.Wrap(err, "configure tso batch allocator")
-	}
-	coordinate.WithTSOAllocator(batch)
-	return wiring, nil
-}
-
 type hlcLeaseRenewalBlocker interface {
 	SetHLCLeaseRenewalBlocker(func() bool)
 }
@@ -2396,7 +2459,7 @@ func (c startupGatedCoordinator) Clock() *kv.HLC {
 }
 
 func (c startupGatedCoordinator) TimestampAllocator() kv.TimestampAllocator {
-	alloc, _ := kv.TimestampAllocatorThrough(c.inner)
+	alloc, _ := kv.ConfiguredTimestampAllocatorThrough(c.inner)
 	return alloc
 }
 
@@ -2812,7 +2875,7 @@ func startRaftServers(
 }
 
 func internalTimestampOptions(coordinate kv.Coordinator) []adapter.InternalOption {
-	if alloc, ok := kv.TimestampAllocatorThrough(coordinate); ok {
+	if alloc, ok := kv.ConfiguredTimestampAllocatorThrough(coordinate); ok {
 		return []adapter.InternalOption{adapter.WithInternalTimestampAllocator(alloc)}
 	}
 	return nil
diff --git a/main_tso_routing_test.go b/main_tso_routing_test.go
index a2527fa5f..db4e9dcc3 100644
--- a/main_tso_routing_test.go
+++ b/main_tso_routing_test.go
@@ -3,6 +3,8 @@ package main
 import (
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -139,7 +141,6 @@ func TestConfigureCoordinatorTSORestoresDurableCutoverWithoutFlags(t *testing.T)
 	t.Cleanup(func() { require.NoError(t, wiring.Close()) })
 	require.NotNil(t, wiring.serverAllocator)
 	require.NotNil(t, wiring.routedAllocator)
-	require.Nil(t, wiring.shadowAllocator)
 	require.True(t, coord.IsTimestampLeader())
 	require.True(t, fsm.CutoverActive())
 
@@ -158,7 +159,8 @@ func TestConfigureCoordinatorTSODurableCutoverOverridesShadowFlag(t *testing.T)
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, wiring.Close()) })
 	require.NotNil(t, wiring.routedAllocator)
-	require.Nil(t, wiring.shadowAllocator, "durable cutover cannot return to legacy shadow issuance")
+	require.Equal(t, kv.TSOModeCutover, wiring.runtimeController.CurrentMode(),
+		"durable cutover cannot return to legacy shadow issuance")
 	require.True(t, coord.IsTimestampLeader())
 	require.True(t, fsm.CutoverActive())
 
@@ -179,10 +181,56 @@ func TestInternalTimestampOptionsPreservesTSOThroughStartupGate(t *testing.T) {
 	require.Same(t, allocator, got)
 	require.Len(t, internalTimestampOptions(gated), 1)
 
+	runtimeAllocator := kv.NewDynamicTimestampAllocator(nil)
+	runtimeCoord := newMainTSOCoordinator(kv.NewHLC(), nil).WithTSOAllocator(runtimeAllocator)
+	runtimeGated := startupGatedCoordinator{inner: runtimeCoord, gate: &startupPublicKVGate{}}
+	got, ok = kv.TimestampAllocatorThrough(runtimeGated)
+	require.False(t, ok)
+	require.Nil(t, got)
+	configured, ok := kv.ConfiguredTimestampAllocatorThrough(runtimeGated)
+	require.True(t, ok)
+	require.Same(t, runtimeAllocator, configured)
+	require.Len(t, internalTimestampOptions(runtimeGated), 1)
+
 	legacy := startupGatedCoordinator{inner: newMainTSOCoordinator(kv.NewHLC(), nil)}
 	require.Empty(t, internalTimestampOptions(legacy))
 }
 
+func TestConfigureCoordinatorTSOModeFileStartsRuntimeController(t *testing.T) {
+	setTSOModeFlags(t, false, false)
+	path := filepath.Join(t.TempDir(), "tso-mode")
+	require.NoError(t, os.WriteFile(path, []byte("shadow\n"), 0o600))
+	*tsoModeFile = path
+	*tsoModeReloadInterval = time.Millisecond
+	clock := kv.NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	fsm := kv.NewTSOStateMachine(clock)
+	engine := &mainTSOEngine{state: raftengine.StateLeader, tsoState: fsm}
+	groups := map[uint64]*kv.ShardGroup{
+		dedicatedTSORaftGroupID: {Engine: engine, TSOState: fsm},
+	}
+	coord := newMainTSOCoordinator(clock, groups)
+
+	wiring, err := configureCoordinatorTSO(coord, groups, mainTSOFloorProvider{})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, wiring.Close()) })
+	require.NotNil(t, wiring.runtimeController)
+	require.Equal(t, kv.TSOModeShadow, wiring.runtimeController.CurrentMode())
+	_, configured := kv.TimestampAllocatorThrough(coord)
+	require.True(t, configured)
+}
+
+func TestConfigureCoordinatorTSOModeFileRequiresDedicatedGroup(t *testing.T) {
+	setTSOModeFlags(t, false, false)
+	path := filepath.Join(t.TempDir(), "tso-mode")
+	require.NoError(t, os.WriteFile(path, []byte("legacy\n"), 0o600))
+	*tsoModeFile = path
+	coord := newMainTSOCoordinator(kv.NewHLC(), nil)
+
+	_, err := configureCoordinatorTSO(coord, nil)
+	require.ErrorIs(t, err, kv.ErrTSOGroupRequired)
+}
+
 func newActiveMainTSOCutover(t *testing.T) (*kv.HLC, *kv.TSOStateMachine, *mainTSOEngine, map[uint64]*kv.ShardGroup) {
 	t.Helper()
 	clock := kv.NewHLC()
@@ -205,15 +253,21 @@ func setTSOModeFlags(t *testing.T, enabled, shadow bool) {
 	oldShadow := *tsoShadowEnabled
 	oldPhaseD := *tsoPhaseDEnabled
 	oldBatchSize := *tsoBatchSize
+	oldModeFile := *tsoModeFile
+	oldReloadInterval := *tsoModeReloadInterval
 	*tsoEnabled = enabled
 	*tsoShadowEnabled = shadow
 	*tsoPhaseDEnabled = false
 	*tsoBatchSize = 8
+	*tsoModeFile = ""
+	*tsoModeReloadInterval = defaultTSOReload
 	t.Cleanup(func() {
 		*tsoEnabled = oldEnabled
 		*tsoShadowEnabled = oldShadow
 		*tsoPhaseDEnabled = oldPhaseD
 		*tsoBatchSize = oldBatchSize
+		*tsoModeFile = oldModeFile
+		*tsoModeReloadInterval = oldReloadInterval
 	})
 }
 
diff --git a/monitoring/prometheus/rules/tso-alerts.yml b/monitoring/prometheus/rules/tso-alerts.yml
new file mode 100644
index 000000000..bbee49800
--- /dev/null
+++ b/monitoring/prometheus/rules/tso-alerts.yml
@@ -0,0 +1,103 @@
+groups:
+  - name: elastickv-tso
+    rules:
+      - alert: ElasticKVTSOAllocationErrorRateHigh
+        expr: |
+          (
+            sum(rate(elastickv_tso_request_duration_seconds_count{operation="reserve",outcome="error"}[5m]))
+            /
+            clamp_min(sum(rate(elastickv_tso_request_duration_seconds_count{operation="reserve"}[5m])), 0.001)
+          ) > 0.01
+          and
+          sum(rate(elastickv_tso_request_duration_seconds_count{operation="reserve"}[5m])) > 0.1
+        for: 10m
+        labels:
+          severity: warning
+        annotations:
+          summary: TSO allocation error rate exceeds 1 percent
+          description: Dedicated TSO reserve attempts have exceeded a 1 percent error rate for 10 minutes.
+
+      - alert: ElasticKVTSOAllocationErrorRateCritical
+        expr: |
+          (
+            sum(rate(elastickv_tso_request_duration_seconds_count{operation="reserve",outcome="error"}[5m]))
+            /
+            clamp_min(sum(rate(elastickv_tso_request_duration_seconds_count{operation="reserve"}[5m])), 0.001)
+          ) > 0.10
+          and
+          sum(rate(elastickv_tso_request_duration_seconds_count{operation="reserve"}[5m])) > 0.1
+        for: 5m
+        labels:
+          severity: critical
+        annotations:
+          summary: TSO allocation error rate exceeds 10 percent
+          description: Dedicated TSO reserve attempts have exceeded a 10 percent error rate for 5 minutes; writes may be failing closed.
+
+      - alert: ElasticKVTSOAllocationLatencyHigh
+        expr: |
+          histogram_quantile(
+            0.99,
+            sum by (le) (
+              rate(elastickv_tso_request_duration_seconds_bucket{operation="reserve",outcome="success"}[5m])
+            )
+          ) > 0.05
+        for: 10m
+        labels:
+          severity: warning
+        annotations:
+          summary: TSO allocation p99 exceeds 50 milliseconds
+          description: Successful dedicated TSO reserve attempts have exceeded 50 ms p99 for 10 minutes.
+
+      - alert: ElasticKVTSOAllocationLatencyCritical
+        expr: |
+          histogram_quantile(
+            0.99,
+            sum by (le) (
+              rate(elastickv_tso_request_duration_seconds_bucket{operation="reserve",outcome="success"}[5m])
+            )
+          ) > 0.20
+        for: 5m
+        labels:
+          severity: critical
+        annotations:
+          summary: TSO allocation p99 exceeds 200 milliseconds
+          description: Successful dedicated TSO reserve attempts have exceeded 200 ms p99 for 5 minutes.
+
+      - alert: ElasticKVTSOShadowOverlapDetected
+        expr: sum(increase(elastickv_tso_shadow_comparisons_total{result="legacy_overlap"}[15m])) > 0
+        for: 5m
+        labels:
+          severity: warning
+        annotations:
+          summary: Legacy timestamps overlap the durable TSO floor
+          description: At least one shadow candidate was discarded in the last 15 minutes. Keep every node in shadow mode until this remains zero for a full 15-minute window.
+
+      - alert: ElasticKVTSOModeReloadRejected
+        expr: sum(increase(elastickv_tso_mode_reload_total{result=~"rejected|parse_error|read_error"}[10m])) > 0
+        for: 1m
+        labels:
+          severity: warning
+        annotations:
+          summary: A runtime TSO mode reload was rejected
+          description: A node could not read or parse its mode file, or the requested transition violated the one-way phase order.
+
+      - alert: ElasticKVTSOModeDivergence
+        expr: count(count by (mode) (elastickv_tso_mode == 1)) > 1
+        for: 15m
+        labels:
+          severity: warning
+        annotations:
+          summary: Cluster nodes remain in different TSO modes
+          description: More than one process-local TSO mode has remained active for 15 minutes. Finish the rolling transition or investigate a rejected reload.
+
+      - alert: ElasticKVTSOPhaseDWithoutCutover
+        expr: |
+          max(elastickv_tso_durable_state{marker="phase_d"})
+          >
+          max(elastickv_tso_durable_state{marker="cutover"})
+        for: 1m
+        labels:
+          severity: critical
+        annotations:
+          summary: Durable TSO marker ordering is invalid
+          description: Phase D is visible without the prerequisite durable cutover marker. Stop rollout and inspect group-0 state.
diff --git a/monitoring/registry.go b/monitoring/registry.go
index 27acdb10c..568bfeeb6 100644
--- a/monitoring/registry.go
+++ b/monitoring/registry.go
@@ -29,6 +29,8 @@ type Registry struct {
 	hlcObserver   *HLCObserver
 	coldStart     *ColdStartMetrics
 	coldStartObs  *ColdStartObserver
+	tso           *TSOMetrics
+	tsoObserver   *TSOObserver
 }
 
 // NewRegistry builds a registry with constant labels that identify the local node.
@@ -59,6 +61,8 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry {
 	r.hlcObserver = newHLCObserver(r.hlc)
 	r.coldStart = newColdStartMetrics(registerer)
 	r.coldStartObs = newColdStartObserver(r.coldStart)
+	r.tso = newTSOMetrics(registerer)
+	r.tsoObserver = newTSOObserver(r.tso)
 	return r
 }
 
@@ -279,3 +283,12 @@ func (r *Registry) HLCObserver() *HLCObserver {
 	}
 	return r.hlcObserver
 }
+
+// TSOObserver returns the shared runtime TSO observer. The kv package consumes
+// it through its narrow observer interface, keeping Prometheus ownership here.
+func (r *Registry) TSOObserver() *TSOObserver {
+	if r == nil {
+		return nil
+	}
+	return r.tsoObserver
+}
diff --git a/monitoring/tso.go b/monitoring/tso.go
new file mode 100644
index 000000000..535db2a40
--- /dev/null
+++ b/monitoring/tso.go
@@ -0,0 +1,146 @@
+package monitoring
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var tsoRequestDurationBuckets = []float64{
+	0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005,
+	0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 5,
+}
+
+const (
+	tsoDivergenceBucketFactor = 16
+	tsoDivergenceBucketCount  = 12
+)
+
+type TSOMetrics struct {
+	requestDuration  *prometheus.HistogramVec
+	shadowComparison *prometheus.CounterVec
+	shadowDivergence *prometheus.HistogramVec
+	mode             *prometheus.GaugeVec
+	reloads          *prometheus.CounterVec
+	durableState     *prometheus.GaugeVec
+}
+
+func newTSOMetrics(registerer prometheus.Registerer) *TSOMetrics {
+	m := &TSOMetrics{
+		requestDuration: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name:    "elastickv_tso_request_duration_seconds",
+				Help:    "End-to-end duration of dedicated TSO reserve and validation attempts by local or remote leader path and outcome.",
+				Buckets: tsoRequestDurationBuckets,
+			},
+			[]string{"operation", "path", "outcome"},
+		),
+		shadowComparison: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "elastickv_tso_shadow_comparisons_total",
+				Help: "Shadow migration comparisons by bounded result: legacy_ahead, legacy_overlap, cutover_active, or error.",
+			},
+			[]string{"result"},
+		),
+		shadowDivergence: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name:    "elastickv_tso_shadow_divergence_timestamps",
+				Help:    "Absolute timestamp distance between a shadow legacy candidate and the preceding durable TSO allocation floor.",
+				Buckets: prometheus.ExponentialBuckets(1, tsoDivergenceBucketFactor, tsoDivergenceBucketCount),
+			},
+			[]string{"result"},
+		),
+		mode: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "elastickv_tso_mode",
+				Help: "Current process-local TSO mode as a one-hot gauge.",
+			},
+			[]string{"mode"},
+		),
+		reloads: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "elastickv_tso_mode_reload_total",
+				Help: "Runtime TSO mode reload outcomes.",
+			},
+			[]string{"result"},
+		),
+		durableState: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "elastickv_tso_durable_state",
+				Help: "Consensus-owned one-way TSO marker state for cutover and phase_d.",
+			},
+			[]string{"marker"},
+		),
+	}
+	if registerer != nil {
+		registerer.MustRegister(
+			m.requestDuration,
+			m.shadowComparison,
+			m.shadowDivergence,
+			m.mode,
+			m.reloads,
+			m.durableState,
+		)
+	}
+	return m
+}
+
+type TSOObserver struct {
+	metrics *TSOMetrics
+}
+
+func newTSOObserver(metrics *TSOMetrics) *TSOObserver {
+	return &TSOObserver{metrics: metrics}
+}
+
+func (o *TSOObserver) ObserveTSORequest(operation, path, outcome string, duration time.Duration) {
+	if o == nil || o.metrics == nil {
+		return
+	}
+	o.metrics.requestDuration.WithLabelValues(operation, path, outcome).Observe(duration.Seconds())
+}
+
+func (o *TSOObserver) ObserveTSOShadowComparison(result string, divergence uint64) {
+	if o == nil || o.metrics == nil {
+		return
+	}
+	o.metrics.shadowComparison.WithLabelValues(result).Inc()
+	if result == "legacy_ahead" || result == "legacy_overlap" {
+		o.metrics.shadowDivergence.WithLabelValues(result).Observe(float64(divergence))
+	}
+}
+
+func (o *TSOObserver) ObserveTSOMode(mode string) {
+	if o == nil || o.metrics == nil {
+		return
+	}
+	for _, candidate := range []string{"legacy", "shadow", "cutover", "phase-d"} {
+		value := 0.0
+		if candidate == mode {
+			value = 1
+		}
+		o.metrics.mode.WithLabelValues(candidate).Set(value)
+	}
+}
+
+func (o *TSOObserver) ObserveTSOModeReload(result string) {
+	if o == nil || o.metrics == nil {
+		return
+	}
+	o.metrics.reloads.WithLabelValues(result).Inc()
+}
+
+func (o *TSOObserver) ObserveTSODurableState(cutoverActive, phaseDActive bool) {
+	if o == nil || o.metrics == nil {
+		return
+	}
+	o.metrics.durableState.WithLabelValues("cutover").Set(boolMetricValue(cutoverActive))
+	o.metrics.durableState.WithLabelValues("phase_d").Set(boolMetricValue(phaseDActive))
+}
+
+func boolMetricValue(value bool) float64 {
+	if value {
+		return 1
+	}
+	return 0
+}
diff --git a/monitoring/tso_test.go b/monitoring/tso_test.go
new file mode 100644
index 000000000..7c220c156
--- /dev/null
+++ b/monitoring/tso_test.go
@@ -0,0 +1,48 @@
+package monitoring
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTSOObserverExportsOperationalMetrics(t *testing.T) {
+	t.Parallel()
+	reg := prometheus.NewRegistry()
+	metrics := newTSOMetrics(reg)
+	observer := newTSOObserver(metrics)
+
+	observer.ObserveTSORequest("reserve", "remote", "success", 3*time.Millisecond)
+	observer.ObserveTSORequest("reserve", "remote", "error", 20*time.Millisecond)
+	observer.ObserveTSOShadowComparison("legacy_ahead", 7)
+	observer.ObserveTSOShadowComparison("legacy_overlap", 3)
+	observer.ObserveTSOMode("shadow")
+	observer.ObserveTSOModeReload("applied")
+	observer.ObserveTSODurableState(true, false)
+
+	require.Equal(t, 1.0, testutil.ToFloat64(metrics.shadowComparison.WithLabelValues("legacy_ahead")))
+	require.Equal(t, 1.0, testutil.ToFloat64(metrics.shadowComparison.WithLabelValues("legacy_overlap")))
+	require.Equal(t, 1.0, testutil.ToFloat64(metrics.mode.WithLabelValues("shadow")))
+	require.Equal(t, 0.0, testutil.ToFloat64(metrics.mode.WithLabelValues("legacy")))
+	require.Equal(t, 1.0, testutil.ToFloat64(metrics.reloads.WithLabelValues("applied")))
+	require.Equal(t, 1.0, testutil.ToFloat64(metrics.durableState.WithLabelValues("cutover")))
+	require.Equal(t, 0.0, testutil.ToFloat64(metrics.durableState.WithLabelValues("phase_d")))
+
+	families, err := reg.Gather()
+	require.NoError(t, err)
+	require.True(t, hasMetricFamily(families, "elastickv_tso_request_duration_seconds"))
+	require.True(t, hasMetricFamily(families, "elastickv_tso_shadow_divergence_timestamps"))
+}
+
+func hasMetricFamily(families []*dto.MetricFamily, name string) bool {
+	for _, family := range families {
+		if family.GetName() == name {
+			return true
+		}
+	}
+	return false
+}

From 229c2069f72c138b55d344da004ae300b89c5b1d Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Thu, 23 Jul 2026 17:19:07 +0900
Subject: [PATCH 3/5] proto: reserve removed raft status fields

---
 proto/service.pb.go | 4 ++--
 proto/service.proto | 2 ++
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/proto/service.pb.go b/proto/service.pb.go
index a70a3434b..bf6cd8ff9 100644
--- a/proto/service.pb.go
+++ b/proto/service.pb.go
@@ -2377,7 +2377,7 @@ const file_service_proto_rawDesc = "" +
 	"\bstart_ts\x18\x01 \x01(\x04R\astartTs\",\n" +
 	"\x10RollbackResponse\x12\x18\n" +
 	"\asuccess\x18\x01 \x01(\bR\asuccess\"\x18\n" +
-	"\x16RaftAdminStatusRequest\"\xa2\x03\n" +
+	"\x16RaftAdminStatusRequest\"\xae\x03\n" +
 	"\x17RaftAdminStatusResponse\x12%\n" +
 	"\x05state\x18\x01 \x01(\x0e2\x0f.RaftAdminStateR\x05state\x12\x1b\n" +
 	"\tleader_id\x18\x02 \x01(\tR\bleaderId\x12%\n" +
@@ -2391,7 +2391,7 @@ const file_service_proto_rawDesc = "" +
 	"fsmPending\x12\x1b\n" +
 	"\tnum_peers\x18\n" +
 	" \x01(\x04R\bnumPeers\x12,\n" +
-	"\x12last_contact_nanos\x18\v \x01(\x03R\x10lastContactNanos\"\x1f\n" +
+	"\x12last_contact_nanos\x18\v \x01(\x03R\x10lastContactNanosJ\x04\b\f\x10\rJ\x04\b\r\x10\x0e\"\x1f\n" +
 	"\x1dRaftAdminConfigurationRequest\"W\n" +
 	"\x0fRaftAdminMember\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\tR\x02id\x12\x18\n" +
diff --git a/proto/service.proto b/proto/service.proto
index 6b01e6012..815e67648 100644
--- a/proto/service.proto
+++ b/proto/service.proto
@@ -187,6 +187,8 @@ enum RaftAdminState {
 message RaftAdminStatusRequest {}
 
 message RaftAdminStatusResponse {
+  reserved 12, 13;
+
   RaftAdminState state = 1;
   string leader_id = 2;
   string leader_address = 3;

From 917f421b4c1f6f36b90d612f59315d6818867941 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Thu, 23 Jul 2026 18:20:40 +0900
Subject: [PATCH 4/5] tso: preserve runtime allocator handles

---
 kv/keyviz_label.go       |  5 ++++
 kv/tso.go                | 12 +++++++++
 kv/tso_runtime_test.go   | 34 +++++++++++++++++++++++++
 main.go                  |  6 +++++
 main_tso_routing_test.go | 55 ++++++++++++++++++++++++++++++++++++++++
 5 files changed, 112 insertions(+)

diff --git a/kv/keyviz_label.go b/kv/keyviz_label.go
index 09a3629c3..7a1fe5daf 100644
--- a/kv/keyviz_label.go
+++ b/kv/keyviz_label.go
@@ -91,6 +91,11 @@ func (c keyVizLabeledCoordinator) RaftLeaderForKey(key []byte) string {
 func (c keyVizLabeledCoordinator) Clock() *HLC { return c.inner.Clock() }
 
 func (c keyVizLabeledCoordinator) TimestampAllocator() TimestampAllocator {
+	alloc, _ := TimestampAllocatorThrough(c.inner)
+	return alloc
+}
+
+func (c keyVizLabeledCoordinator) ConfiguredTimestampAllocator() TimestampAllocator {
 	alloc, _ := ConfiguredTimestampAllocatorThrough(c.inner)
 	return alloc
 }
diff --git a/kv/tso.go b/kv/tso.go
index 0f0be8181..99e67b37a 100644
--- a/kv/tso.go
+++ b/kv/tso.go
@@ -48,6 +48,14 @@ type TimestampAllocatorProvider interface {
 	TimestampAllocator() TimestampAllocator
 }
 
+// ConfiguredTimestampAllocatorProvider lets coordinator decorators expose the
+// configured allocator without resolving runtime-mode wrappers. Long-lived
+// services use this to keep a DynamicTimestampAllocator reference while the
+// active mode is still legacy.
+type ConfiguredTimestampAllocatorProvider interface {
+	ConfiguredTimestampAllocator() TimestampAllocator
+}
+
 type TimestampAfterAllocator interface {
 	NextAfter(ctx context.Context, min uint64) (uint64, error)
 }
@@ -231,6 +239,10 @@ func TimestampAllocatorThrough(coord Coordinator) (TimestampAllocator, bool) {
 // that must keep a DynamicTimestampAllocator reference across later mode-file
 // reloads while still falling back to HLC when that allocator reports legacy.
 func ConfiguredTimestampAllocatorThrough(coord Coordinator) (TimestampAllocator, bool) {
+	if provider, ok := coord.(ConfiguredTimestampAllocatorProvider); ok {
+		alloc := provider.ConfiguredTimestampAllocator()
+		return alloc, alloc != nil
+	}
 	if provider, ok := coord.(TimestampAllocatorProvider); ok {
 		alloc := provider.TimestampAllocator()
 		return alloc, alloc != nil
diff --git a/kv/tso_runtime_test.go b/kv/tso_runtime_test.go
index 3ed8b39ab..c2b794170 100644
--- a/kv/tso_runtime_test.go
+++ b/kv/tso_runtime_test.go
@@ -60,6 +60,26 @@ func TestDynamicTimestampAllocatorPreservesLegacyFallbackUntilActivated(t *testi
 	require.Equal(t, legacyTS+100, dedicatedTS)
 }
 
+func TestConfiguredTimestampAllocatorThroughPrefersConfiguredProvider(t *testing.T) {
+	t.Parallel()
+
+	active := &phaseDTestAllocator{next: 10}
+	configured := NewDynamicTimestampAllocator(nil)
+	provider := configuredAllocatorTestProvider{
+		Coordinator: NewShardedCoordinator(distribution.NewEngine(), nil, 1, NewHLC(), nil),
+		active:      active,
+		configured:  configured,
+	}
+
+	got, ok := TimestampAllocatorThrough(provider)
+	require.True(t, ok)
+	require.Same(t, active, got)
+
+	got, ok = ConfiguredTimestampAllocatorThrough(provider)
+	require.True(t, ok)
+	require.Same(t, configured, got)
+}
+
 func TestDynamicTimestampAllocatorRejectsValidationWithoutAllocator(t *testing.T) {
 	t.Parallel()
 	dynamic := NewDynamicTimestampAllocator(nil)
@@ -349,6 +369,20 @@ type runtimeTSOState struct {
 	phaseD  atomic.Bool
 }
 
+type configuredAllocatorTestProvider struct {
+	Coordinator
+	active     TimestampAllocator
+	configured TimestampAllocator
+}
+
+func (p configuredAllocatorTestProvider) TimestampAllocator() TimestampAllocator {
+	return p.active
+}
+
+func (p configuredAllocatorTestProvider) ConfiguredTimestampAllocator() TimestampAllocator {
+	return p.configured
+}
+
 func (s *runtimeTSOState) CutoverActive() bool { return s.cutover.Load() }
 func (s *runtimeTSOState) PhaseDActive() bool  { return s.phaseD.Load() }
 
diff --git a/main.go b/main.go
index af302f1ff..dad62e9f7 100644
--- a/main.go
+++ b/main.go
@@ -2417,6 +2417,7 @@ var _ kv.LeaseReadableCoordinator = (*startupGatedCoordinator)(nil)
 var _ kv.AllGroupsLeaseReadableCoordinator = (*startupGatedCoordinator)(nil)
 var _ kv.GroupRoutableCoordinator = (*startupGatedCoordinator)(nil)
 var _ kv.TimestampAllocatorProvider = (*startupGatedCoordinator)(nil)
+var _ kv.ConfiguredTimestampAllocatorProvider = (*startupGatedCoordinator)(nil)
 var _ kv.AppliedReadTimestampVoucher = (*startupGatedCoordinator)(nil)
 
 func (c startupGatedCoordinator) Dispatch(ctx context.Context, reqs *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
@@ -2459,6 +2460,11 @@ func (c startupGatedCoordinator) Clock() *kv.HLC {
 }
 
 func (c startupGatedCoordinator) TimestampAllocator() kv.TimestampAllocator {
+	alloc, _ := kv.TimestampAllocatorThrough(c.inner)
+	return alloc
+}
+
+func (c startupGatedCoordinator) ConfiguredTimestampAllocator() kv.TimestampAllocator {
 	alloc, _ := kv.ConfiguredTimestampAllocatorThrough(c.inner)
 	return alloc
 }
diff --git a/main_tso_routing_test.go b/main_tso_routing_test.go
index db4e9dcc3..d56044534 100644
--- a/main_tso_routing_test.go
+++ b/main_tso_routing_test.go
@@ -9,10 +9,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/bootjp/elastickv/adapter"
 	"github.com/bootjp/elastickv/distribution"
 	"github.com/bootjp/elastickv/internal/raftengine"
 	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
+	pb "github.com/bootjp/elastickv/proto"
 	"github.com/stretchr/testify/require"
 )
 
@@ -196,6 +198,46 @@ func TestInternalTimestampOptionsPreservesTSOThroughStartupGate(t *testing.T) {
 	require.Empty(t, internalTimestampOptions(legacy))
 }
 
+func TestInternalForwardUsesRuntimeAllocatorAfterModeReload(t *testing.T) {
+	t.Parallel()
+
+	setTSOModeFlags(t, false, false)
+	path := filepath.Join(t.TempDir(), "tso-mode")
+	require.NoError(t, os.WriteFile(path, []byte("legacy\n"), 0o600))
+	*tsoModeFile = path
+
+	clock := kv.NewHLC()
+	clock.SetPhysicalCeiling(time.Now().Add(time.Minute).UnixMilli())
+	fsm := kv.NewTSOStateMachine(clock)
+	engine := &mainTSOEngine{state: raftengine.StateLeader, tsoState: fsm}
+	groups := map[uint64]*kv.ShardGroup{
+		dedicatedTSORaftGroupID: {Engine: engine, TSOState: fsm},
+	}
+	coord := newMainTSOCoordinator(clock, groups)
+	wiring, err := configureCoordinatorTSO(coord, groups, mainTSOFloorProvider{})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, wiring.Close()) })
+	require.Equal(t, kv.TSOModeLegacy, wiring.runtimeController.CurrentMode())
+
+	gated := startupGatedCoordinator{inner: coord, gate: &startupPublicKVGate{}}
+	opts := internalTimestampOptions(gated)
+	require.Len(t, opts, 1)
+	require.NoError(t, wiring.runtimeController.ApplyMode(kv.TSOModeShadow))
+	require.NoError(t, wiring.runtimeController.ApplyMode(kv.TSOModeCutover))
+
+	txn := &mainForwardTxn{}
+	internal := adapter.NewInternalWithEngine(txn, engine, nil, nil, opts...)
+	reqs := []*pb.Request{{
+		Mutations: []*pb.Mutation{{Op: pb.Op_PUT, Key: []byte("k"), Value: []byte("v")}},
+	}}
+	resp, err := internal.Forward(context.Background(), &pb.ForwardRequest{Requests: reqs})
+	require.NoError(t, err)
+	require.True(t, resp.GetSuccess())
+	require.Len(t, txn.requests, 1)
+	require.Greater(t, txn.requests[0].GetTs(), uint64(1),
+		"timestamp 1 would mean the long-lived Internal server lost the runtime allocator and used its nil-clock fallback")
+}
+
 func TestConfigureCoordinatorTSOModeFileStartsRuntimeController(t *testing.T) {
 	setTSOModeFlags(t, false, false)
 	path := filepath.Join(t.TempDir(), "tso-mode")
@@ -291,6 +333,19 @@ func (a *mainTimestampAllocator) Next(context.Context) (uint64, error) {
 	return a.next, nil
 }
 
+type mainForwardTxn struct {
+	requests []*pb.Request
+}
+
+func (t *mainForwardTxn) Commit(_ context.Context, reqs []*pb.Request) (*kv.TransactionResponse, error) {
+	t.requests = reqs
+	return &kv.TransactionResponse{CommitIndex: 1}, nil
+}
+
+func (t *mainForwardTxn) Abort(context.Context, []*pb.Request) (*kv.TransactionResponse, error) {
+	return &kv.TransactionResponse{}, nil
+}
+
 func (mainTSOFloorProvider) GlobalCommittedTimestampFloor(context.Context) (uint64, error) {
 	return 0, nil
 }

From 2402e770c465c654be0893bb466d1bf81af7e375 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Thu, 23 Jul 2026 18:58:29 +0900
Subject: [PATCH 5/5] tso: cover runtime allocator reload

---
 main_tso_routing_test.go | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/main_tso_routing_test.go b/main_tso_routing_test.go
index d56044534..7d0ea5ec6 100644
--- a/main_tso_routing_test.go
+++ b/main_tso_routing_test.go
@@ -222,11 +222,13 @@ func TestInternalForwardUsesRuntimeAllocatorAfterModeReload(t *testing.T) {
 	gated := startupGatedCoordinator{inner: coord, gate: &startupPublicKVGate{}}
 	opts := internalTimestampOptions(gated)
 	require.Len(t, opts, 1)
-	require.NoError(t, wiring.runtimeController.ApplyMode(kv.TSOModeShadow))
-	require.NoError(t, wiring.runtimeController.ApplyMode(kv.TSOModeCutover))
 
 	txn := &mainForwardTxn{}
 	internal := adapter.NewInternalWithEngine(txn, engine, nil, nil, opts...)
+
+	require.NoError(t, wiring.runtimeController.ApplyMode(kv.TSOModeShadow))
+	require.NoError(t, wiring.runtimeController.ApplyMode(kv.TSOModeCutover))
+
 	reqs := []*pb.Request{{
 		Mutations: []*pb.Mutation{{Op: pb.Op_PUT, Key: []byte("k"), Value: []byte("v")}},
 	}}