diff --git a/server/internal/scheduler/orchestrate/deps.go b/server/internal/scheduler/orchestrate/deps.go
new file mode 100644
index 0000000..2dba0cb
--- /dev/null
+++ b/server/internal/scheduler/orchestrate/deps.go
@@ -0,0 +1,200 @@
+// Package orchestrate implements the replacement orchestration state machine
+// (task 15E). It is driven every 30 s by the OrchestrateLoop (task 15H).
+//
+// Key components:
+//   - Replacer.Tick   — drains the 15D replace queue and advances in-flight records.
+//   - Grayscale.Advance — gradually ramps new/recovered node weight 10→25→50→75→100.
+package orchestrate
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/scheduler/probe"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Phase constants
+// ─────────────────────────────────────────────────────────────────────────────
+
+// Phase is the orchestration phase of a replacement record.
+type Phase string
+
+const (
+	PhasePending     Phase = "pending"      // waiting for breaker / quota approval
+	PhaseCreating    Phase = "creating"     // CreateNode called; waiting for new node
+	PhaseProbing     Phase = "probing"      // waiting for consecutive probe passes
+	PhaseActivating  Phase = "activating"   // promoting new node; starting grayscale
+	PhaseDrainingOld Phase = "draining_old" // destroying old (already-down) node
+	PhaseDone        Phase = "done"         // finished successfully
+	PhaseFailed      Phase = "failed"       // terminal failure; human review needed
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Tunable constants
+// ─────────────────────────────────────────────────────────────────────────────
+
+const (
+	// MaxAttempts is the maximum number of create+probe tries before marking failed.
+	MaxAttempts = 3
+
+	// ProbeCyclesRequired is the number of consecutive Tick cycles with a passing
+	// probe snapshot required before the new node is promoted to "up".
+	ProbeCyclesRequired = 2
+
+	// ProbeTimeout is the maximum time allowed in the probing phase per attempt.
+	ProbeTimeout = 15 * time.Minute
+
+	// GrayscaleInterval is the time between successive weight ramp steps.
+	GrayscaleInterval = 6 * time.Hour
+)
+
+// GrayscaleWeights is the weight ladder for the warmup ramp.
+// New / recovered nodes start at 10 and advance every GrayscaleInterval.
+var GrayscaleWeights = []int{10, 25, 50, 75, 100}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Data types
+// ─────────────────────────────────────────────────────────────────────────────
+
+// NodeSpec describes the replacement node to be provisioned.
+type NodeSpec struct {
+	Tier       string
+	Region     string
+	Role       string
+	ProviderID string // preferred provider for this attempt
+	RealitySNI string // SNI rotation
+	RealityPBK string
+	HY2Port    int
+	NameZH     string
+	NameEn     string
+	Tags       []string
+}
+
+// NodeInfo holds the properties of an existing node needed to build a
+// replacement spec or to look up tier / region for the breaker.
+type NodeInfo struct {
+	ID         string
+	Tier       string
+	Region     string
+	Role       string
+	ProviderID string
+	RealitySNI string
+	RealityPBK string
+	HY2Port    int
+	NameZH     string
+	NameEn     string
+	Tags       []string
+}
+
+// ProviderInfo describes a cloud provider available for provisioning.
+type ProviderInfo struct {
+	ID      string   // provider identifier
+	Regions []string // supported regions (empty means all regions)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Service interfaces
+// ─────────────────────────────────────────────────────────────────────────────
+
+// ProvisionService is the #14 provisioning interface.
+// The real implementation is provided by task #14; a mock is used in tests.
+type ProvisionService interface {
+	// CreateNode provisions a new node. idempotencyKey makes the call crash-safe:
+	// replaying the same key returns the already-created node without booting again.
+	CreateNode(ctx context.Context, spec NodeSpec, idempotencyKey string) (nodeID string, err error)
+
+	// DestroyNode tears down a node and releases its IP.
+	DestroyNode(ctx context.Context, nodeID string) error
+
+	// RotateIP swaps the elastic IP on an existing node without re-creating it.
+	RotateIP(ctx context.Context, nodeID string) (newNodeID string, err error)
+
+	// ListProviders returns providers available for the given tier and region.
+	// The returned slice is ordered; callers use rotation for provider selection.
+	ListProviders(ctx context.Context, tier, region string) ([]ProviderInfo, error)
+}
+
+// LifecycleService provides node lifecycle management (real impl: task #5).
+type LifecycleService interface {
+	// GetNode returns the current properties of the node, or nil if not found.
+	GetNode(ctx context.Context, nodeID string) (*NodeInfo, error)
+
+	// TransitionStatus performs an optimistic-lock status transition.
+	// Returns (1, nil) on success, (0, nil) on lock conflict.
+	TransitionStatus(ctx context.Context, nodeID string, from, to string, detail map[string]any) (int, error)
+
+	// SetWeight updates the routing weight for the node.
+	SetWeight(ctx context.Context, nodeID string, weight int) error
+
+	// BumpVersion increments the global directory version so clients re-fetch.
+	BumpVersion(ctx context.Context) error
+
+	// WriteAuditLog records an audit trail entry.
+	WriteAuditLog(ctx context.Context, actor, action, target, meta string) error
+}
+
+// ProbeSnapshotter reads the most-recent probe snapshots for a node (15A store).
+type ProbeSnapshotter interface {
+	SnapshotsByNode(ctx context.Context, nodeID string) (map[string]probe.ProbeSnapshot, error)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Breaker (15F stub)
+// ─────────────────────────────────────────────────────────────────────────────
+
+// Breaker is the 15F circuit-breaker interface.
+type Breaker interface {
+	// Allow returns true if a new replacement is permitted to proceed.
+	Allow(tier, region string) bool
+
+	// Record increments the failure counter after a replaced node has been destroyed.
+	Record(tier, region string)
+}
+
+// StubBreaker always permits replacements.
+// TODO(15F): replace with the real Breaker once task 15F is implemented.
+type StubBreaker struct{}
+
+// Allow implements Breaker; always returns true.
+func (StubBreaker) Allow(_, _ string) bool { return true }
+
+// Record implements Breaker; no-op until 15F is implemented.
+func (StubBreaker) Record(_, _ string) {}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Notifier (15G stub)
+// ─────────────────────────────────────────────────────────────────────────────
+
+// Notifier is the 15G alerting interface.
+type Notifier interface {
+	NotifyFault(ctx context.Context, nodeID, reason string) error
+}
+
+// LogNotifier logs faults via slog. Used when no real notifier is wired.
+type LogNotifier struct{}
+
+// NotifyFault implements Notifier.
+func (LogNotifier) NotifyFault(_ context.Context, nodeID, reason string) error {
+	slog.Warn("orchestrate: replacement failed — manual review required",
+		"node_id", nodeID,
+		"reason", reason,
+	)
+	return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Clock (for testability)
+// ─────────────────────────────────────────────────────────────────────────────
+
+// Clock abstracts wall-clock time so tests can fast-forward without sleeping.
+type Clock interface {
+	Now() time.Time
+}
+
+// RealClock is the production clock.
+type RealClock struct{}
+
+// Now implements Clock.
+func (RealClock) Now() time.Time { return time.Now().UTC() }
diff --git a/server/internal/scheduler/orchestrate/grayscale.go b/server/internal/scheduler/orchestrate/grayscale.go
new file mode 100644
index 0000000..ea87fa5
--- /dev/null
+++ b/server/internal/scheduler/orchestrate/grayscale.go
@@ -0,0 +1,183 @@
+package orchestrate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Redis key namespace for grayscale records
+// ─────────────────────────────────────────────────────────────────────────────
+
+const (
+	// grayKeyPrefix is the namespace for per-node grayscale warm-up records.
+	// Full key: sched:gray:{nodeID}
+	grayKeyPrefix = "sched:gray:"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// GrayscaleRecord
+// ─────────────────────────────────────────────────────────────────────────────
+
+// GrayscaleRecord is the Redis-persisted state of a node in weight warm-up.
+// Key: sched:gray:{nodeID}  Value: JSON-encoded GrayscaleRecord.
+type GrayscaleRecord struct {
+	NodeID         string    `json:"nodeId"`
+	CurrentWeight  int       `json:"currentWeight"`
+	StartedAt      time.Time `json:"startedAt"`
+	LastAdvancedAt time.Time `json:"lastAdvancedAt"`
+}
+
+// startGrayscale creates a new GrayscaleRecord for nodeID at weight 10 (the
+// initial weight). It is called from stepActivating immediately after the node
+// is transitioned to "up".
+func startGrayscale(ctx context.Context, rdb *redis.Client, clock Clock, nodeID string) error {
+	now := clock.Now()
+	rec := GrayscaleRecord{
+		NodeID:         nodeID,
+		CurrentWeight:  GrayscaleWeights[0], // 10
+		StartedAt:      now,
+		LastAdvancedAt: now,
+	}
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	return rdb.Set(ctx, grayKeyPrefix+nodeID, data, 0).Err()
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Grayscale
+// ─────────────────────────────────────────────────────────────────────────────
+
+// Grayscale advances the weight ramp for all nodes currently in warm-up.
+// It is called by the CapacityLoop (task 15H); the implementation lives here
+// (task 15E) while wiring is done in 15H.
+//
+// Weight ladder: 10 → 25 → 50 → 75 → 100, one step every 6 h.
+// When weight 100 is reached, the grayscale record is deleted.
+type Grayscale struct {
+	rdb   *redis.Client
+	lc    LifecycleService
+	clock Clock
+}
+
+// NewGrayscale creates a Grayscale with the given dependencies.
+// If clock is nil, RealClock is used.
+func NewGrayscale(rdb *redis.Client, lc LifecycleService, clock Clock) *Grayscale {
+	if clock == nil {
+		clock = RealClock{}
+	}
+	return &Grayscale{rdb: rdb, lc: lc, clock: clock}
+}
+
+// Advance scans all sched:gray:* keys and advances any node whose
+// LastAdvancedAt is at least GrayscaleInterval (6 h) in the past.
+func (g *Grayscale) Advance(ctx context.Context) error {
+	keys, err := g.scanGrayKeys(ctx)
+	if err != nil {
+		return fmt.Errorf("grayscale: scan keys: %w", err)
+	}
+
+	for _, key := range keys {
+		if advErr := g.advanceNode(ctx, key); advErr != nil {
+			slog.Error("grayscale: advance node", "key", key, "error", advErr)
+			// continue; one error must not halt other nodes
+		}
+	}
+	return nil
+}
+
+// advanceNode loads the record at key, advances the weight if the interval has
+// elapsed, calls SetWeight, writes an audit log, and deletes the record when
+// the final weight (100) is reached.
+func (g *Grayscale) advanceNode(ctx context.Context, key string) error {
+	val, err := g.rdb.Get(ctx, key).Result()
+	if err == redis.Nil {
+		return nil // already deleted
+	}
+	if err != nil {
+		return fmt.Errorf("grayscale: get %s: %w", key, err)
+	}
+	var rec GrayscaleRecord
+	if err := json.Unmarshal([]byte(val), &rec); err != nil {
+		return fmt.Errorf("grayscale: unmarshal %s: %w", key, err)
+	}
+
+	now := g.clock.Now()
+	if now.Sub(rec.LastAdvancedAt) < GrayscaleInterval {
+		return nil // interval not elapsed yet
+	}
+
+	// Find the next weight in the ladder.
+	nextWeight, ok := nextGrayscaleWeight(rec.CurrentWeight)
+	if !ok {
+		// Already at maximum; remove the record.
+		_ = g.rdb.Del(ctx, key).Err()
+		return nil
+	}
+
+	// Apply the new weight.
+	if err := g.lc.SetWeight(ctx, rec.NodeID, nextWeight); err != nil {
+		return fmt.Errorf("grayscale: set weight %d for %s: %w", nextWeight, rec.NodeID, err)
+	}
+
+	// Audit trail.
+	meta := fmt.Sprintf(`{"from":%d,"to":%d,"node":%q}`, rec.CurrentWeight, nextWeight, rec.NodeID)
+	_ = g.lc.WriteAuditLog(ctx, "grayscale", "weight_advanced", "node:"+rec.NodeID, meta)
+
+	slog.Info("grayscale: weight advanced",
+		"node", rec.NodeID,
+		"from", rec.CurrentWeight,
+		"to", nextWeight,
+	)
+
+	if nextWeight >= GrayscaleWeights[len(GrayscaleWeights)-1] {
+		// Final weight reached: remove the grayscale record.
+		_ = g.rdb.Del(ctx, key).Err()
+		return nil
+	}
+
+	// Persist updated record.
+	rec.CurrentWeight = nextWeight
+	rec.LastAdvancedAt = now
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	return g.rdb.Set(ctx, key, data, 0).Err()
+}
+
+// nextGrayscaleWeight returns the next weight after current in GrayscaleWeights.
+// Returns (0, false) if current is already at or beyond the final weight.
+func nextGrayscaleWeight(current int) (int, bool) {
+	for i, w := range GrayscaleWeights {
+		if w == current && i+1 < len(GrayscaleWeights) {
+			return GrayscaleWeights[i+1], true
+		}
+	}
+	return 0, false
+}
+
+// scanGrayKeys returns all Redis keys matching sched:gray:*.
+func (g *Grayscale) scanGrayKeys(ctx context.Context) ([]string, error) {
+	var keys []string
+	iter := g.rdb.Scan(ctx, 0, grayKeyPrefix+"*", 0).Iterator()
+	for iter.Next(ctx) {
+		k := iter.Val()
+		// Skip any keys that don't have the expected format (e.g. the index key).
+		if strings.HasPrefix(k, grayKeyPrefix) {
+			keys = append(keys, k)
+		}
+	}
+	if err := iter.Err(); err != nil {
+		return nil, err
+	}
+	return keys, nil
+}
diff --git a/server/internal/scheduler/orchestrate/replacer.go b/server/internal/scheduler/orchestrate/replacer.go
new file mode 100644
index 0000000..ff97352
--- /dev/null
+++ b/server/internal/scheduler/orchestrate/replacer.go
@@ -0,0 +1,625 @@
+package orchestrate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/wangjia/pangolin/server/internal/scheduler/probe"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Redis key constants
+// ─────────────────────────────────────────────────────────────────────────────
+
+const (
+	// replaceQueueKey is the Redis list populated by 15D when a node is confirmed
+	// blocked. We read from the right (RPOP) for FIFO ordering; 15D writes with LPUSH.
+	replaceQueueKey = "detect:replace:queue"
+
+	// replaceKeyPrefix is the namespace for per-replacement orchestration records.
+	// Full key: sched:replace:{uuid}
+	replaceKeyPrefix = "sched:replace:"
+
+	// replaceIndexKey is a Redis set holding all in-flight replacement UUIDs.
+	replaceIndexKey = "sched:replace:index"
+
+	// replaceTTL is how long a terminal (done / failed) record is kept for auditing.
+	replaceTTL = 7 * 24 * time.Hour
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Persistent record types
+// ─────────────────────────────────────────────────────────────────────────────
+
+// ReplaceRecord is the crash-recoverable orchestration state stored in Redis.
+// Key: sched:replace:{uuid}  Value: JSON-encoded ReplaceRecord.
+//
+// All phase transitions read the record first and write it last; a crashed
+// process resumes from the persisted state without repeating already-executed
+// actions (CreateNode is guarded by an idempotency key; DestroyNode is
+// idempotent by design).
+type ReplaceRecord struct {
+	Phase             Phase     `json:"phase"`
+	OldNode           string    `json:"oldNode"`                    // node ID from the 15D replace queue
+	NewNode           string    `json:"newNode,omitempty"`           // new node ID, set after CreateNode
+	CurrentProviderID string    `json:"currentProviderId,omitempty"` // provider for the current attempt
+	Attempts          int       `json:"attempts"`                    // how many create+probe attempts so far
+	ProviderTried     []string  `json:"providerTried"`               // provider IDs from failed attempts
+	ProbeStreak       int       `json:"probeStreak"`                 // consecutive passing probe Ticks
+	PhaseStartedAt    time.Time `json:"phaseStartedAt"`             // when the current phase began
+	UpdatedAt         time.Time `json:"updatedAt"`
+}
+
+// queueEntry is the JSON shape pushed by 15D (detect/engine.go pushReplaceQueue).
+type queueEntry struct {
+	NodeID          string `json:"nodeId"`
+	ReplacementUUID string `json:"replacementUuid"`
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Replacer
+// ─────────────────────────────────────────────────────────────────────────────
+
+// Replacer drives the replacement orchestration state machine.
+// It is called every 30 s by the OrchestrateLoop (task 15H).
+type Replacer struct {
+	rdb      *redis.Client
+	prov     ProvisionService
+	lc       LifecycleService
+	snaps    ProbeSnapshotter
+	breaker  Breaker
+	notifier Notifier
+	clock    Clock
+}
+
+// Config holds all dependencies for NewReplacer.
+type Config struct {
+	RDB      *redis.Client
+	Prov     ProvisionService
+	LC       LifecycleService
+	Snaps    ProbeSnapshotter
+	Breaker  Breaker
+	Notifier Notifier
+	Clock    Clock
+}
+
+// NewReplacer constructs a Replacer. Nil optional deps are replaced with stubs.
+func NewReplacer(cfg Config) *Replacer {
+	if cfg.Breaker == nil {
+		cfg.Breaker = StubBreaker{}
+	}
+	if cfg.Notifier == nil {
+		cfg.Notifier = LogNotifier{}
+	}
+	if cfg.Clock == nil {
+		cfg.Clock = RealClock{}
+	}
+	return &Replacer{
+		rdb:      cfg.RDB,
+		prov:     cfg.Prov,
+		lc:       cfg.LC,
+		snaps:    cfg.Snaps,
+		breaker:  cfg.Breaker,
+		notifier: cfg.Notifier,
+		clock:    cfg.Clock,
+	}
+}
+
+// Tick drains the 15D replace queue and advances every in-flight record one
+// step. It is designed to be called every 30 s and to be idempotent across
+// crashes: all state is persisted in Redis before any external action, so a
+// restart resumes from the saved phase.
+func (r *Replacer) Tick(ctx context.Context) error {
+	if err := r.drainQueue(ctx); err != nil {
+		// Non-fatal: log and continue so in-flight records still advance.
+		slog.Error("orchestrate: drain replace queue", "error", err)
+	}
+	return r.advanceAll(ctx)
+}
+
+// drainQueue pops entries from detect:replace:queue and creates pending records.
+func (r *Replacer) drainQueue(ctx context.Context) error {
+	const maxDrain = 100 // guard against burst
+	for i := 0; i < maxDrain; i++ {
+		raw, err := r.rdb.RPop(ctx, replaceQueueKey).Result()
+		if err == redis.Nil {
+			break // queue empty
+		}
+		if err != nil {
+			return fmt.Errorf("orchestrate: rpop: %w", err)
+		}
+		var entry queueEntry
+		if err := json.Unmarshal([]byte(raw), &entry); err != nil {
+			slog.Error("orchestrate: bad queue entry", "raw", raw, "error", err)
+			continue
+		}
+		if err := r.ensureRecord(ctx, entry.ReplacementUUID, entry.NodeID); err != nil {
+			slog.Error("orchestrate: ensure record",
+				"uuid", entry.ReplacementUUID,
+				"node", entry.NodeID,
+				"error", err,
+			)
+		}
+	}
+	return nil
+}
+
+// ensureRecord creates a pending orchestration record for uuid/oldNode if one
+// does not already exist, then adds uuid to the in-flight index. It is safe
+// to call multiple times (SetNX + SAdd are both idempotent).
+func (r *Replacer) ensureRecord(ctx context.Context, uuid, oldNode string) error {
+	key := replaceKeyPrefix + uuid
+	rec := ReplaceRecord{
+		Phase:          PhasePending,
+		OldNode:        oldNode,
+		ProviderTried:  []string{},
+		PhaseStartedAt: r.clock.Now(),
+		UpdatedAt:      r.clock.Now(),
+	}
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	// SetNX: only stores if the key does not exist, preserving any in-progress record.
+	r.rdb.SetNX(ctx, key, data, 0) //nolint:errcheck // best-effort; SAdd follows
+	return r.rdb.SAdd(ctx, replaceIndexKey, uuid).Err()
+}
+
+// advanceAll loads every UUID from the in-flight index and advances each record.
+func (r *Replacer) advanceAll(ctx context.Context) error {
+	uuids, err := r.rdb.SMembers(ctx, replaceIndexKey).Result()
+	if err != nil {
+		return fmt.Errorf("orchestrate: smembers index: %w", err)
+	}
+	for _, uuid := range uuids {
+		if advErr := r.advanceRecord(ctx, uuid); advErr != nil {
+			slog.Error("orchestrate: advance record", "uuid", uuid, "error", advErr)
+			// continue; one failure must not halt other replacements
+		}
+	}
+	return nil
+}
+
+// advanceRecord loads a single record and advances it by one phase step.
+func (r *Replacer) advanceRecord(ctx context.Context, uuid string) error {
+	rec, err := r.loadRecord(ctx, uuid)
+	if err != nil {
+		return err
+	}
+	if rec == nil {
+		// Key expired or was deleted; clean up the index entry.
+		_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
+		return nil
+	}
+
+	switch rec.Phase {
+	case PhasePending:
+		return r.stepPending(ctx, uuid, rec)
+	case PhaseCreating:
+		return r.stepCreating(ctx, uuid, rec)
+	case PhaseProbing:
+		return r.stepProbing(ctx, uuid, rec)
+	case PhaseActivating:
+		return r.stepActivating(ctx, uuid, rec)
+	case PhaseDrainingOld:
+		return r.stepDrainingOld(ctx, uuid, rec)
+	case PhaseDone, PhaseFailed:
+		// Terminal: the record exists only for audit; remove from active index.
+		_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
+		return nil
+	default:
+		return fmt.Errorf("orchestrate: unknown phase %q for %s", rec.Phase, uuid)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Phase step functions
+// ─────────────────────────────────────────────────────────────────────────────
+
+// stepPending checks the circuit breaker and watermark/quota, then advances
+// to creating. If the breaker blocks, the record stays pending (retried next Tick).
+func (r *Replacer) stepPending(ctx context.Context, uuid string, rec *ReplaceRecord) error {
+	nodeInfo, err := r.lc.GetNode(ctx, rec.OldNode)
+	if err != nil {
+		return fmt.Errorf("orchestrate: get node %s: %w", rec.OldNode, err)
+	}
+	if nodeInfo == nil {
+		return fmt.Errorf("orchestrate: old node %s not found", rec.OldNode)
+	}
+
+	// Circuit-breaker check (15F).
+	if !r.breaker.Allow(nodeInfo.Tier, nodeInfo.Region) {
+		slog.Info("orchestrate: breaker blocked replacement",
+			"uuid", uuid, "tier", nodeInfo.Tier, "region", nodeInfo.Region)
+		return nil // stay pending; retry next Tick
+	}
+
+	// Watermark / quota check — stub (always passes).
+	// TODO(15F): implement real capacity-quota guard here.
+
+	rec.Phase = PhaseCreating
+	rec.PhaseStartedAt = r.clock.Now()
+	rec.UpdatedAt = r.clock.Now()
+	return r.saveRecord(ctx, uuid, rec)
+}
+
+// stepCreating calls CreateNode (idempotent via idempotency key) and advances
+// to probing. If NewNode is already set (crash recovery after CreateNode
+// succeeded but before the record was saved with phase=probing), CreateNode is
+// NOT called again.
+func (r *Replacer) stepCreating(ctx context.Context, uuid string, rec *ReplaceRecord) error {
+	if rec.NewNode == "" {
+		nodeInfo, err := r.lc.GetNode(ctx, rec.OldNode)
+		if err != nil {
+			return fmt.Errorf("orchestrate: get node %s: %w", rec.OldNode, err)
+		}
+		if nodeInfo == nil {
+			return fmt.Errorf("orchestrate: old node %s not found", rec.OldNode)
+		}
+
+		providerID, err := r.pickProvider(ctx, nodeInfo.Tier, nodeInfo.Region, rec.ProviderTried)
+		if err != nil {
+			return fmt.Errorf("orchestrate: pick provider: %w", err)
+		}
+
+		spec := NodeSpec{
+			Tier:       nodeInfo.Tier,
+			Region:     nodeInfo.Region,
+			Role:       nodeInfo.Role,
+			ProviderID: providerID,
+			RealitySNI: nodeInfo.RealitySNI,
+			RealityPBK: nodeInfo.RealityPBK,
+			HY2Port:    nodeInfo.HY2Port,
+			NameZH:     nodeInfo.NameZH,
+			NameEn:     nodeInfo.NameEn,
+			Tags:       nodeInfo.Tags,
+		}
+		idemKey := idempotencyKey(uuid, rec.Attempts)
+
+		newNodeID, err := r.prov.CreateNode(ctx, spec, idemKey)
+		if err != nil {
+			return fmt.Errorf("orchestrate: create node (attempt %d): %w", rec.Attempts, err)
+		}
+		rec.NewNode = newNodeID
+		rec.CurrentProviderID = providerID
+	}
+
+	// Advance to probing; reset streak and set phase start time.
+	rec.Phase = PhaseProbing
+	rec.ProbeStreak = 0
+	rec.PhaseStartedAt = r.clock.Now()
+	rec.UpdatedAt = r.clock.Now()
+	return r.saveRecord(ctx, uuid, rec)
+}
+
+// stepProbing checks probe snapshots for the new node.
+// On 2 consecutive passing Ticks it advances to activating.
+// On timeout (15 min) or repeated failure it retries with a different provider,
+// or marks the record failed after MaxAttempts.
+func (r *Replacer) stepProbing(ctx context.Context, uuid string, rec *ReplaceRecord) error {
+	now := r.clock.Now()
+
+	// Timeout guard.
+	if now.Sub(rec.PhaseStartedAt) > ProbeTimeout {
+		return r.failProbeAttempt(ctx, uuid, rec, "probe timeout (15 min)")
+	}
+
+	// Read probe snapshots for the new node (from the 15A probe store).
+	snapshots, err := r.snaps.SnapshotsByNode(ctx, rec.NewNode)
+	if err != nil {
+		return fmt.Errorf("orchestrate: snapshots for %s: %w", rec.NewNode, err)
+	}
+
+	if probePass(snapshots) {
+		rec.ProbeStreak++
+	} else {
+		rec.ProbeStreak = 0
+	}
+
+	if rec.ProbeStreak >= ProbeCyclesRequired {
+		// Probe window passed: advance to activating.
+		rec.Phase = PhaseActivating
+		rec.PhaseStartedAt = now
+		rec.UpdatedAt = now
+		return r.saveRecord(ctx, uuid, rec)
+	}
+
+	// Not yet passed: update streak and wait for next Tick.
+	rec.UpdatedAt = now
+	return r.saveRecord(ctx, uuid, rec)
+}
+
+// failProbeAttempt destroys the failed new node and either schedules a retry
+// (with a different provider) or marks the replacement as permanently failed.
+func (r *Replacer) failProbeAttempt(ctx context.Context, uuid string, rec *ReplaceRecord, reason string) error {
+	slog.Warn("orchestrate: probing failed",
+		"uuid", uuid,
+		"new_node", rec.NewNode,
+		"attempt", rec.Attempts,
+		"reason", reason,
+	)
+
+	// Destroy the bad new node (bad IP must not enter the pool).
+	if rec.NewNode != "" {
+		if err := r.prov.DestroyNode(ctx, rec.NewNode); err != nil {
+			slog.Error("orchestrate: destroy failed new node",
+				"uuid", uuid, "node", rec.NewNode, "error", err)
+			// Continue: still mark the attempt as failed.
+		}
+		// Track which provider was tried so the next attempt avoids it.
+		if rec.CurrentProviderID != "" {
+			rec.ProviderTried = appendUnique(rec.ProviderTried, rec.CurrentProviderID)
+		}
+		rec.NewNode = ""
+		rec.CurrentProviderID = ""
+	}
+	rec.Attempts++
+
+	if rec.Attempts >= MaxAttempts {
+		// All attempts exhausted: mark failed and alert.
+		rec.Phase = PhaseFailed
+		rec.UpdatedAt = r.clock.Now()
+		if err := r.saveRecord(ctx, uuid, rec); err != nil {
+			return err
+		}
+		_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
+		_ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err()
+
+		alertReason := fmt.Sprintf("probing failed after %d attempts: %s", rec.Attempts, reason)
+		if notifyErr := r.notifier.NotifyFault(ctx, rec.OldNode, alertReason); notifyErr != nil {
+			slog.Error("orchestrate: notify fault", "uuid", uuid, "error", notifyErr)
+		}
+		slog.Error("orchestrate: replacement permanently failed — manual review required",
+			"uuid", uuid, "old_node", rec.OldNode, "attempts", rec.Attempts)
+		return nil
+	}
+
+	// Schedule retry: go back to creating (different provider picked next call).
+	rec.Phase = PhaseCreating
+	rec.ProbeStreak = 0
+	rec.PhaseStartedAt = r.clock.Now()
+	rec.UpdatedAt = r.clock.Now()
+	return r.saveRecord(ctx, uuid, rec)
+}
+
+// stepActivating sets initial weight, promotes the new node from probing→up,
+// bumps the directory version, and registers grayscale warm-up.
+func (r *Replacer) stepActivating(ctx context.Context, uuid string, rec *ReplaceRecord) error {
+	// Set initial weight for grayscale ramp.
+	if err := r.lc.SetWeight(ctx, rec.NewNode, GrayscaleWeights[0]); err != nil {
+		return fmt.Errorf("orchestrate: set initial weight: %w", err)
+	}
+
+	// Optimistic-lock transition: probing → up.
+	detail := map[string]any{
+		"from":             "probing",
+		"to":               "up",
+		"replacement_uuid": uuid,
+		"old_node":         rec.OldNode,
+	}
+	affected, err := r.lc.TransitionStatus(ctx, rec.NewNode, "probing", "up", detail)
+	if err != nil {
+		return fmt.Errorf("orchestrate: transition probing→up: %w", err)
+	}
+	if affected == 0 {
+		// Lock conflict: another writer changed the state. Retry next Tick.
+		slog.Info("orchestrate: probing→up conflict, will retry",
+			"uuid", uuid, "new_node", rec.NewNode)
+		return nil
+	}
+
+	// Bump directory version so clients re-fetch the updated node list.
+	if err := r.lc.BumpVersion(ctx); err != nil {
+		slog.Error("orchestrate: bump version", "uuid", uuid, "error", err)
+	}
+
+	// Register the new node for grayscale warm-up.
+	if err := startGrayscale(ctx, r.rdb, r.clock, rec.NewNode); err != nil {
+		slog.Error("orchestrate: start grayscale", "uuid", uuid, "node", rec.NewNode, "error", err)
+	}
+
+	rec.Phase = PhaseDrainingOld
+	rec.PhaseStartedAt = r.clock.Now()
+	rec.UpdatedAt = r.clock.Now()
+	return r.saveRecord(ctx, uuid, rec)
+}
+
+// stepDrainingOld destroys the old node, records the event in the breaker, and
+// finalises the replacement.
+func (r *Replacer) stepDrainingOld(ctx context.Context, uuid string, rec *ReplaceRecord) error {
+	// Look up old node for breaker recording (best-effort; node may be gone).
+	nodeInfo, _ := r.lc.GetNode(ctx, rec.OldNode)
+
+	// Destroy the old (already-down) node and release its IP.
+	if err := r.prov.DestroyNode(ctx, rec.OldNode); err != nil {
+		return fmt.Errorf("orchestrate: destroy old node %s: %w", rec.OldNode, err)
+	}
+
+	// Inform the 15F circuit breaker of this successful replacement.
+	if nodeInfo != nil {
+		r.breaker.Record(nodeInfo.Tier, nodeInfo.Region)
+	}
+
+	rec.Phase = PhaseDone
+	rec.UpdatedAt = r.clock.Now()
+	if err := r.saveRecord(ctx, uuid, rec); err != nil {
+		return err
+	}
+
+	slog.Info("orchestrate: replacement complete",
+		"uuid", uuid, "old_node", rec.OldNode, "new_node", rec.NewNode)
+
+	// Write audit log.
+	_ = r.lc.WriteAuditLog(ctx, "orchestrate", "replacement_done",
+		"node:"+rec.OldNode,
+		fmt.Sprintf(`{"uuid":%q,"new_node":%q}`, uuid, rec.NewNode),
+	)
+
+	// Remove from active index; set TTL for 7-day audit retention.
+	_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
+	_ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err()
+	return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Probe-pass check
+// ─────────────────────────────────────────────────────────────────────────────
+
+// probePass returns true when the snapshot set indicates the new node is healthy:
+//   - Domestic (CN): ≥ 2/3 ISPs are passing.
+//   - Overseas: at least one vantage is present and all overseas vantages pass L1.
+func probePass(snapshots map[string]probe.ProbeSnapshot) bool {
+	if len(snapshots) == 0 {
+		return false
+	}
+
+	domesticISPs := make(map[string]bool) // ISP name → failed?
+	overseasTotal := 0
+	overseasOK := 0
+
+	for _, snap := range snapshots {
+		v := snap.Vantage
+		rpt := snap.Report
+		switch {
+		case v.Country == "CN":
+			// Normalise "3rd-" prefix (third-party vantages share ISP grouping).
+			isp := strings.TrimPrefix(v.ISP, "3rd-")
+			failed := isSnapshotFailed(rpt)
+			if prev, seen := domesticISPs[isp]; seen {
+				domesticISPs[isp] = prev || failed
+			} else {
+				domesticISPs[isp] = failed
+			}
+		case v.Country != "":
+			overseasTotal++
+			if rpt.L1.OK {
+				overseasOK++
+			}
+		}
+	}
+
+	if len(domesticISPs) == 0 || overseasTotal == 0 {
+		return false // insufficient probe data
+	}
+
+	// Domestic: ≥ 2/3 ISPs must be passing (not failed).
+	passISPs := 0
+	for _, failed := range domesticISPs {
+		if !failed {
+			passISPs++
+		}
+	}
+	total := len(domesticISPs)
+	domesticOK := passISPs*3 >= total*2
+
+	// Overseas: every vantage must pass L1.
+	overseasPass := overseasOK == overseasTotal
+
+	return domesticOK && overseasPass
+}
+
+// isSnapshotFailed mirrors detect/signals.go isReportFailed: check the highest
+// available layer (L3 > L2 > L1).
+func isSnapshotFailed(rpt probe.NodeReport) bool {
+	if rpt.L3 != nil {
+		return !rpt.L3.OK
+	}
+	if !rpt.L1.OK {
+		return true
+	}
+	if rpt.L2 != nil && !rpt.L2.OK {
+		return true
+	}
+	return false
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Provider selection
+// ─────────────────────────────────────────────────────────────────────────────
+
+// pickProvider selects a provider for a new attempt. It prefers providers NOT
+// in previouslyTried so each retry rotates to a different vendor.
+func (r *Replacer) pickProvider(ctx context.Context, tier, region string, previouslyTried []string) (string, error) {
+	providers, err := r.prov.ListProviders(ctx, tier, region)
+	if err != nil {
+		return "", fmt.Errorf("orchestrate: list providers: %w", err)
+	}
+	if len(providers) == 0 {
+		return "", fmt.Errorf("orchestrate: no providers for tier=%s region=%s", tier, region)
+	}
+
+	tried := make(map[string]bool, len(previouslyTried))
+	for _, id := range previouslyTried {
+		tried[id] = true
+	}
+
+	for _, p := range providers {
+		if !tried[p.ID] {
+			return p.ID, nil
+		}
+	}
+	// All providers have been tried; fall back to the first one.
+	return providers[0].ID, nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Redis helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+// loadRecord reads and unmarshals a ReplaceRecord. Returns (nil, nil) if absent.
+func (r *Replacer) loadRecord(ctx context.Context, uuid string) (*ReplaceRecord, error) {
+	val, err := r.rdb.Get(ctx, replaceKeyPrefix+uuid).Result()
+	if err == redis.Nil {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("orchestrate: load %s: %w", uuid, err)
+	}
+	var rec ReplaceRecord
+	if err := json.Unmarshal([]byte(val), &rec); err != nil {
+		return nil, fmt.Errorf("orchestrate: unmarshal %s: %w", uuid, err)
+	}
+	return &rec, nil
+}
+
+// saveRecord marshals and persists rec under sched:replace:{uuid}.
+func (r *Replacer) saveRecord(ctx context.Context, uuid string, rec *ReplaceRecord) error {
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return fmt.Errorf("orchestrate: marshal %s: %w", uuid, err)
+	}
+	if err := r.rdb.Set(ctx, replaceKeyPrefix+uuid, data, 0).Err(); err != nil {
+		return fmt.Errorf("orchestrate: save %s: %w", uuid, err)
+	}
+	return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Misc helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+// idempotencyKey derives the CreateNode idempotency key for the given attempt.
+// Attempt 0 uses the replacement UUID directly; retries append a suffix to
+// ensure each attempt is independently idempotent.
+func idempotencyKey(replacementUUID string, attempt int) string {
+	if attempt == 0 {
+		return replacementUUID
+	}
+	return fmt.Sprintf("%s:retry:%d", replacementUUID, attempt)
+}
+
+// appendUnique appends s to slice only if s is not already present.
+func appendUnique(slice []string, s string) []string {
+	for _, v := range slice {
+		if v == s {
+			return slice
+		}
+	}
+	return append(slice, s)
+}
diff --git a/server/internal/scheduler/orchestrate/replacer_test.go b/server/internal/scheduler/orchestrate/replacer_test.go
new file mode 100644
index 0000000..7fcaf4d
--- /dev/null
+++ b/server/internal/scheduler/orchestrate/replacer_test.go
@@ -0,0 +1,821 @@
+package orchestrate_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+
+	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
+	"github.com/wangjia/pangolin/server/internal/scheduler/probe"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// In-process Redis
+// ─────────────────────────────────────────────────────────────────────────────
+
+func newTestRedis(t *testing.T) (*redis.Client, *miniredis.Miniredis) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	return rdb, mr
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock ProvisionService
+// ─────────────────────────────────────────────────────────────────────────────
+
+type createCall struct {
+	Spec           orchestrate.NodeSpec
+	IdempotencyKey string
+	ReturnedID     string
+}
+
+type mockProvision struct {
+	mu  sync.Mutex
+	seq int
+
+	providers    []orchestrate.ProviderInfo
+	createCalls  []createCall
+	destroyCalls []string
+	createErr    error
+
+	// idem maps idempotency key → nodeID (simulates provider idempotency).
+	idem map[string]string
+}
+
+func newMockProvision(providers ...orchestrate.ProviderInfo) *mockProvision {
+	return &mockProvision{
+		providers: providers,
+		idem:      map[string]string{},
+	}
+}
+
+func (m *mockProvision) CreateNode(_ context.Context, spec orchestrate.NodeSpec, idemKey string) (string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.createErr != nil {
+		return "", m.createErr
+	}
+	// Idempotency replay.
+	if existing, ok := m.idem[idemKey]; ok {
+		m.createCalls = append(m.createCalls, createCall{spec, idemKey, existing})
+		return existing, nil
+	}
+	m.seq++
+	nodeID := fmt.Sprintf("new-node-%d", m.seq)
+	m.idem[idemKey] = nodeID
+	m.createCalls = append(m.createCalls, createCall{spec, idemKey, nodeID})
+	return nodeID, nil
+}
+
+func (m *mockProvision) DestroyNode(_ context.Context, nodeID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.destroyCalls = append(m.destroyCalls, nodeID)
+	return nil
+}
+
+func (m *mockProvision) RotateIP(_ context.Context, _ string) (string, error) { return "", nil }
+
+func (m *mockProvision) ListProviders(_ context.Context, _, _ string) ([]orchestrate.ProviderInfo, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return append([]orchestrate.ProviderInfo{}, m.providers...), nil
+}
+
+func (m *mockProvision) createCount() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return len(m.createCalls)
+}
+
+func (m *mockProvision) destroyCount() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return len(m.destroyCalls)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock LifecycleService
+// ─────────────────────────────────────────────────────────────────────────────
+
+type transitionEvent struct {
+	NodeID, From, To string
+}
+
+type mockLC struct {
+	mu          sync.Mutex
+	nodes       map[string]*orchestrate.NodeInfo
+	weights     map[string]int
+	version     int64
+	auditLogs   []string
+	transitions []transitionEvent
+}
+
+func newMockLC(nodes ...*orchestrate.NodeInfo) *mockLC {
+	m := &mockLC{
+		nodes:   make(map[string]*orchestrate.NodeInfo),
+		weights: make(map[string]int),
+	}
+	for _, n := range nodes {
+		cp := *n
+		m.nodes[n.ID] = &cp
+	}
+	return m
+}
+
+func (m *mockLC) addNode(n *orchestrate.NodeInfo) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	cp := *n
+	m.nodes[n.ID] = &cp
+}
+
+func (m *mockLC) GetNode(_ context.Context, nodeID string) (*orchestrate.NodeInfo, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	n, ok := m.nodes[nodeID]
+	if !ok {
+		return nil, nil
+	}
+	cp := *n
+	return &cp, nil
+}
+
+func (m *mockLC) TransitionStatus(_ context.Context, nodeID, from, to string, _ map[string]any) (int, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.transitions = append(m.transitions, transitionEvent{nodeID, from, to})
+	return 1, nil // always succeed in tests
+}
+
+func (m *mockLC) SetWeight(_ context.Context, nodeID string, weight int) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.weights[nodeID] = weight
+	return nil
+}
+
+func (m *mockLC) BumpVersion(_ context.Context) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.version++
+	return nil
+}
+
+func (m *mockLC) WriteAuditLog(_ context.Context, actor, action, target, _ string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.auditLogs = append(m.auditLogs, actor+"|"+action+"|"+target)
+	return nil
+}
+
+func (m *mockLC) weightOf(nodeID string) int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.weights[nodeID]
+}
+
+func (m *mockLC) versionOf() int64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.version
+}
+
+func (m *mockLC) hasAudit(entry string) bool {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, l := range m.auditLogs {
+		if l == entry {
+			return true
+		}
+	}
+	return false
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock ProbeSnapshotter
+// ─────────────────────────────────────────────────────────────────────────────
+
+type mockSnaps struct {
+	mu   sync.Mutex
+	data map[string]map[string]probe.ProbeSnapshot
+}
+
+func (s *mockSnaps) SnapshotsByNode(_ context.Context, nodeID string) (map[string]probe.ProbeSnapshot, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	d, ok := s.data[nodeID]
+	if !ok {
+		return nil, nil
+	}
+	out := make(map[string]probe.ProbeSnapshot, len(d))
+	for k, v := range d {
+		out[k] = v
+	}
+	return out, nil
+}
+
+func (s *mockSnaps) setPass(nodeID string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.data == nil {
+		s.data = make(map[string]map[string]probe.ProbeSnapshot)
+	}
+	s.data[nodeID] = passingSnapshots()
+}
+
+// passingSnapshots returns a snapshot set where 2/3 domestic ISPs pass + overseas OK.
+func passingSnapshots() map[string]probe.ProbeSnapshot {
+	return map[string]probe.ProbeSnapshot{
+		"CN:Telecom": {
+			Vantage: probe.VantagePoint{Country: "CN", ISP: "ChinaTelecom"},
+			Report: probe.NodeReport{
+				L1: probe.L1Result{OK: true},
+				L3: &probe.L3Result{OK: true},
+			},
+		},
+		"CN:Unicom": {
+			Vantage: probe.VantagePoint{Country: "CN", ISP: "ChinaUnicom"},
+			Report: probe.NodeReport{
+				L1: probe.L1Result{OK: true},
+				L3: &probe.L3Result{OK: true},
+			},
+		},
+		"CN:Mobile": {
+			Vantage: probe.VantagePoint{Country: "CN", ISP: "ChinaMobile"},
+			Report: probe.NodeReport{
+				L1: probe.L1Result{OK: false},
+				L3: &probe.L3Result{OK: false},
+			},
+		},
+		"SG:AWS": {
+			Vantage: probe.VantagePoint{Country: "SG", ISP: "AWS"},
+			Report:  probe.NodeReport{L1: probe.L1Result{OK: true}},
+		},
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock Notifier
+// ─────────────────────────────────────────────────────────────────────────────
+
+type mockNotifier struct {
+	mu    sync.Mutex
+	calls []string
+}
+
+func (n *mockNotifier) NotifyFault(_ context.Context, nodeID, _ string) error {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	n.calls = append(n.calls, nodeID)
+	return nil
+}
+
+func (n *mockNotifier) count() int {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	return len(n.calls)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock Clock
+// ─────────────────────────────────────────────────────────────────────────────
+
+type mockClock struct {
+	mu  sync.Mutex
+	now time.Time
+}
+
+func newMockClock(t time.Time) *mockClock { return &mockClock{now: t} }
+
+func (c *mockClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.now
+}
+
+func (c *mockClock) advance(d time.Duration) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.now = c.now.Add(d)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test harness
+// ─────────────────────────────────────────────────────────────────────────────
+
+type harness struct {
+	rdb      *redis.Client
+	prov     *mockProvision
+	lc       *mockLC
+	snaps    *mockSnaps
+	notifier *mockNotifier
+	clock    *mockClock
+	replacer *orchestrate.Replacer
+}
+
+func newHarness(t *testing.T, nodes ...*orchestrate.NodeInfo) *harness {
+	t.Helper()
+	rdb, _ := newTestRedis(t)
+	prov := newMockProvision(
+		orchestrate.ProviderInfo{ID: "provider-A"},
+		orchestrate.ProviderInfo{ID: "provider-B"},
+		orchestrate.ProviderInfo{ID: "provider-C"},
+	)
+	lc := newMockLC(nodes...)
+	snaps := &mockSnaps{}
+	notifier := &mockNotifier{}
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+
+	r := orchestrate.NewReplacer(orchestrate.Config{
+		RDB:      rdb,
+		Prov:     prov,
+		LC:       lc,
+		Snaps:    snaps,
+		Notifier: notifier,
+		Clock:    clock,
+	})
+	return &harness{rdb: rdb, prov: prov, lc: lc, snaps: snaps, notifier: notifier, clock: clock, replacer: r}
+}
+
+func (h *harness) pushQueue(t *testing.T, oldNodeID, replacementUUID string) {
+	t.Helper()
+	data, _ := json.Marshal(map[string]string{
+		"nodeId":          oldNodeID,
+		"replacementUuid": replacementUUID,
+	})
+	if err := h.rdb.LPush(context.Background(), "detect:replace:queue", data).Err(); err != nil {
+		t.Fatalf("pushQueue: %v", err)
+	}
+}
+
+func (h *harness) tick(t *testing.T) {
+	t.Helper()
+	if err := h.replacer.Tick(context.Background()); err != nil {
+		t.Fatalf("Tick() error: %v", err)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// State-machine phase explanation (for reference):
+//
+//   Tick 1: drainQueue creates pending record → advanceAll: stepPending → creating
+//   Tick 2: stepCreating → CreateNode called → probing
+//   Tick 3: stepProbing (probeStreak=1, need 2)
+//   Tick 4: stepProbing (probeStreak=2) → activating
+//   Tick 5: stepActivating → SetWeight(10)+TransitionStatus+BumpVersion+grayscale → draining_old
+//   Tick 6: stepDrainingOld → DestroyNode(oldNode) → done
+// ─────────────────────────────────────────────────────────────────────────────
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: happy-path replacement (6 ticks)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestHappyPathReplacement(t *testing.T) {
+	const (
+		oldNode         = "old-node-001"
+		replacementUUID = "uuid-happy"
+	)
+	h := newHarness(t,
+		&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg", Role: "entry"},
+	)
+	h.pushQueue(t, oldNode, replacementUUID)
+
+	// Tick 1: pending → creating (CreateNode not yet called).
+	h.tick(t)
+	if h.prov.createCount() != 0 {
+		t.Errorf("tick 1: CreateNode should not be called yet; got %d calls", h.prov.createCount())
+	}
+
+	// Tick 2: creating → CreateNode → probing.
+	h.tick(t)
+	if h.prov.createCount() != 1 {
+		t.Fatalf("tick 2: CreateNode calls = %d; want 1", h.prov.createCount())
+	}
+	call0 := h.prov.createCalls[0]
+	// The idempotency key for attempt 0 must equal the replacement UUID.
+	if call0.IdempotencyKey != replacementUUID {
+		t.Errorf("CreateNode idempotency key = %q; want %q", call0.IdempotencyKey, replacementUUID)
+	}
+
+	newNodeID := call0.ReturnedID
+	// Simulate agent self-registration: add the new node to the lifecycle mock.
+	h.lc.addNode(&orchestrate.NodeInfo{ID: newNodeID, Tier: "free", Region: "hkg"})
+	// Set probe snapshots to PASS for the new node.
+	h.snaps.setPass(newNodeID)
+
+	h.tick(t) // Tick 3: probing streak=1 (< 2 required).
+	h.tick(t) // Tick 4: probing streak=2 → phase saved as activating.
+	h.tick(t) // Tick 5: activating → SetWeight(10) + TransitionStatus(probing→up) + BumpVersion + grayscale → draining_old.
+
+	// After stepActivating:
+	if w := h.lc.weightOf(newNodeID); w != 10 {
+		t.Errorf("new node weight after activating = %d; want 10", w)
+	}
+	if h.lc.versionOf() == 0 {
+		t.Error("BumpVersion was not called after activating")
+	}
+
+	h.tick(t) // Tick 6: draining_old → DestroyNode(oldNode) → done.
+
+	// Old node must be destroyed.
+	oldDestroyed := false
+	for _, id := range h.prov.destroyCalls {
+		if id == oldNode {
+			oldDestroyed = true
+			break
+		}
+	}
+	if !oldDestroyed {
+		t.Errorf("old node %s not destroyed; destroyCalls = %v", oldNode, h.prov.destroyCalls)
+	}
+
+	// Grayscale record must exist for the new node.
+	if _, err := h.rdb.Get(context.Background(), "sched:gray:"+newNodeID).Result(); err == redis.Nil {
+		t.Errorf("grayscale record for %s not found", newNodeID)
+	}
+
+	// Audit log must contain replacement_done.
+	if !h.lc.hasAudit("orchestrate|replacement_done|node:" + oldNode) {
+		t.Errorf("replacement_done audit log not found; got: %v", h.lc.auditLogs)
+	}
+
+	// New node must NOT have been destroyed.
+	for _, id := range h.prov.destroyCalls {
+		if id == newNodeID {
+			t.Errorf("new node %s was incorrectly destroyed", newNodeID)
+		}
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: probe failure exhausts MaxAttempts → failed + alert (7 ticks + 3 clock advances)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestProbeFailMaxAttempts(t *testing.T) {
+	const (
+		oldNode         = "old-node-fail"
+		replacementUUID = "uuid-fail"
+	)
+	h := newHarness(t,
+		&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg", Role: "entry"},
+	)
+	h.pushQueue(t, oldNode, replacementUUID)
+
+	ctx := context.Background()
+
+	// ── Attempt 0 ────────────────────────────────────────────────────────────
+
+	h.tick(t) // Tick 1: pending → creating.
+	h.tick(t) // Tick 2: creating → CreateNode(idem=uuid) → probing.
+
+	if h.prov.createCount() != 1 {
+		t.Fatalf("attempt 0: CreateNode calls = %d; want 1", h.prov.createCount())
+	}
+	if h.prov.createCalls[0].IdempotencyKey != replacementUUID {
+		t.Errorf("attempt 0 idem key = %q; want %q",
+			h.prov.createCalls[0].IdempotencyKey, replacementUUID)
+	}
+
+	newNode0 := h.prov.createCalls[0].ReturnedID
+	h.snaps.setPass(newNode0) // even if snapshots pass, timeout overrides
+
+	h.clock.advance(orchestrate.ProbeTimeout + time.Second) // cause timeout
+
+	h.tick(t) // Tick 3: probing → timeout → DestroyNode(newNode0) → creating (attempts=1).
+
+	if h.prov.destroyCount() != 1 {
+		t.Fatalf("after attempt 0 timeout: DestroyNode calls = %d; want 1", h.prov.destroyCount())
+	}
+	if h.prov.destroyCalls[0] != newNode0 {
+		t.Errorf("attempt 0: expected %s destroyed; got %s", newNode0, h.prov.destroyCalls[0])
+	}
+	// Old node must NOT have been destroyed yet.
+	for _, id := range h.prov.destroyCalls {
+		if id == oldNode {
+			t.Errorf("old node destroyed during attempt 0 (want: only after success)")
+		}
+	}
+
+	// ── Attempt 1 ────────────────────────────────────────────────────────────
+
+	h.tick(t) // Tick 4: creating → CreateNode(idem=uuid:retry:1) → probing.
+
+	if h.prov.createCount() != 2 {
+		t.Fatalf("attempt 1: CreateNode calls = %d; want 2", h.prov.createCount())
+	}
+	call1 := h.prov.createCalls[1]
+	if call1.IdempotencyKey != replacementUUID+":retry:1" {
+		t.Errorf("attempt 1 idem key = %q; want %q:retry:1",
+			call1.IdempotencyKey, replacementUUID)
+	}
+
+	newNode1 := call1.ReturnedID
+	h.snaps.setPass(newNode1)
+	h.clock.advance(orchestrate.ProbeTimeout + time.Second)
+
+	h.tick(t) // Tick 5: probing → timeout → DestroyNode(newNode1) → creating (attempts=2).
+
+	if h.prov.destroyCount() != 2 {
+		t.Fatalf("after attempt 1 timeout: DestroyNode calls = %d; want 2", h.prov.destroyCount())
+	}
+
+	// ── Attempt 2 (final) ────────────────────────────────────────────────────
+
+	h.tick(t) // Tick 6: creating → CreateNode(idem=uuid:retry:2) → probing.
+
+	if h.prov.createCount() != 3 {
+		t.Fatalf("attempt 2: CreateNode calls = %d; want 3", h.prov.createCount())
+	}
+	call2 := h.prov.createCalls[2]
+	if call2.IdempotencyKey != replacementUUID+":retry:2" {
+		t.Errorf("attempt 2 idem key = %q; want %q:retry:2",
+			call2.IdempotencyKey, replacementUUID)
+	}
+
+	newNode2 := call2.ReturnedID
+	h.snaps.setPass(newNode2)
+	h.clock.advance(orchestrate.ProbeTimeout + time.Second)
+
+	h.tick(t) // Tick 7: probing → timeout → DestroyNode(newNode2) → FAILED.
+
+	if h.prov.destroyCount() != 3 {
+		t.Fatalf("after attempt 2 timeout: DestroyNode calls = %d; want 3", h.prov.destroyCount())
+	}
+
+	// Old node must never have been destroyed.
+	for _, id := range h.prov.destroyCalls {
+		if id == oldNode {
+			t.Errorf("old node %s was incorrectly destroyed during failed replacement", oldNode)
+		}
+	}
+
+	// NotifyFault must be called exactly once.
+	if n := h.notifier.count(); n != 1 {
+		t.Errorf("NotifyFault calls = %d; want 1", n)
+	}
+
+	// Record must be in failed phase.
+	raw, err := h.rdb.Get(ctx, "sched:replace:"+replacementUUID).Result()
+	if err != nil {
+		t.Fatalf("read final record: %v", err)
+	}
+	var p struct {
+		Phase orchestrate.Phase `json:"phase"`
+	}
+	if err := json.Unmarshal([]byte(raw), &p); err != nil {
+		t.Fatalf("unmarshal phase: %v", err)
+	}
+	if p.Phase != orchestrate.PhaseFailed {
+		t.Errorf("final phase = %q; want %q", p.Phase, orchestrate.PhaseFailed)
+	}
+
+	// Each attempt must have used a different provider (rotation check).
+	p0 := h.prov.createCalls[0].Spec.ProviderID
+	p1 := h.prov.createCalls[1].Spec.ProviderID
+	p2 := h.prov.createCalls[2].Spec.ProviderID
+	if p0 == p1 || p1 == p2 || p0 == p2 {
+		t.Errorf("providers should differ across retries; got [%s, %s, %s]", p0, p1, p2)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: crash recovery — CreateNode NOT called again when creating is complete
+//
+// Simulates a crash where CreateNode succeeded and newNode was set in memory
+// but the record was NOT yet saved with phase=probing. On restart the record
+// shows phase=creating with newNode set; we expect the Replacer to resume
+// from probing without calling CreateNode again.
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCrashRecovery(t *testing.T) {
+	const (
+		oldNode          = "old-node-crash"
+		replacementUUID  = "uuid-crash"
+		simulatedNewNode = "new-node-crash"
+	)
+	h := newHarness(t,
+		&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg", Role: "entry"},
+	)
+
+	ctx := context.Background()
+
+	// Inject a pre-crash record: phase=creating, newNode already set.
+	// This models a crash AFTER CreateNode returned but BEFORE the record was
+	// saved with phase=probing.
+	type crashRecord struct {
+		Phase             string    `json:"phase"`
+		OldNode           string    `json:"oldNode"`
+		NewNode           string    `json:"newNode"`
+		CurrentProviderID string    `json:"currentProviderId"`
+		Attempts          int       `json:"attempts"`
+		ProviderTried     []string  `json:"providerTried"`
+		ProbeStreak       int       `json:"probeStreak"`
+		PhaseStartedAt    time.Time `json:"phaseStartedAt"`
+		UpdatedAt         time.Time `json:"updatedAt"`
+	}
+	recJSON, _ := json.Marshal(crashRecord{
+		Phase:          "creating",
+		OldNode:        oldNode,
+		NewNode:        simulatedNewNode,
+		ProviderTried:  []string{},
+		PhaseStartedAt: h.clock.Now(),
+		UpdatedAt:      h.clock.Now(),
+	})
+	if err := h.rdb.Set(ctx, "sched:replace:"+replacementUUID, recJSON, 0).Err(); err != nil {
+		t.Fatalf("inject crash record: %v", err)
+	}
+	if err := h.rdb.SAdd(ctx, "sched:replace:index", replacementUUID).Err(); err != nil {
+		t.Fatalf("inject index: %v", err)
+	}
+
+	// Add the simulated new node to the lifecycle mock.
+	h.lc.addNode(&orchestrate.NodeInfo{ID: simulatedNewNode, Tier: "free", Region: "hkg"})
+	h.snaps.setPass(simulatedNewNode)
+
+	// Tick 1 ("restart"): phase=creating, newNode already set → skip CreateNode → probing.
+	h.tick(t)
+	if h.prov.createCount() != 0 {
+		t.Fatalf("tick 1 (recovery): CreateNode called %d times; want 0", h.prov.createCount())
+	}
+
+	h.tick(t) // Tick 2: probing streak=1.
+	h.tick(t) // Tick 3: probing streak=2 → activating.
+	h.tick(t) // Tick 4: activating → SetWeight(10) + TransitionStatus + BumpVersion + grayscale → draining_old.
+	h.tick(t) // Tick 5: draining_old → DestroyNode(oldNode) → done.
+
+	// CreateNode must never have been called.
+	if h.prov.createCount() != 0 {
+		t.Errorf("CreateNode called %d times during crash recovery; want 0", h.prov.createCount())
+	}
+
+	// New node must be at weight 10.
+	if w := h.lc.weightOf(simulatedNewNode); w != 10 {
+		t.Errorf("new node weight = %d; want 10", w)
+	}
+
+	// Old node must be destroyed.
+	oldDestroyed := false
+	for _, id := range h.prov.destroyCalls {
+		if id == oldNode {
+			oldDestroyed = true
+			break
+		}
+	}
+	if !oldDestroyed {
+		t.Errorf("old node %s not destroyed in crash-recovery path; destroyCalls = %v",
+			oldNode, h.prov.destroyCalls)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: idempotency — duplicate queue entries produce a single replacement
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestQueueReplayIdempotent(t *testing.T) {
+	const (
+		oldNode         = "old-node-idem"
+		replacementUUID = "uuid-idem"
+	)
+	h := newHarness(t,
+		&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg"},
+	)
+
+	// Push the same entry twice (duplicate delivery).
+	h.pushQueue(t, oldNode, replacementUUID)
+	h.pushQueue(t, oldNode, replacementUUID)
+
+	// Tick 1: both entries drained; SetNX ensures a single record; stepPending → creating.
+	h.tick(t)
+	if h.prov.createCount() != 0 {
+		t.Errorf("tick 1: CreateNode should not be called yet; got %d", h.prov.createCount())
+	}
+
+	// Tick 2: creating → CreateNode called exactly once.
+	h.tick(t)
+	if h.prov.createCount() != 1 {
+		t.Errorf("CreateNode calls = %d after two queue entries; want 1", h.prov.createCount())
+	}
+
+	// Only one UUID must be in the in-flight index.
+	members, err := h.rdb.SMembers(context.Background(), "sched:replace:index").Result()
+	if err != nil {
+		t.Fatalf("smembers: %v", err)
+	}
+	if len(members) != 1 {
+		t.Errorf("index member count = %d; want 1", len(members))
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: grayscale weight ramp 10 → 25 → 50 → 75 → 100 (each step = 6 h)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestGrayscaleAdvance(t *testing.T) {
+	const nodeID = "gray-node-001"
+
+	rdb, _ := newTestRedis(t)
+	lc := newMockLC(&orchestrate.NodeInfo{ID: nodeID})
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+
+	// Seed a grayscale record at weight 10.
+	grayKey := "sched:gray:" + nodeID
+	type grayRec struct {
+		NodeID         string    `json:"nodeId"`
+		CurrentWeight  int       `json:"currentWeight"`
+		StartedAt      time.Time `json:"startedAt"`
+		LastAdvancedAt time.Time `json:"lastAdvancedAt"`
+	}
+	data, _ := json.Marshal(grayRec{
+		NodeID:         nodeID,
+		CurrentWeight:  10,
+		StartedAt:      clock.Now(),
+		LastAdvancedAt: clock.Now(),
+	})
+	if err := rdb.Set(context.Background(), grayKey, data, 0).Err(); err != nil {
+		t.Fatalf("seed gray record: %v", err)
+	}
+
+	g := orchestrate.NewGrayscale(rdb, lc, clock)
+	ctx := context.Background()
+
+	// Advance the clock by GrayscaleInterval (6 h) and call Advance() four times.
+	// Expected progression: 10 → 25 → 50 → 75 → 100 (record deleted).
+	wantWeights := []int{25, 50, 75, 100}
+
+	for step, want := range wantWeights {
+		clock.advance(orchestrate.GrayscaleInterval)
+
+		if err := g.Advance(ctx); err != nil {
+			t.Fatalf("step %d Advance(): %v", step, err)
+		}
+
+		if got := lc.weightOf(nodeID); got != want {
+			t.Errorf("step %d: weight = %d; want %d", step, got, want)
+		}
+
+		// Audit log must record the weight advance.
+		auditKey := "grayscale|weight_advanced|node:" + nodeID
+		if !lc.hasAudit(auditKey) {
+			t.Errorf("step %d: audit log %q not found; logs = %v", step, auditKey, lc.auditLogs)
+		}
+	}
+
+	// After weight 100, the grayscale record must be deleted.
+	if _, err := rdb.Get(ctx, grayKey).Result(); err != redis.Nil {
+		t.Error("grayscale record should be deleted after weight 100")
+	}
+
+	// Final weight must be 100.
+	if lc.weightOf(nodeID) != 100 {
+		t.Errorf("final weight = %d; want 100", lc.weightOf(nodeID))
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: grayscale does not advance before the interval elapses
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestGrayscaleNoAdvanceBeforeInterval(t *testing.T) {
+	const nodeID = "gray-node-wait"
+
+	rdb, _ := newTestRedis(t)
+	lc := newMockLC(&orchestrate.NodeInfo{ID: nodeID})
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+
+	grayKey := "sched:gray:" + nodeID
+	type grayRec struct {
+		NodeID         string    `json:"nodeId"`
+		CurrentWeight  int       `json:"currentWeight"`
+		StartedAt      time.Time `json:"startedAt"`
+		LastAdvancedAt time.Time `json:"lastAdvancedAt"`
+	}
+	data, _ := json.Marshal(grayRec{
+		NodeID:         nodeID,
+		CurrentWeight:  10,
+		StartedAt:      clock.Now(),
+		LastAdvancedAt: clock.Now(),
+	})
+	rdb.Set(context.Background(), grayKey, data, 0)
+
+	g := orchestrate.NewGrayscale(rdb, lc, clock)
+
+	// Advance only 5 h (< 6 h interval).
+	clock.advance(5 * time.Hour)
+
+	if err := g.Advance(context.Background()); err != nil {
+		t.Fatalf("Advance(): %v", err)
+	}
+
+	// Weight must NOT have been updated.
+	if w := lc.weightOf(nodeID); w != 0 {
+		t.Errorf("weight was updated too early: got %d; want 0", w)
+	}
+}