merge: maestro/tsk_NPgPRxBGv0g9 into main (bulk integration)

server/internal/scheduler/orchestrate/breaker.go

@@ -0,0 +1,307 @@
+package orchestrate
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"math"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/wangjia/pangolin/server/internal/idgen"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Redis key constants
+// ─────────────────────────────────────────────────────────────────────────────
+
+const (
+	// breakerWindowKey is the sliding-window ZSET key prefix.
+	// Full key: sched:breaker:{tier}:{region}
+	// Schema: member = replacement UUID, score = Unix timestamp (seconds).
+	breakerWindowKey = "sched:breaker:"
+
+	// breakerTripKey is the trip-flag key prefix.
+	// Full key: sched:breaker:trip:{tier}:{region}
+	// TTL = BreakerWindowMin. Used for: alert dedup + admin visibility.
+	// The key auto-expires when the sliding window clears (enabling auto-recovery).
+	breakerTripKey = "sched:breaker:trip:"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// RedisBreaker
+// ─────────────────────────────────────────────────────────────────────────────
+
+// RedisBreaker is the production 15F circuit-breaker. It satisfies the
+// Breaker interface (Allow + Record) and additionally exposes Reset for the
+// admin handler (task #8).
+//
+// Mechanism:
+//
+//	ZSET sched:breaker:{tier}:{region}
+//	  member = replacement UUID (unique per completed replacement)
+//	  score  = Unix timestamp of completion
+//
+//	Window: BreakerWindowMin (default 60 min). Each Allow call
+//	  runs ZREMRANGEBYSCORE first to expire out-of-window entries.
+//
+//	Threshold N:
+//	  N = ceil(poolTarget × BreakerFractionPct / 100), min BreakerMinN (3).
+//
+//	Trip:
+//	  When window-count ≥ N: Allow returns false. The first detection in a
+//	  window epoch sets a trip-flag key with TTL = BreakerWindowMin and emits
+//	  a critical alert (idempotent via SetNX).
+//
+//	Auto-recovery:
+//	  After BreakerWindowMin minutes with no new Record calls, all ZSET entries
+//	  expire out of the window and the trip flag TTL also expires → Allow true.
+//
+//	Manual reset:
+//	  Admin calls Reset, which deletes the trip flag and the ZSET immediately.
+type RedisBreaker struct {
+	rdb      *redis.Client
+	pools    PoolReader
+	notifier Notifier
+	lc       LifecycleService // for audit log; may be nil
+	cfgMgr   *ConfigManager
+	clock    Clock
+}
+
+// RedisBreakerConfig holds all dependencies for NewRedisBreaker.
+type RedisBreakerConfig struct {
+	RDB      *redis.Client
+	Pools    PoolReader
+	Notifier Notifier
+	LC       LifecycleService // for audit log; may be nil
+	CfgMgr   *ConfigManager
+	Clock    Clock
+}
+
+// NewRedisBreaker constructs a RedisBreaker. If Clock is nil, RealClock is used.
+func NewRedisBreaker(cfg RedisBreakerConfig) *RedisBreaker {
+	if cfg.Clock == nil {
+		cfg.Clock = RealClock{}
+	}
+	return &RedisBreaker{
+		rdb:      cfg.RDB,
+		pools:    cfg.Pools,
+		notifier: cfg.Notifier,
+		lc:       cfg.LC,
+		cfgMgr:   cfg.CfgMgr,
+		clock:    cfg.Clock,
+	}
+}
+
+// Allow implements Breaker. Returns false when the pool's sliding-window
+// replacement count has reached the threshold (breaker is tripped), blocking
+// further replacements from starting.
+//
+// On Redis errors the breaker fails open (returns true) to avoid a single
+// point of failure halting all replacements.
+func (b *RedisBreaker) Allow(tier, region string) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	cfg := b.cfgMgr.Current()
+	window := time.Duration(cfg.Capacity.BreakerWindowMin) * time.Minute
+	threshold := b.computeThreshold(ctx, tier, region, cfg)
+
+	zKey := breakerWindowKey + tier + ":" + region
+	tripKey := breakerTripKey + tier + ":" + region
+	cutoff := b.clock.Now().Add(-window)
+
+	// Expire entries outside the sliding window.
+	cutoffStr := fmt.Sprintf("%d", cutoff.Unix())
+	if err := b.rdb.ZRemRangeByScore(ctx, zKey, "-inf", cutoffStr).Err(); err != nil {
+		slog.Error("breaker: ZREMRANGEBYSCORE failed; failing open",
+			"tier", tier, "region", region, "error", err)
+		return true // fail open
+	}
+
+	// Count remaining (in-window) entries.
+	count, err := b.rdb.ZCard(ctx, zKey).Result()
+	if err != nil {
+		slog.Error("breaker: ZCARD failed; failing open",
+			"tier", tier, "region", region, "error", err)
+		return true // fail open
+	}
+
+	if count < int64(threshold) {
+		return true // below threshold — allow replacement
+	}
+
+	// Count ≥ threshold: trip. Emit alert exactly once per window epoch via SetNX.
+	set, setErr := b.rdb.SetNX(ctx, tripKey, "1", window).Result()
+	if setErr != nil {
+		slog.Error("breaker: trip flag set failed",
+			"tier", tier, "region", region, "error", setErr)
+	}
+	if set {
+		b.emitTripAlert(ctx, tier, region, int(count), threshold, cfg)
+	}
+
+	slog.Warn("breaker: replacement blocked — circuit open",
+		"tier", tier, "region", region,
+		"window_count", count, "threshold", threshold,
+	)
+	return false
+}
+
+// Record implements Breaker. Records a completed replacement for the given
+// pool in the sliding-window ZSET. Called by the orchestrator after the old
+// node is successfully destroyed.
+//
+// Record also checks whether the count has just reached the threshold and,
+// if so, trips the breaker and emits a critical alert.
+func (b *RedisBreaker) Record(tier, region string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	cfg := b.cfgMgr.Current()
+	window := time.Duration(cfg.Capacity.BreakerWindowMin) * time.Minute
+	threshold := b.computeThreshold(ctx, tier, region, cfg)
+
+	zKey := breakerWindowKey + tier + ":" + region
+	tripKey := breakerTripKey + tier + ":" + region
+	now := b.clock.Now()
+
+	// Add a unique entry for this replacement.
+	member := idgen.NewString()
+	if err := b.rdb.ZAdd(ctx, zKey, redis.Z{
+		Score:  float64(now.Unix()),
+		Member: member,
+	}).Err(); err != nil {
+		slog.Error("breaker: ZADD failed",
+			"tier", tier, "region", region, "error", err)
+		return
+	}
+
+	// Check if we have just reached the threshold; trip once per epoch.
+	count, err := b.rdb.ZCard(ctx, zKey).Result()
+	if err != nil {
+		slog.Error("breaker: ZCARD after record failed",
+			"tier", tier, "region", region, "error", err)
+		return
+	}
+
+	if count >= int64(threshold) {
+		set, setErr := b.rdb.SetNX(ctx, tripKey, "1", window).Result()
+		if setErr != nil {
+			slog.Error("breaker: trip flag on record failed",
+				"tier", tier, "region", region, "error", setErr)
+			return
+		}
+		if set {
+			// First trip in this window epoch: emit critical alert.
+			b.emitTripAlert(ctx, tier, region, int(count), threshold, cfg)
+		}
+	}
+}
+
+// Reset clears a tripped breaker for the given pool by deleting both the trip
+// flag and the sliding-window ZSET. This lets new replacements proceed
+// immediately without waiting for the window to auto-expire.
+//
+// An audit log entry is written with actor and action metadata.
+// Reset is not part of the Breaker interface; it is called only by the admin
+// handler (task #8).
+func (b *RedisBreaker) Reset(ctx context.Context, tier, region, actor string) error {
+	zKey := breakerWindowKey + tier + ":" + region
+	tripKey := breakerTripKey + tier + ":" + region
+
+	pipe := b.rdb.Pipeline()
+	pipe.Del(ctx, tripKey)
+	pipe.Del(ctx, zKey)
+	if _, err := pipe.Exec(ctx); err != nil {
+		return fmt.Errorf("breaker: reset %s/%s: %w", tier, region, err)
+	}
+
+	slog.Info("breaker: manually reset by operator",
+		"tier", tier, "region", region, "actor", actor)
+
+	// Audit trail.
+	meta := fmt.Sprintf(
+		`{"tier":%q,"region":%q,"actor":%q,"action":"breaker_reset"}`,
+		tier, region, actor,
+	)
+	if b.lc != nil {
+		if err := b.lc.WriteAuditLog(
+			ctx, actor, "breaker_reset", "pool:"+tier+":"+region, meta,
+		); err != nil {
+			slog.Error("breaker: write audit log failed", "error", err)
+		}
+	}
+	return nil
+}
+
+// IsTripped reports whether the trip flag is currently set for the pool.
+// Useful for admin status queries.
+func (b *RedisBreaker) IsTripped(ctx context.Context, tier, region string) (bool, error) {
+	tripKey := breakerTripKey + tier + ":" + region
+	exists, err := b.rdb.Exists(ctx, tripKey).Result()
+	if err != nil {
+		return false, fmt.Errorf("breaker: IsTripped %s/%s: %w", tier, region, err)
+	}
+	return exists > 0, nil
+}
+
+// WindowCount returns the number of completed replacements currently recorded
+// in the sliding window for the pool. Useful for admin status queries.
+func (b *RedisBreaker) WindowCount(ctx context.Context, tier, region string) (int64, error) {
+	cfg := b.cfgMgr.Current()
+	window := time.Duration(cfg.Capacity.BreakerWindowMin) * time.Minute
+	cutoff := b.clock.Now().Add(-window)
+	cutoffStr := fmt.Sprintf("%d", cutoff.Unix())
+
+	zKey := breakerWindowKey + tier + ":" + region
+	if err := b.rdb.ZRemRangeByScore(ctx, zKey, "-inf", cutoffStr).Err(); err != nil {
+		return 0, fmt.Errorf("breaker: WindowCount cleanup %s/%s: %w", tier, region, err)
+	}
+	count, err := b.rdb.ZCard(ctx, zKey).Result()
+	if err != nil {
+		return 0, fmt.Errorf("breaker: WindowCount ZCARD %s/%s: %w", tier, region, err)
+	}
+	return count, nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+// computeThreshold returns the trip threshold for a pool:
+//
+//	N = ceil(poolTarget × BreakerFractionPct / 100), minimum BreakerMinN.
+//
+// Falls back to BreakerMinN when the pool cannot be found or has zero target.
+func (b *RedisBreaker) computeThreshold(ctx context.Context, tier, region string, cfg *SchedConfig) int {
+	_, target, err := b.pools.PoolCapacity(ctx, tier, region)
+	if err != nil || target <= 0 {
+		return cfg.Capacity.BreakerMinN
+	}
+	n := int(math.Ceil(float64(target) * float64(cfg.Capacity.BreakerFractionPct) / 100.0))
+	if n < cfg.Capacity.BreakerMinN {
+		n = cfg.Capacity.BreakerMinN
+	}
+	return n
+}
+
+// emitTripAlert sends a critical alert to the Notifier for a tripped breaker.
+func (b *RedisBreaker) emitTripAlert(ctx context.Context, tier, region string, count, threshold int, cfg *SchedConfig) {
+	poolID := "pool:" + tier + ":" + region
+	reason := fmt.Sprintf(
+		"circuit breaker tripped: %d replacements in %d-min window (threshold=%d) — manual Reset required",
+		count, cfg.Capacity.BreakerWindowMin, threshold,
+	)
+	slog.Error("breaker: circuit breaker tripped",
+		"tier", tier, "region", region,
+		"window_count", count, "threshold", threshold,
+	)
+	if b.notifier != nil {
+		if err := b.notifier.NotifyFault(ctx, poolID, reason); err != nil {
+			slog.Error("breaker: emit trip alert failed",
+				"pool", poolID, "error", err)
+		}
+	}
+}

server/internal/scheduler/orchestrate/breaker_test.go

@@ -0,0 +1,295 @@
+package orchestrate_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock PoolReader
+// ─────────────────────────────────────────────────────────────────────────────
+
+// mockPoolReader is a simple in-memory PoolReader for tests.
+type mockPoolReader struct {
+	mu    sync.Mutex
+	pools map[string]orchestrate.PoolStat // key = tier+":"+region
+}
+
+func newMockPoolReader() *mockPoolReader {
+	return &mockPoolReader{pools: make(map[string]orchestrate.PoolStat)}
+}
+
+func (r *mockPoolReader) setPool(tier, region string, up, target int) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.pools[tier+":"+region] = orchestrate.PoolStat{
+		Tier: tier, Region: region, UpCount: up, Target: target,
+	}
+}
+
+func (r *mockPoolReader) AllPools(_ context.Context) ([]orchestrate.PoolStat, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]orchestrate.PoolStat, 0, len(r.pools))
+	for _, s := range r.pools {
+		out = append(out, s)
+	}
+	return out, nil
+}
+
+func (r *mockPoolReader) PoolCapacity(_ context.Context, tier, region string) (int, int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	s, ok := r.pools[tier+":"+region]
+	if !ok {
+		return 0, 0, nil
+	}
+	return s.UpCount, s.Target, nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Helper: build a RedisBreaker for tests
+// ─────────────────────────────────────────────────────────────────────────────
+
+func newTestBreaker(t *testing.T, pools *mockPoolReader, notifier *mockNotifier, clock *mockClock) (*orchestrate.RedisBreaker, *orchestrate.ConfigManager) {
+	t.Helper()
+	rdb, _ := newTestRedis(t)
+	cfgMgr := orchestrate.NewConfigManager("", nil)
+	_ = cfgMgr.Load()
+
+	return orchestrate.NewRedisBreaker(orchestrate.RedisBreakerConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   cfgMgr,
+		Clock:    clock,
+	}), cfgMgr
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: N-th Record trips the breaker; (N+1)-th Allow returns false
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestBreakerTripsAtThreshold(t *testing.T) {
+	const (
+		tier   = "free"
+		region = "hkg"
+	)
+
+	pools := newMockPoolReader()
+	// Pool of 10 nodes → threshold = ceil(10 × 30%) = 3.
+	pools.setPool(tier, region, 10, 10)
+
+	notifier := &mockNotifier{}
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+
+	b, _ := newTestBreaker(t, pools, notifier, clock)
+
+	// Before any records: Allow should return true.
+	if !b.Allow(tier, region) {
+		t.Fatal("Allow should return true before any records")
+	}
+
+	// Record N-1 replacements (threshold-1 = 2); Allow should still be true.
+	b.Record(tier, region) // count = 1
+	b.Record(tier, region) // count = 2
+	if !b.Allow(tier, region) {
+		t.Fatalf("Allow should return true with count < threshold (count=2, threshold=3)")
+	}
+
+	// Record the N-th replacement; now count = 3 = threshold.
+	b.Record(tier, region) // count = 3
+
+	// (N+1)-th Allow: breaker is tripped → false.
+	if b.Allow(tier, region) {
+		t.Error("Allow should return false when count >= threshold (tripped)")
+	}
+
+	// Critical alert must have been emitted exactly once.
+	if n := notifier.count(); n != 1 {
+		t.Errorf("NotifyFault calls = %d; want 1 (on trip)", n)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Window slide-out auto-recovers the breaker
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestBreakerAutoRecoveryAfterWindowExpiry(t *testing.T) {
+	const (
+		tier   = "premium"
+		region = "sin"
+	)
+
+	pools := newMockPoolReader()
+	// Pool of 10 → threshold = 3.
+	pools.setPool(tier, region, 10, 10)
+
+	notifier := &mockNotifier{}
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+	b, _ := newTestBreaker(t, pools, notifier, clock)
+
+	// Record N entries to trip.
+	b.Record(tier, region)
+	b.Record(tier, region)
+	b.Record(tier, region)
+	if b.Allow(tier, region) {
+		t.Fatal("breaker should be tripped after 3 records (threshold=3)")
+	}
+
+	// Advance clock past the 1-h window; all ZSET entries are now stale.
+	clock.advance(61 * time.Minute)
+
+	// Allow should return true: ZREMRANGEBYSCORE removes stale entries,
+	// count drops to 0 < threshold, and the trip flag TTL has expired.
+	if !b.Allow(tier, region) {
+		t.Error("Allow should return true after the window has expired (auto-recovery)")
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Reset manually clears the breaker with audit
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestBreakerResetRestoresAllow(t *testing.T) {
+	const (
+		tier   = "free"
+		region = "tyo"
+	)
+
+	pools := newMockPoolReader()
+	pools.setPool(tier, region, 10, 10) // threshold = 3
+
+	notifier := &mockNotifier{}
+	lc := newMockLC()
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+	rdb, _ := newTestRedis(t)
+	cfgMgr := orchestrate.NewConfigManager("", nil)
+	_ = cfgMgr.Load()
+
+	b := orchestrate.NewRedisBreaker(orchestrate.RedisBreakerConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		LC:       lc,
+		CfgMgr:   cfgMgr,
+		Clock:    clock,
+	})
+
+	// Trip the breaker.
+	b.Record(tier, region)
+	b.Record(tier, region)
+	b.Record(tier, region)
+	if b.Allow(tier, region) {
+		t.Fatal("breaker should be tripped")
+	}
+
+	// Reset.
+	ctx := context.Background()
+	if err := b.Reset(ctx, tier, region, "admin-alice"); err != nil {
+		t.Fatalf("Reset error: %v", err)
+	}
+
+	// Allow should now return true.
+	if !b.Allow(tier, region) {
+		t.Error("Allow should return true after Reset")
+	}
+
+	// Audit log must mention the reset.
+	if !lc.hasAudit("admin-alice|breaker_reset|pool:" + tier + ":" + region) {
+		t.Errorf("audit log missing breaker_reset entry; got: %v", lc.auditLogs)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: BreakerMinN = 3 enforced on tiny pool (target < 10)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestBreakerMinNEnforcedOnSmallPool(t *testing.T) {
+	const (
+		tier   = "free"
+		region = "fra"
+	)
+
+	pools := newMockPoolReader()
+	// Pool of 2 nodes; 30% of 2 = 0.6 → ceil = 1, but min = 3.
+	pools.setPool(tier, region, 2, 2)
+
+	notifier := &mockNotifier{}
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+	b, _ := newTestBreaker(t, pools, notifier, clock)
+
+	// 2 records should NOT trip (threshold = 3, not 1).
+	b.Record(tier, region)
+	b.Record(tier, region)
+	if !b.Allow(tier, region) {
+		t.Error("Allow should return true: count=2, threshold=min(3)")
+	}
+
+	// 3rd record trips.
+	b.Record(tier, region)
+	if b.Allow(tier, region) {
+		t.Error("Allow should return false: count=3 >= minN(3)")
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Trip alert is emitted only once per window epoch (dedup)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestBreakerAlertSentOnlyOnce(t *testing.T) {
+	const (
+		tier   = "free"
+		region = "ams"
+	)
+
+	pools := newMockPoolReader()
+	pools.setPool(tier, region, 10, 10) // threshold = 3
+
+	notifier := &mockNotifier{}
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+	b, _ := newTestBreaker(t, pools, notifier, clock)
+
+	// Trip.
+	b.Record(tier, region)
+	b.Record(tier, region)
+	b.Record(tier, region)
+
+	// Call Allow multiple times while tripped.
+	for i := 0; i < 5; i++ {
+		b.Allow(tier, region)
+	}
+
+	// NotifyFault must have been called exactly once (from Record reaching N).
+	if n := notifier.count(); n != 1 {
+		t.Errorf("NotifyFault calls = %d; want exactly 1 (trip alert dedup)", n)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: pool not found → falls back to BreakerMinN (3)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestBreakerFallsBackToMinNForUnknownPool(t *testing.T) {
+	pools := newMockPoolReader()
+	// No pool registered for "unknown"/"xyz".
+
+	notifier := &mockNotifier{}
+	clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
+	b, _ := newTestBreaker(t, pools, notifier, clock)
+
+	b.Record("unknown", "xyz")
+	b.Record("unknown", "xyz")
+	if !b.Allow("unknown", "xyz") {
+		t.Error("Allow should return true: count=2, threshold=minN(3) for unknown pool")
+	}
+
+	b.Record("unknown", "xyz")
+	if b.Allow("unknown", "xyz") {
+		t.Error("Allow should return false: count=3 >= minN(3)")
+	}
+}

server/internal/scheduler/orchestrate/capacity.go

@@ -0,0 +1,242 @@
+package orchestrate
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// PoolReader — pool capacity interface
+// ─────────────────────────────────────────────────────────────────────────────
+
+// PoolStat holds capacity metrics for a single tier+region pool.
+type PoolStat struct {
+	Tier    string
+	Region  string
+	UpCount int // number of nodes currently in "up" status
+	Target  int // configured target node count for this pool
+}
+
+// PoolReader provides pool capacity information for the capacity monitor and
+// circuit breaker. The real implementation (wired by task 15H) queries the
+// lifecycle service.
+type PoolReader interface {
+	// AllPools returns all known pools with their current UpCount and Target.
+	AllPools(ctx context.Context) ([]PoolStat, error)
+
+	// PoolCapacity returns (upCount, target) for the named pool.
+	// Returns (0, 0, nil) when the pool is not found.
+	PoolCapacity(ctx context.Context, tier, region string) (int, int, error)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Redis key constants
+// ─────────────────────────────────────────────────────────────────────────────
+
+const (
+	// capAlertPoolPrefix is the dedup key prefix for pool-watermark alerts.
+	// Full key: sched:cap:pool:{tier}:{region}   TTL = AlertSuppressMin
+	capAlertPoolPrefix = "sched:cap:pool:"
+
+	// capAlertProbePrefix is the dedup key prefix for probe-lost-contact alerts.
+	// Full key: sched:cap:probe:{probeId}   TTL = AlertSuppressMin
+	capAlertProbePrefix = "sched:cap:probe:"
+
+	// probeHBKeyPrefix is the heartbeat key prefix written by the probe package.
+	// Full key: probe:hb:{probeId}   TTL = 15 min (from probe.heartbeatTTL)
+	probeHBKeyPrefix = "probe:hb:"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// CapacityMonitor
+// ─────────────────────────────────────────────────────────────────────────────
+
+// CapacityMonitorConfig holds all dependencies for CapacityMonitor.
+type CapacityMonitorConfig struct {
+	RDB      *redis.Client
+	Pools    PoolReader
+	Notifier Notifier
+	Gray     *Grayscale        // grayscale weight-ramp driver
+	LC       LifecycleService  // for WriteAuditLog; may be nil
+	CfgMgr   *ConfigManager
+	Clock    Clock
+}
+
+// CapacityMonitor checks pool watermarks and probe heartbeats, and drives
+// the grayscale weight ramp. It is called by the CapacityLoop (task 15H)
+// approximately every 1–2 minutes.
+//
+// Each RunOnce call:
+//  1. Checks pool fill-rates; emits a throttled alert per pool below threshold.
+//  2. Checks probe heartbeats for every configured probe ID; emits a throttled
+//     「探针失联」alert for each probe whose heartbeat key is absent.
+//  3. Calls Grayscale.Advance to advance any in-progress weight ramps.
+type CapacityMonitor struct {
+	rdb      *redis.Client
+	pools    PoolReader
+	notifier Notifier
+	gray     *Grayscale
+	lc       LifecycleService
+	cfgMgr   *ConfigManager
+	clock    Clock
+}
+
+// NewCapacityMonitor constructs a CapacityMonitor from the given config.
+// If Clock is nil, RealClock is used.
+func NewCapacityMonitor(cfg CapacityMonitorConfig) *CapacityMonitor {
+	if cfg.Clock == nil {
+		cfg.Clock = RealClock{}
+	}
+	return &CapacityMonitor{
+		rdb:      cfg.RDB,
+		pools:    cfg.Pools,
+		notifier: cfg.Notifier,
+		gray:     cfg.Gray,
+		lc:       cfg.LC,
+		cfgMgr:   cfg.CfgMgr,
+		clock:    cfg.Clock,
+	}
+}
+
+// RunOnce executes one full capacity-monitoring cycle. Errors from individual
+// steps are logged but do not abort subsequent steps.
+func (m *CapacityMonitor) RunOnce(ctx context.Context) error {
+	cfg := m.cfgMgr.Current()
+
+	// Step 1: pool watermark checks.
+	if err := m.checkWatermarks(ctx, cfg); err != nil {
+		slog.Error("capacity: watermark check error", "error", err)
+	}
+
+	// Step 2: probe heartbeat checks.
+	if err := m.checkProbeHeartbeats(ctx, cfg); err != nil {
+		slog.Error("capacity: probe heartbeat check error", "error", err)
+	}
+
+	// Step 3: advance grayscale weight ramps (Grayscale.Advance is 15E's
+	// warm-up entry point; CapacityLoop drives it per the spec).
+	if m.gray != nil {
+		if err := m.gray.Advance(ctx); err != nil {
+			slog.Error("capacity: grayscale advance error", "error", err)
+		}
+	}
+
+	return nil
+}
+
+// checkWatermarks iterates all known pools. For each pool whose fill-rate
+// (upCount/target) is below WatermarkThreshold, an alert is emitted — but
+// at most once per AlertSuppressMin to suppress repeated notifications.
+func (m *CapacityMonitor) checkWatermarks(ctx context.Context, cfg *SchedConfig) error {
+	stats, err := m.pools.AllPools(ctx)
+	if err != nil {
+		return fmt.Errorf("capacity: list pools: %w", err)
+	}
+
+	threshold := cfg.Capacity.WatermarkThreshold
+	suppressTTL := time.Duration(cfg.Capacity.AlertSuppressMin) * time.Minute
+
+	for _, stat := range stats {
+		if stat.Target <= 0 {
+			continue // skip misconfigured pools with no target
+		}
+		fillRate := float64(stat.UpCount) / float64(stat.Target)
+		if fillRate >= threshold {
+			continue // pool is healthy
+		}
+
+		// Throttle: at most one alert per pool per AlertSuppressMin.
+		dedupKey := capAlertPoolPrefix + stat.Tier + ":" + stat.Region
+		set, setErr := m.rdb.SetNX(ctx, dedupKey, "1", suppressTTL).Result()
+		if setErr != nil {
+			slog.Error("capacity: watermark dedup key error",
+				"key", dedupKey, "error", setErr)
+			continue
+		}
+		if !set {
+			continue // alert already sent within suppression window
+		}
+
+		poolID := "pool:" + stat.Tier + ":" + stat.Region
+		reason := fmt.Sprintf(
+			"capacity watermark below %.0f%%: %d/%d up (fill_rate=%.1f%%)",
+			threshold*100, stat.UpCount, stat.Target, fillRate*100,
+		)
+
+		slog.Warn("capacity: pool below watermark",
+			"tier", stat.Tier,
+			"region", stat.Region,
+			"up", stat.UpCount,
+			"target", stat.Target,
+			"fill_rate_pct", fmt.Sprintf("%.1f", fillRate*100),
+		)
+
+		if m.notifier != nil {
+			if notifyErr := m.notifier.NotifyFault(ctx, poolID, reason); notifyErr != nil {
+				slog.Error("capacity: notify watermark alert",
+					"pool", poolID, "error", notifyErr)
+			}
+		}
+	}
+	return nil
+}
+
+// checkProbeHeartbeats verifies that every configured probe agent has a live
+// heartbeat key in Redis. A missing key means the probe has not reported
+// within the 15-minute heartbeat TTL (probe package constant heartbeatTTL).
+//
+// Semantics: absence of the key is "no recent data", NOT "node blocked". It
+// must NOT feed into any determination logic — only ops alerting.
+func (m *CapacityMonitor) checkProbeHeartbeats(ctx context.Context, cfg *SchedConfig) error {
+	probeIDs := cfg.Capacity.ExpectedProbeIDs
+	if len(probeIDs) == 0 {
+		return nil // no probes configured; nothing to check
+	}
+
+	suppressTTL := time.Duration(cfg.Capacity.AlertSuppressMin) * time.Minute
+
+	for _, pid := range probeIDs {
+		hbKey := probeHBKeyPrefix + pid
+		exists, err := m.rdb.Exists(ctx, hbKey).Result()
+		if err != nil {
+			slog.Error("capacity: probe heartbeat check error",
+				"probe_id", pid, "error", err)
+			continue
+		}
+		if exists > 0 {
+			continue // heartbeat key present — probe is alive
+		}
+
+		// Heartbeat key absent: throttled 「探针失联」alert.
+		dedupKey := capAlertProbePrefix + pid
+		set, setErr := m.rdb.SetNX(ctx, dedupKey, "1", suppressTTL).Result()
+		if setErr != nil {
+			slog.Error("capacity: probe alert dedup key error",
+				"probe_id", pid, "error", setErr)
+			continue
+		}
+		if !set {
+			continue // alert already sent within suppression window
+		}
+
+		probeNodeID := "probe:" + pid
+		reason := "探针失联: no heartbeat received in last 15 min (no data, not a block signal)"
+
+		slog.Warn("capacity: probe lost contact",
+			"probe_id", pid,
+			"hb_key", hbKey,
+		)
+
+		if m.notifier != nil {
+			if notifyErr := m.notifier.NotifyFault(ctx, probeNodeID, reason); notifyErr != nil {
+				slog.Error("capacity: notify probe lost contact",
+					"probe_id", pid, "error", notifyErr)
+			}
+		}
+	}
+	return nil
+}

server/internal/scheduler/orchestrate/capacity_test.go

@@ -0,0 +1,258 @@
+package orchestrate_test
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Utility: write a temporary YAML config file
+// ─────────────────────────────────────────────────────────────────────────────
+
+func writeTempYAML(t *testing.T, content string) (string, error) {
+	t.Helper()
+	f, err := os.CreateTemp("", "sched_config_*.yaml")
+	if err != nil {
+		return "", err
+	}
+	t.Cleanup(func() { os.Remove(f.Name()) })
+	if _, err := f.WriteString(content); err != nil {
+		f.Close()
+		return "", err
+	}
+	f.Close()
+	return f.Name(), nil
+}
+
+// defaultCfgMgr returns a ConfigManager loaded with production defaults.
+func defaultCfgMgr(t *testing.T) *orchestrate.ConfigManager {
+	t.Helper()
+	m := orchestrate.NewConfigManager("", nil)
+	if err := m.Load(); err != nil {
+		t.Fatalf("ConfigManager.Load: %v", err)
+	}
+	return m
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Pool below 70% watermark → alert emitted
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCapacityWatermarkAlert(t *testing.T) {
+	rdb, _ := newTestRedis(t)
+	pools := newMockPoolReader()
+	// 6 out of 10 up = 60% < 70% → alert expected.
+	pools.setPool("free", "hkg", 6, 10)
+
+	notifier := &mockNotifier{}
+	cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   defaultCfgMgr(t),
+	})
+
+	if err := cm.RunOnce(context.Background()); err != nil {
+		t.Fatalf("RunOnce: %v", err)
+	}
+
+	if notifier.count() != 1 {
+		t.Errorf("NotifyFault calls = %d; want 1 (watermark alert)", notifier.count())
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Pool at exactly 70% → no alert
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCapacityWatermarkNoAlertAtThreshold(t *testing.T) {
+	rdb, _ := newTestRedis(t)
+	pools := newMockPoolReader()
+	// 7/10 = 70% == threshold → healthy, no alert.
+	pools.setPool("free", "hkg", 7, 10)
+
+	notifier := &mockNotifier{}
+	cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   defaultCfgMgr(t),
+	})
+
+	if err := cm.RunOnce(context.Background()); err != nil {
+		t.Fatalf("RunOnce: %v", err)
+	}
+
+	if notifier.count() != 0 {
+		t.Errorf("NotifyFault calls = %d; want 0 (fill rate = 70%% >= threshold)", notifier.count())
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Repeated RunOnce within suppress window → only 1 alert
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCapacityWatermarkAlertSuppression(t *testing.T) {
+	rdb, mr := newTestRedis(t)
+	pools := newMockPoolReader()
+	pools.setPool("free", "sin", 5, 10) // 50% < 70%
+
+	notifier := &mockNotifier{}
+	cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   defaultCfgMgr(t),
+	})
+
+	ctx := context.Background()
+
+	// Three RunOnce calls within the 10-min suppress window.
+	for i := 0; i < 3; i++ {
+		if err := cm.RunOnce(ctx); err != nil {
+			t.Fatalf("RunOnce %d: %v", i, err)
+		}
+	}
+	if notifier.count() != 1 {
+		t.Errorf("NotifyFault calls = %d; want 1 (suppressed repeats)", notifier.count())
+	}
+
+	// Advance miniredis TTL past the 10-min suppress window.
+	mr.FastForward(11 * time.Minute)
+
+	// After suppress window expires → another alert.
+	if err := cm.RunOnce(ctx); err != nil {
+		t.Fatalf("RunOnce after expire: %v", err)
+	}
+	if notifier.count() != 2 {
+		t.Errorf("NotifyFault calls = %d; want 2 after suppress window expires", notifier.count())
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Missing probe heartbeat → alert; present probe → no alert
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCapacityProbeHeartbeatMissingAlert(t *testing.T) {
+	rdb, _ := newTestRedis(t)
+	pools := newMockPoolReader()
+	notifier := &mockNotifier{}
+
+	ctx := context.Background()
+	// probe-b has a live heartbeat; probe-a does not.
+	if err := rdb.Set(ctx, "probe:hb:probe-b", "1", 15*time.Minute).Err(); err != nil {
+		t.Fatalf("seed probe-b heartbeat: %v", err)
+	}
+
+	yamlContent := `
+capacity:
+  expected_probe_ids:
+    - probe-a
+    - probe-b
+  alert_suppress_min: 10
+`
+	tmpFile, err := writeTempYAML(t, yamlContent)
+	if err != nil {
+		t.Fatalf("write temp YAML: %v", err)
+	}
+	cfgMgr := orchestrate.NewConfigManager(tmpFile, nil)
+	if err := cfgMgr.Load(); err != nil {
+		t.Fatalf("Load YAML config: %v", err)
+	}
+
+	cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   cfgMgr,
+	})
+
+	if err := cm.RunOnce(ctx); err != nil {
+		t.Fatalf("RunOnce: %v", err)
+	}
+
+	// Only probe-a should trigger an alert.
+	if notifier.count() != 1 {
+		t.Errorf("NotifyFault calls = %d; want 1 (only probe-a missing)", notifier.count())
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Both probes alive → no alert
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCapacityProbeHeartbeatBothAlive(t *testing.T) {
+	rdb, _ := newTestRedis(t)
+	pools := newMockPoolReader()
+	notifier := &mockNotifier{}
+
+	ctx := context.Background()
+	rdb.Set(ctx, "probe:hb:probe-a", "1", 15*time.Minute)
+	rdb.Set(ctx, "probe:hb:probe-b", "1", 15*time.Minute)
+
+	yamlContent := `
+capacity:
+  expected_probe_ids:
+    - probe-a
+    - probe-b
+`
+	tmpFile, err := writeTempYAML(t, yamlContent)
+	if err != nil {
+		t.Fatalf("write temp YAML: %v", err)
+	}
+	cfgMgr := orchestrate.NewConfigManager(tmpFile, nil)
+	if err := cfgMgr.Load(); err != nil {
+		t.Fatalf("Load YAML: %v", err)
+	}
+
+	cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   cfgMgr,
+	})
+
+	if err := cm.RunOnce(ctx); err != nil {
+		t.Fatalf("RunOnce: %v", err)
+	}
+
+	if notifier.count() != 0 {
+		t.Errorf("NotifyFault calls = %d; want 0 (both probes alive)", notifier.count())
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Multiple pools — only below-threshold pools alert
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestCapacityMultiplePoolsSelectiveAlert(t *testing.T) {
+	rdb, _ := newTestRedis(t)
+	pools := newMockPoolReader()
+	// hkg: 60% → alert
+	pools.setPool("free", "hkg", 6, 10)
+	// sin: 90% → no alert
+	pools.setPool("free", "sin", 9, 10)
+	// tyo: 50% → alert
+	pools.setPool("free", "tyo", 5, 10)
+
+	notifier := &mockNotifier{}
+	cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{
+		RDB:      rdb,
+		Pools:    pools,
+		Notifier: notifier,
+		CfgMgr:   defaultCfgMgr(t),
+	})
+
+	if err := cm.RunOnce(context.Background()); err != nil {
+		t.Fatalf("RunOnce: %v", err)
+	}
+
+	// Expect 2 alerts: hkg and tyo; sin is healthy.
+	if n := notifier.count(); n != 2 {
+		t.Errorf("NotifyFault calls = %d; want 2 (hkg + tyo)", n)
+	}
+}

server/internal/scheduler/orchestrate/config.go

@@ -0,0 +1,402 @@
+package orchestrate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"syscall"
+
+	"gopkg.in/yaml.v3"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// SchedConfig — master configuration tree
+// ─────────────────────────────────────────────────────────────────────────────
+
+// SchedConfig is the single source of truth for all scheduler thresholds.
+// It covers: 15D detection, 15F capacity/breaker, and 15E probe/grayscale.
+//
+// Values are layered in order:
+//  1. Defaults (DefaultSchedConfig).
+//  2. Optional YAML file (path set via SCHED_CONFIG_FILE env var, or passed
+//     to NewConfigManager).
+//  3. Environment variable overrides (applied on top of YAML).
+//
+// The live config is accessed via ConfigManager.Current() — a lock-free
+// atomic.Pointer read. Hot reload is triggered by SIGHUP; each reload
+// computes a JSON diff and calls the provided AuditFn.
+type SchedConfig struct {
+	// Detect holds 15D detection-engine thresholds.
+	Detect DetectSection `yaml:"detect"`
+	// Capacity holds 15F capacity-monitor and breaker thresholds.
+	Capacity CapacitySection `yaml:"capacity"`
+	// Probe holds 15E probe-window and grayscale-ramp settings.
+	Probe ProbeSection `yaml:"probe"`
+}
+
+// DetectSection mirrors detect.DetectConfig; task 15H bridges the two at
+// wiring time.
+type DetectSection struct {
+	// DomesticFailNumerator / DomesticFailDenominator define the ISP-failure
+	// fraction that triggers the suspect rule (default: 2/3).
+	DomesticFailNumerator   int `yaml:"domestic_fail_numerator"`
+	DomesticFailDenominator int `yaml:"domestic_fail_denominator"`
+
+	// SuspectStreakMin is the number of consecutive failing cycles before a
+	// node enters blocked_suspect (default: 2).
+	SuspectStreakMin int `yaml:"suspect_streak_min"`
+
+	// TrafficDropThreshold is the minimum percentage drop in online connections
+	// over 15 min that activates the traffic-warning rule (default: 80.0).
+	TrafficDropThreshold float64 `yaml:"traffic_drop_threshold"`
+
+	// TrafficBaselineMin is the minimum current online-connection count for
+	// the traffic-warning relaxation to apply (default: 20).
+	TrafficBaselineMin int `yaml:"traffic_baseline_min"`
+
+	// ConfirmedStreakMin is the number of consecutive cycles in blocked_suspect
+	// before promotion to blocked_confirmed (default: 6).
+	ConfirmedStreakMin int `yaml:"confirmed_streak_min"`
+
+	// RecoverStreakMin is the number of consecutive passing cycles while in
+	// blocked_suspect required to recover to up (default: 2).
+	RecoverStreakMin int `yaml:"recover_streak_min"`
+
+	// SuspectWeight is the routing weight applied when a node first enters
+	// blocked_suspect (default: 10).
+	SuspectWeight int `yaml:"suspect_weight"`
+}
+
+// CapacitySection holds 15F capacity-monitor and circuit-breaker thresholds.
+type CapacitySection struct {
+	// WatermarkThreshold is the minimum pool fill-rate (upCount/target) before
+	// an alert is emitted (default: 0.70 = 70%).
+	WatermarkThreshold float64 `yaml:"watermark_threshold"`
+
+	// AlertSuppressMin is the minimum gap in minutes between repeated alerts
+	// for the same pool or probe (default: 10 min).
+	AlertSuppressMin int `yaml:"alert_suppress_min"`
+
+	// BreakerWindowMin is the sliding-window duration in minutes for the
+	// circuit breaker (default: 60 min = 1 h).
+	BreakerWindowMin int `yaml:"breaker_window_min"`
+
+	// BreakerFractionPct is the percentage of pool target capacity that, when
+	// replaced within one window, trips the breaker (default: 30).
+	BreakerFractionPct int `yaml:"breaker_fraction_pct"`
+
+	// BreakerMinN is the absolute lower bound for the breaker trip threshold
+	// (default: 3). Overrides fraction when fraction yields a smaller number.
+	BreakerMinN int `yaml:"breaker_min_n"`
+
+	// ManualAlertThresh is the consecutive probe-failure count for a single
+	// node that triggers a 转人工 (escalate-to-human) alert. Should match
+	// 15E's MaxAttempts (default: 3).
+	ManualAlertThresh int `yaml:"manual_alert_thresh"`
+
+	// ExpectedProbeIDs is the list of probe-agent IDs that must maintain live
+	// heartbeat keys in Redis. A missing key triggers a 「探针失联」alert.
+	ExpectedProbeIDs []string `yaml:"expected_probe_ids"`
+}
+
+// ProbeSection holds 15E probe-window and grayscale-ramp settings.
+type ProbeSection struct {
+	// GrayscaleIntervalHours is the time in hours between successive grayscale
+	// weight-ramp steps (default: 6 h).
+	GrayscaleIntervalHours int `yaml:"grayscale_interval_hours"`
+
+	// ProbeTimeoutMin is the maximum time in minutes allowed for a node to
+	// pass probing per attempt (default: 15 min).
+	ProbeTimeoutMin int `yaml:"probe_timeout_min"`
+
+	// MaxAttempts is the maximum number of create+probe attempts per
+	// replacement record before marking it failed (default: 3).
+	MaxAttempts int `yaml:"max_attempts"`
+
+	// ProbeCyclesRequired is the number of consecutive passing probe ticks
+	// needed before a new node is promoted to up (default: 2).
+	ProbeCyclesRequired int `yaml:"probe_cycles_required"`
+}
+
+// DefaultSchedConfig returns a SchedConfig pre-filled with production defaults.
+func DefaultSchedConfig() SchedConfig {
+	return SchedConfig{
+		Detect: DetectSection{
+			DomesticFailNumerator:   2,
+			DomesticFailDenominator: 3,
+			SuspectStreakMin:        2,
+			TrafficDropThreshold:    80.0,
+			TrafficBaselineMin:      20,
+			ConfirmedStreakMin:       6,
+			RecoverStreakMin:        2,
+			SuspectWeight:           10,
+		},
+		Capacity: CapacitySection{
+			WatermarkThreshold: 0.70,
+			AlertSuppressMin:   10,
+			BreakerWindowMin:   60,
+			BreakerFractionPct: 30,
+			BreakerMinN:        3,
+			ManualAlertThresh:  3,
+		},
+		Probe: ProbeSection{
+			GrayscaleIntervalHours: 6,
+			ProbeTimeoutMin:        15,
+			MaxAttempts:            3,
+			ProbeCyclesRequired:    2,
+		},
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// ConfigManager
+// ─────────────────────────────────────────────────────────────────────────────
+
+// AuditFn is called after each successful config reload with the old and new
+// configurations serialised as JSON strings. The caller is responsible for
+// deciding how to persist the diff (e.g. via LifecycleService.WriteAuditLog).
+// May be nil to skip auditing.
+type AuditFn func(ctx context.Context, oldJSON, newJSON string)
+
+// ConfigManager holds the live SchedConfig and supports SIGHUP-triggered
+// hot reload. Config reads are lock-free (atomic.Pointer).
+//
+// Usage:
+//
+//	mgr := NewConfigManager("/etc/pangolin/sched.yaml", auditFn)
+//	if err := mgr.Load(); err != nil { log.Fatal(err) }
+//	mgr.WatchSIGHUP(ctx)
+//	cfg := mgr.Current()
+type ConfigManager struct {
+	path    string              // optional YAML file path (may be empty)
+	cfg     atomic.Pointer[SchedConfig]
+	auditFn AuditFn
+}
+
+// NewConfigManager creates a ConfigManager.
+//
+//   path:    optional path to the YAML config file. If SCHED_CONFIG_FILE is
+//            set in the environment it overrides this argument.
+//   auditFn: called on every reload when the config changes; may be nil.
+//
+// The manager is seeded with DefaultSchedConfig so Current() is never nil.
+func NewConfigManager(path string, fn AuditFn) *ConfigManager {
+	if p := os.Getenv("SCHED_CONFIG_FILE"); p != "" {
+		path = p
+	}
+	m := &ConfigManager{path: path, auditFn: fn}
+	dflt := DefaultSchedConfig()
+	m.cfg.Store(&dflt)
+	return m
+}
+
+// Current returns the live config. Never nil.
+func (m *ConfigManager) Current() *SchedConfig {
+	return m.cfg.Load()
+}
+
+// Load reads the config (YAML file if configured, then env overrides) and
+// atomically replaces the live config. Safe to call multiple times.
+func (m *ConfigManager) Load() error {
+	next, err := loadSchedConfig(m.path)
+	if err != nil {
+		return err
+	}
+	m.swap(context.Background(), next)
+	return nil
+}
+
+// WatchSIGHUP starts a goroutine that calls Reload whenever SIGHUP is
+// received. The goroutine stops when ctx is cancelled.
+func (m *ConfigManager) WatchSIGHUP(ctx context.Context) {
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGHUP)
+	go func() {
+		defer signal.Stop(ch)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ch:
+				if err := m.Reload(ctx); err != nil {
+					slog.Error("sched config: SIGHUP reload failed", "error", err)
+				} else {
+					slog.Info("sched config: reloaded via SIGHUP")
+				}
+			}
+		}
+	}()
+}
+
+// Reload re-reads the config and hot-swaps the live value. Idempotent.
+func (m *ConfigManager) Reload(ctx context.Context) error {
+	next, err := loadSchedConfig(m.path)
+	if err != nil {
+		return err
+	}
+	m.swap(ctx, next)
+	return nil
+}
+
+// swap atomically replaces the live config and records a diff in the audit
+// log when the config actually changed.
+func (m *ConfigManager) swap(ctx context.Context, next *SchedConfig) {
+	old := m.cfg.Swap(next)
+	if m.auditFn == nil {
+		return
+	}
+	oldJSON, _ := json.Marshal(old)
+	newJSON, _ := json.Marshal(next)
+	if string(oldJSON) == string(newJSON) {
+		return // no change; skip audit
+	}
+	m.auditFn(ctx, string(oldJSON), string(newJSON))
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Loading logic
+// ─────────────────────────────────────────────────────────────────────────────
+
+// loadSchedConfig builds a SchedConfig using the layering strategy:
+// defaults → YAML file → env-var overrides.
+func loadSchedConfig(path string) (*SchedConfig, error) {
+	cfg := DefaultSchedConfig()
+
+	// Layer 2: optional YAML file.
+	if path != "" {
+		data, err := os.ReadFile(path)
+		if err != nil && !os.IsNotExist(err) {
+			return nil, fmt.Errorf("sched config: read %s: %w", path, err)
+		}
+		if err == nil {
+			if err := yaml.Unmarshal(data, &cfg); err != nil {
+				return nil, fmt.Errorf("sched config: parse %s: %w", path, err)
+			}
+		}
+	}
+
+	// Layer 3: environment variable overrides.
+	applySchedEnvOverrides(&cfg)
+
+	return &cfg, nil
+}
+
+// applySchedEnvOverrides overlays environment-variable values on top of cfg.
+// Only non-empty env vars are applied; empty vars preserve the current value.
+func applySchedEnvOverrides(cfg *SchedConfig) {
+	// ── 15D detect ────────────────────────────────────────────────────────────
+	if v := posIntEnv("DETECT_DOMESTIC_FAIL_NUM"); v > 0 {
+		cfg.Detect.DomesticFailNumerator = v
+	}
+	if v := posIntEnv("DETECT_DOMESTIC_FAIL_DEN"); v > 0 {
+		cfg.Detect.DomesticFailDenominator = v
+	}
+	if v := posIntEnv("DETECT_SUSPECT_STREAK_MIN"); v > 0 {
+		cfg.Detect.SuspectStreakMin = v
+	}
+	if v := posFloatEnv("DETECT_TRAFFIC_DROP_THRESHOLD"); v > 0 {
+		cfg.Detect.TrafficDropThreshold = v
+	}
+	if v := posIntEnv("DETECT_TRAFFIC_BASELINE_MIN"); v > 0 {
+		cfg.Detect.TrafficBaselineMin = v
+	}
+	if v := posIntEnv("DETECT_CONFIRMED_STREAK_MIN"); v > 0 {
+		cfg.Detect.ConfirmedStreakMin = v
+	}
+	if v := posIntEnv("DETECT_RECOVER_STREAK_MIN"); v > 0 {
+		cfg.Detect.RecoverStreakMin = v
+	}
+	if v := posIntEnv("DETECT_SUSPECT_WEIGHT"); v > 0 {
+		cfg.Detect.SuspectWeight = v
+	}
+
+	// ── 15F capacity / breaker ────────────────────────────────────────────────
+	if v := posFloatEnv("CAP_WATERMARK_THRESHOLD"); v > 0 {
+		cfg.Capacity.WatermarkThreshold = v
+	}
+	if v := posIntEnv("CAP_ALERT_SUPPRESS_MIN"); v > 0 {
+		cfg.Capacity.AlertSuppressMin = v
+	}
+	if v := posIntEnv("CAP_BREAKER_WINDOW_MIN"); v > 0 {
+		cfg.Capacity.BreakerWindowMin = v
+	}
+	if v := posIntEnv("CAP_BREAKER_FRACTION_PCT"); v > 0 {
+		cfg.Capacity.BreakerFractionPct = v
+	}
+	if v := posIntEnv("CAP_BREAKER_MIN_N"); v > 0 {
+		cfg.Capacity.BreakerMinN = v
+	}
+	if v := posIntEnv("CAP_MANUAL_ALERT_THRESH"); v > 0 {
+		cfg.Capacity.ManualAlertThresh = v
+	}
+	if v := os.Getenv("CAP_EXPECTED_PROBE_IDS"); v != "" {
+		cfg.Capacity.ExpectedProbeIDs = splitCSVEnv(v)
+	}
+
+	// ── 15E probe / grayscale ─────────────────────────────────────────────────
+	if v := posIntEnv("PROBE_GRAYSCALE_INTERVAL_H"); v > 0 {
+		cfg.Probe.GrayscaleIntervalHours = v
+	}
+	if v := posIntEnv("PROBE_TIMEOUT_MIN"); v > 0 {
+		cfg.Probe.ProbeTimeoutMin = v
+	}
+	if v := posIntEnv("PROBE_MAX_ATTEMPTS"); v > 0 {
+		cfg.Probe.MaxAttempts = v
+	}
+	if v := posIntEnv("PROBE_CYCLES_REQUIRED"); v > 0 {
+		cfg.Probe.ProbeCyclesRequired = v
+	}
+}
+
+// posIntEnv reads key as a positive integer. Returns 0 if the env var is
+// unset, empty, zero, or non-parseable (logs a warning for parse errors).
+func posIntEnv(key string) int {
+	v := os.Getenv(key)
+	if v == "" {
+		return 0
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil || n <= 0 {
+		if err != nil {
+			slog.Warn("sched config: invalid integer env var", "key", key, "value", v)
+		}
+		return 0
+	}
+	return n
+}
+
+// posFloatEnv reads key as a positive float64. Returns 0 on error.
+func posFloatEnv(key string) float64 {
+	v := os.Getenv(key)
+	if v == "" {
+		return 0
+	}
+	f, err := strconv.ParseFloat(v, 64)
+	if err != nil || f <= 0 {
+		if err != nil {
+			slog.Warn("sched config: invalid float env var", "key", key, "value", v)
+		}
+		return 0
+	}
+	return f
+}
+
+// splitCSVEnv splits a comma-separated env-var value, trimming whitespace and
+// dropping empty fields.
+func splitCSVEnv(s string) []string {
+	parts := strings.Split(s, ",")
+	out := parts[:0]
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}

server/internal/scheduler/orchestrate/config_test.go

@@ -0,0 +1,289 @@
+package orchestrate_test
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Default values
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestDefaultSchedConfig(t *testing.T) {
+	cfg := orchestrate.DefaultSchedConfig()
+
+	// Spot-check a representative value from each section.
+	if cfg.Detect.DomesticFailNumerator != 2 {
+		t.Errorf("Detect.DomesticFailNumerator = %d; want 2", cfg.Detect.DomesticFailNumerator)
+	}
+	if cfg.Capacity.WatermarkThreshold != 0.70 {
+		t.Errorf("Capacity.WatermarkThreshold = %f; want 0.70", cfg.Capacity.WatermarkThreshold)
+	}
+	if cfg.Capacity.BreakerMinN != 3 {
+		t.Errorf("Capacity.BreakerMinN = %d; want 3", cfg.Capacity.BreakerMinN)
+	}
+	if cfg.Probe.GrayscaleIntervalHours != 6 {
+		t.Errorf("Probe.GrayscaleIntervalHours = %d; want 6", cfg.Probe.GrayscaleIntervalHours)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: YAML file overrides defaults
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestConfigManagerYAMLOverrides(t *testing.T) {
+	yaml := `
+detect:
+  suspect_streak_min: 5
+  confirmed_streak_min: 12
+capacity:
+  watermark_threshold: 0.60
+  breaker_min_n: 5
+  expected_probe_ids:
+    - pb-1
+    - pb-2
+probe:
+  grayscale_interval_hours: 8
+  max_attempts: 4
+`
+	tmpFile, err := writeTempYAML(t, yaml)
+	if err != nil {
+		t.Fatalf("writeTempYAML: %v", err)
+	}
+
+	m := orchestrate.NewConfigManager(tmpFile, nil)
+	if err := m.Load(); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+
+	cfg := m.Current()
+
+	if cfg.Detect.SuspectStreakMin != 5 {
+		t.Errorf("Detect.SuspectStreakMin = %d; want 5", cfg.Detect.SuspectStreakMin)
+	}
+	if cfg.Detect.ConfirmedStreakMin != 12 {
+		t.Errorf("Detect.ConfirmedStreakMin = %d; want 12", cfg.Detect.ConfirmedStreakMin)
+	}
+	if cfg.Capacity.WatermarkThreshold != 0.60 {
+		t.Errorf("Capacity.WatermarkThreshold = %f; want 0.60", cfg.Capacity.WatermarkThreshold)
+	}
+	if cfg.Capacity.BreakerMinN != 5 {
+		t.Errorf("Capacity.BreakerMinN = %d; want 5", cfg.Capacity.BreakerMinN)
+	}
+	if len(cfg.Capacity.ExpectedProbeIDs) != 2 {
+		t.Errorf("ExpectedProbeIDs count = %d; want 2", len(cfg.Capacity.ExpectedProbeIDs))
+	} else {
+		if cfg.Capacity.ExpectedProbeIDs[0] != "pb-1" || cfg.Capacity.ExpectedProbeIDs[1] != "pb-2" {
+			t.Errorf("ExpectedProbeIDs = %v; want [pb-1 pb-2]", cfg.Capacity.ExpectedProbeIDs)
+		}
+	}
+	if cfg.Probe.GrayscaleIntervalHours != 8 {
+		t.Errorf("Probe.GrayscaleIntervalHours = %d; want 8", cfg.Probe.GrayscaleIntervalHours)
+	}
+	if cfg.Probe.MaxAttempts != 4 {
+		t.Errorf("Probe.MaxAttempts = %d; want 4", cfg.Probe.MaxAttempts)
+	}
+
+	// Unset fields must retain defaults.
+	if cfg.Detect.DomesticFailNumerator != 2 {
+		t.Errorf("Detect.DomesticFailNumerator should be default 2; got %d", cfg.Detect.DomesticFailNumerator)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Environment variable overrides YAML
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestConfigManagerEnvOverridesYAML(t *testing.T) {
+	yaml := `
+detect:
+  suspect_streak_min: 5
+capacity:
+  breaker_min_n: 7
+`
+	tmpFile, err := writeTempYAML(t, yaml)
+	if err != nil {
+		t.Fatalf("writeTempYAML: %v", err)
+	}
+
+	// Set env vars AFTER writing YAML to ensure they override.
+	t.Setenv("DETECT_SUSPECT_STREAK_MIN", "9")
+	t.Setenv("CAP_BREAKER_MIN_N", "15")
+
+	m := orchestrate.NewConfigManager(tmpFile, nil)
+	if err := m.Load(); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+
+	cfg := m.Current()
+
+	// Env var should override YAML value.
+	if cfg.Detect.SuspectStreakMin != 9 {
+		t.Errorf("Detect.SuspectStreakMin = %d; want 9 (env override)", cfg.Detect.SuspectStreakMin)
+	}
+	if cfg.Capacity.BreakerMinN != 15 {
+		t.Errorf("Capacity.BreakerMinN = %d; want 15 (env override)", cfg.Capacity.BreakerMinN)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: SCHED_CONFIG_FILE env var overrides path argument
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestConfigManagerSCHED_CONFIG_FILE(t *testing.T) {
+	yaml := `
+capacity:
+  breaker_window_min: 120
+`
+	tmpFile, err := writeTempYAML(t, yaml)
+	if err != nil {
+		t.Fatalf("writeTempYAML: %v", err)
+	}
+
+	t.Setenv("SCHED_CONFIG_FILE", tmpFile)
+
+	// Pass a wrong path as arg; env var should win.
+	m := orchestrate.NewConfigManager("/nonexistent/path.yaml", nil)
+	if err := m.Load(); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+
+	if cfg := m.Current(); cfg.Capacity.BreakerWindowMin != 120 {
+		t.Errorf("BreakerWindowMin = %d; want 120 (read from SCHED_CONFIG_FILE)", cfg.Capacity.BreakerWindowMin)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Reload changes values and triggers audit callback with diff
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestConfigManagerReloadWithAudit(t *testing.T) {
+	yaml1 := `
+capacity:
+  watermark_threshold: 0.70
+  breaker_min_n: 3
+`
+	tmpFile, err := writeTempYAML(t, yaml1)
+	if err != nil {
+		t.Fatalf("writeTempYAML: %v", err)
+	}
+
+	var auditCalls []struct{ old, new string }
+	auditFn := func(ctx context.Context, old, new string) {
+		auditCalls = append(auditCalls, struct{ old, new string }{old, new})
+	}
+
+	m := orchestrate.NewConfigManager(tmpFile, orchestrate.AuditFn(auditFn))
+	if err := m.Load(); err != nil {
+		t.Fatalf("initial Load: %v", err)
+	}
+
+	initialThreshold := m.Current().Capacity.WatermarkThreshold
+
+	// Rewrite the YAML with a changed value.
+	yaml2 := `
+capacity:
+  watermark_threshold: 0.55
+  breaker_min_n: 5
+`
+	if err := os.WriteFile(tmpFile, []byte(yaml2), 0o644); err != nil {
+		t.Fatalf("rewrite YAML: %v", err)
+	}
+
+	// Reload should pick up the new values.
+	if err := m.Reload(context.Background()); err != nil {
+		t.Fatalf("Reload: %v", err)
+	}
+
+	newThreshold := m.Current().Capacity.WatermarkThreshold
+
+	if initialThreshold == newThreshold {
+		t.Errorf("WatermarkThreshold not updated after reload: both = %f", newThreshold)
+	}
+	if newThreshold != 0.55 {
+		t.Errorf("WatermarkThreshold after reload = %f; want 0.55", newThreshold)
+	}
+	if m.Current().Capacity.BreakerMinN != 5 {
+		t.Errorf("BreakerMinN after reload = %d; want 5", m.Current().Capacity.BreakerMinN)
+	}
+
+	// Audit callback must have been invoked once (old != new).
+	if len(auditCalls) != 1 {
+		t.Errorf("audit callback count = %d; want 1", len(auditCalls))
+	} else {
+		if auditCalls[0].old == auditCalls[0].new {
+			t.Error("audit callback old == new; diff expected")
+		}
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: Reload with no change → audit NOT called (diff is empty)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestConfigManagerReloadNoChangeSkipsAudit(t *testing.T) {
+	yaml := `
+capacity:
+  watermark_threshold: 0.75
+`
+	tmpFile, err := writeTempYAML(t, yaml)
+	if err != nil {
+		t.Fatalf("writeTempYAML: %v", err)
+	}
+
+	auditCount := 0
+	auditFn := func(_ context.Context, _, _ string) { auditCount++ }
+
+	m := orchestrate.NewConfigManager(tmpFile, orchestrate.AuditFn(auditFn))
+	if err := m.Load(); err != nil {
+		t.Fatalf("initial Load: %v", err)
+	}
+
+	// Reload without changing the YAML file.
+	if err := m.Reload(context.Background()); err != nil {
+		t.Fatalf("Reload: %v", err)
+	}
+
+	// No diff → audit should NOT be called on the second load.
+	// (First Load always swaps from defaults to YAML; that may be different.)
+	// What we really check: the reload didn't trigger another audit if nothing changed.
+	// The auditFn may have been called 1 time (from Load) but NOT from Reload.
+	//
+	// Actually, Load also calls swap. If the defaults differ from the YAML,
+	// the initial Load increments auditCount. Then Reload sees no diff → no extra call.
+	countAfterReload := auditCount
+
+	// Reload again — still unchanged → same count.
+	if err := m.Reload(context.Background()); err != nil {
+		t.Fatalf("second Reload: %v", err)
+	}
+	if auditCount != countAfterReload {
+		t.Errorf("audit called %d times after second reload; expected 0 extra calls (no diff)", auditCount-countAfterReload)
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Test: comma-separated env var for ExpectedProbeIDs
+// ─────────────────────────────────────────────────────────────────────────────
+
+func TestConfigManagerProbeIDsFromEnv(t *testing.T) {
+	t.Setenv("CAP_EXPECTED_PROBE_IDS", "alpha, beta , gamma")
+
+	m := orchestrate.NewConfigManager("", nil)
+	if err := m.Load(); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	ids := m.Current().Capacity.ExpectedProbeIDs
+	if len(ids) != 3 {
+		t.Fatalf("ExpectedProbeIDs = %v; want 3 entries", ids)
+	}
+	want := []string{"alpha", "beta", "gamma"}
+	for i, w := range want {
+		if ids[i] != w {
+			t.Errorf("ids[%d] = %q; want %q", i, ids[i], w)
+		}
+	}
+}