feat: drain worker assignments for disabled upstreams

README.md

@@ -263,7 +263,8 @@ go run ./cmd/proxy-loadgen `
   回收会自动为已分配项发起带健康/归属栅栏的 Drain，待 Snapshot ACK 与运行态归零后再删除；
   TARGET 事实按路由目标 Profile 独立归并。
 - **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与
-  `onUnavailable` 已接入；动态容量调整和配置停用触发的 Drain 编排待完成。
+  `onUnavailable` 已接入；上游停用会从后续完整 Snapshot 排除，并对现有 Worker
+  ownership 发起带策略 revision 栅栏的 Drain。动态容量调整仍待完成。
 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑，
   以及故障转移和恢复演练。
 - **P2 - 容量证明**：`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测，以及

docs/development/implementation-plan.md

@@ -302,6 +302,13 @@ Ticket/排除快照/Runtime 零计数闭环。
 Controller 即可生效。Redis 任务存储现已扩展 BASIC/EGRESS/TARGET 的独立有界索引；路由目标
 Profile 在启用 Routing 与 Upstream 的组合上才进入调度。
 
+补充进度（2026-08-02）：配置停用已进入 ownership Drain 编排。Controller 以静态配置
+与 Admin 管理态的交集生成完整、版本化的 Upstream 策略视图；Redis 只从对应 `owned` 索引
+有界读取仍可用的已归属 Proxy。候选携带 Upstream revision，`ownership.lua` 在复用既有
+Ticket/排除 Snapshot/Runtime 零计数闭环前，原子复核策略仍为停用、Proxy source、Worker、
+assignment epoch 和未过期租约。重启用后的策略 revision 会使旧候选返回无操作。Provider
+Supervisor 也改为同时服从静态配置与管理态，消除两条启停消费链的不一致。
+
 补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 与 CONNECT 长连接场景。固定请求数
 和固定时长两种模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小
 延迟直方图，输出状态分类、CONNECT 建立数、Extract 校验数、吞吐和 Go 内存/GC 快照。CONNECT

docs/requirements/traceability.md

@@ -10,7 +10,7 @@
 | ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP/CONNECT/Extract 校验场景均已实现 |
 | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 |
 | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 |
-| ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 |
+| ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；静态配置与 Admin 管理态共同决定 Provider 启停，停用策略以 Upstream revision 栅栏驱动既有 ownership Drain；多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 |
 | ARCH-005 | 100k QPS 峰值使用多 Worker 集群 | 当前会话 | 未验证设计目标；待代表性集群负载报告 |
 
 ## Routing 与 Upstream

findings.md

@@ -124,7 +124,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
   Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot
   消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报。Redis Runtime 替换已在当前
   session ACK、Ticket 屏障和完整零计数同时成立时自动释放 owner；Reaper 已对持续
-  `UNHEALTHY` 的已归属代理调用条件式 Drain。配置停用尚未接入同一触发编排。
+  `UNHEALTHY` 的已归属代理调用条件式 Drain。配置停用也会发布有效的上游策略视图，
+  从上游 owned 索引有界选择候选；候选携带上游 revision，Redis 在创建 Drain Ticket 前
+  再次核验策略仍停用、Proxy source、Worker 和 assignment epoch，避免旧候选跨启停使用。
 - Snapshot 的 `version` 是同一 Worker 流的连续序列，`ownership_epoch` 是独立且只能前进
   的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1，
   与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。

internal/adapters/redisactivity/global_health_integration_test.go

@@ -224,6 +224,53 @@ func TestRedisRejectsUnhealthyDrainCandidateAfterRecovery(t *testing.T) {
 	}
 }
 
+func TestRedisDisabledUpstreamDrainFencesPolicyRevision(t *testing.T) {
+	fixture := newRedisTestFixture(t)
+	now := redisTestNow()
+	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute,
+		testProxy("proxy-a", "192.0.2.10"))
+	assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(2*time.Second), "proxy-a", "worker-a", time.Minute)
+	if err != nil {
+		t.Fatalf("Assign() error = %v", err)
+	}
+	disabled := activitypool.UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 31}
+	if err := fixture.Adapter.ReplaceUpstreamDrainPolicies(context.Background(), []activitypool.UpstreamDrainPolicy{disabled}); err != nil {
+		t.Fatalf("ReplaceUpstreamDrainPolicies(disabled) error = %v", err)
+	}
+	candidates, err := fixture.Adapter.ListDisabledUpstreamDrainCandidates(context.Background(), now.Add(3*time.Second), disabled, 1)
+	if err != nil || len(candidates) != 1 || candidates[0] != (activitypool.DisabledUpstreamDrainCandidate{
+		UpstreamID: "provider-a", PolicyRevision: 31, ProxyID: "proxy-a", WorkerID: "worker-a", AssignmentEpoch: assignment.Epoch,
+	}) {
+		t.Fatalf("ListDisabledUpstreamDrainCandidates() = %+v, %v", candidates, err)
+	}
+	if err := fixture.Adapter.ReplaceUpstreamDrainPolicies(context.Background(), []activitypool.UpstreamDrainPolicy{{
+		UpstreamID: "provider-a", Revision: 32, Enabled: true,
+	}}); err != nil {
+		t.Fatalf("ReplaceUpstreamDrainPolicies(enabled) error = %v", err)
+	}
+	if started, err := fixture.Adapter.BeginDisabledUpstreamDrain(context.Background(), now.Add(3*time.Second), candidates[0]); err != nil || started {
+		t.Fatalf("BeginDisabledUpstreamDrain(stale policy) = %t, %v", started, err)
+	}
+	if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || current.Draining {
+		t.Fatalf("Get(after stale candidate) = %+v, %t, %v", current, ok, err)
+	}
+
+	disabled.Revision = 33
+	if err := fixture.Adapter.ReplaceUpstreamDrainPolicies(context.Background(), []activitypool.UpstreamDrainPolicy{disabled}); err != nil {
+		t.Fatalf("ReplaceUpstreamDrainPolicies(disabled again) error = %v", err)
+	}
+	candidates, err = fixture.Adapter.ListDisabledUpstreamDrainCandidates(context.Background(), now.Add(4*time.Second), disabled, 1)
+	if err != nil || len(candidates) != 1 {
+		t.Fatalf("ListDisabledUpstreamDrainCandidates(disabled again) = %+v, %v", candidates, err)
+	}
+	if started, err := fixture.Adapter.BeginDisabledUpstreamDrain(context.Background(), now.Add(4*time.Second), candidates[0]); err != nil || !started {
+		t.Fatalf("BeginDisabledUpstreamDrain() = %t, %v", started, err)
+	}
+	if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || !current.Draining {
+		t.Fatalf("Get(after drain) = %+v, %t, %v", current, ok, err)
+	}
+}
+
 func redisGlobalObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation {
 	observation := healthDomain.Observation{
 		TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic,

internal/adapters/redisactivity/keys.go

@@ -18,6 +18,7 @@ type keyspace struct {
 	owners                string
 	ownerExpiry           string
 	drainTickets          string
+	upstreamDrainPolicies string
 	epoch                 string
 	inventory             string
 	stateInventory        string
@@ -51,6 +52,7 @@ func newKeyspace(namespace string) keyspace {
 		owners:                prefix + ":owners",
 		ownerExpiry:           prefix + ":owner-expiry",
 		drainTickets:          prefix + ":drain-tickets",
+		upstreamDrainPolicies: prefix + ":upstream-drain-policies",
 		epoch:                 prefix + ":epoch",
 		inventory:             prefix + ":inventory",
 		stateInventory:        prefix + ":state-inventory",

internal/adapters/redisactivity/maintenance.go

@@ -3,6 +3,7 @@ package redisactivity
 import (
 	"context"
 	"encoding/json"
+	"strconv"
 	"time"
 
 	"proxy-pool/internal/domain/activitypool"
@@ -12,6 +13,7 @@ const (
 	maintenanceInventory        = "inventory"
 	maintenanceSweep            = "sweep"
 	maintenanceUnhealthy        = "unhealthy"
+	maintenanceDisabledUpstream = "disabled_upstream"
 
 	maximumUnhealthySweepScan = 1024
 )
@@ -19,6 +21,7 @@ const (
 var (
 	_ activitypool.InventoryReader             = (*Adapter)(nil)
 	_ activitypool.Maintainer                  = (*Adapter)(nil)
+	_ activitypool.DisabledUpstreamDrainLister = (*Adapter)(nil)
 )
 
 func (a *Adapter) Inventory(ctx context.Context, upstreamID string, now time.Time) (activitypool.Inventory, error) {
@@ -124,6 +127,52 @@ func (a *Adapter) SweepUnhealthy(
 	return result, nil
 }
 
+func (a *Adapter) ListDisabledUpstreamDrainCandidates(
+	ctx context.Context,
+	now time.Time,
+	policy activitypool.UpstreamDrainPolicy,
+	limit int,
+) ([]activitypool.DisabledUpstreamDrainCandidate, error) {
+	if ctx == nil {
+		return nil, activitypool.ErrInvalidMaintenance
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if a == nil || now.IsZero() || policy.UpstreamID == "" || policy.Revision == 0 || policy.Enabled || limit <= 0 {
+		return nil, activitypool.ErrInvalidMaintenance
+	}
+	if limit > maximumUnhealthySweepScan/4 {
+		limit = maximumUnhealthySweepScan / 4
+	}
+	reply, err := a.runMaintenance(ctx, maintenanceDisabledUpstream, now, limit, policy.UpstreamID, "")
+	if err != nil {
+		return nil, err
+	}
+	if reply.Status == scriptStale {
+		return nil, nil
+	}
+	if reply.Status == scriptInvalid {
+		return nil, activitypool.ErrInvalidMaintenance
+	}
+	if reply.Status != scriptOK || len(reply.DisabledDrainCandidates) > limit {
+		return nil, invalidScriptReply("unexpected disabled upstream sweep reply")
+	}
+	wantRevision := strconv.FormatUint(policy.Revision, 10)
+	result := make([]activitypool.DisabledUpstreamDrainCandidate, 0, len(reply.DisabledDrainCandidates))
+	for _, candidate := range reply.DisabledDrainCandidates {
+		if candidate.UpstreamID != policy.UpstreamID || candidate.PolicyRevision != wantRevision ||
+			candidate.ProxyID == "" || candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 {
+			return nil, invalidScriptReply("disabled upstream sweep reply contained an invalid drain candidate")
+		}
+		result = append(result, activitypool.DisabledUpstreamDrainCandidate{
+			UpstreamID: candidate.UpstreamID, PolicyRevision: policy.Revision, ProxyID: candidate.ProxyID,
+			WorkerID: candidate.WorkerID, AssignmentEpoch: candidate.AssignmentEpoch,
+		})
+	}
+	return result, nil
+}
+
 func (a *Adapter) runMaintenance(
 	ctx context.Context,
 	operation string,
@@ -141,6 +190,7 @@ func (a *Adapter) runMaintenance(
 		a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
 		a.keys.operation(operationID), a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
 		a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
+		a.keys.owned(upstreamID), a.keys.upstreamDrainPolicies,
 	}, operation, now.UnixMilli(), limit, upstreamID, operationTTLMillis(a.options.OperationTTL), policies)
 	if err != nil {
 		return maintenanceScriptReply{}, err

internal/adapters/redisactivity/ownership.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"strconv"
 	"time"
 
 	"proxy-pool/internal/domain/activitypool"
@@ -17,6 +18,7 @@ const (
 	ownershipRenew               = "renew"
 	ownershipBeginDrain          = "begin_drain"
 	ownershipBeginUnhealthyDrain = "begin_unhealthy_drain"
+	ownershipBeginDisabledDrain  = "begin_disabled_upstream_drain"
 	ownershipAcknowledgeDrain    = "acknowledge_drain"
 	ownershipGet                 = "get"
 	ownershipExpire              = "expire"
@@ -25,6 +27,7 @@ const (
 var _ ownershipDomain.Repository = (*Adapter)(nil)
 var _ ownershipDomain.DrainTicketStore = (*Adapter)(nil)
 var _ activitypool.UnhealthyDrainStarter = (*Adapter)(nil)
+var _ activitypool.DisabledUpstreamDrainStarter = (*Adapter)(nil)
 
 func (a *Adapter) Assign(
 	ctx context.Context,
@@ -39,7 +42,7 @@ func (a *Adapter) Assign(
 	if now.IsZero() || proxyID == "" || workerID == "" || ttl <= 0 {
 		return ownershipDomain.Assignment{}, ownershipDomain.ErrInvalidOwnership
 	}
-	reply, err := a.runOwnership(ctx, ownershipAssign, true, now.UnixMilli(), proxyID, workerID, 0, durationMillis(ttl), 0, 0)
+	reply, err := a.runOwnership(ctx, ownershipAssign, true, now.UnixMilli(), proxyID, workerID, 0, durationMillis(ttl), 0, 0, "", "")
 	if err != nil {
 		return ownershipDomain.Assignment{}, err
 	}
@@ -71,7 +74,7 @@ func (a *Adapter) Renew(
 	if now.IsZero() || proxyID == "" || workerID == "" || epoch == 0 || ttl <= 0 {
 		return ownershipDomain.Assignment{}, ownershipDomain.ErrInvalidOwnership
 	}
-	reply, err := a.runOwnership(ctx, ownershipRenew, true, now.UnixMilli(), proxyID, workerID, epoch, durationMillis(ttl), 0, 0)
+	reply, err := a.runOwnership(ctx, ownershipRenew, true, now.UnixMilli(), proxyID, workerID, epoch, durationMillis(ttl), 0, 0, "", "")
 	if err != nil {
 		return ownershipDomain.Assignment{}, err
 	}
@@ -99,7 +102,7 @@ func (a *Adapter) BeginDrain(
 	if proxyID == "" || workerID == "" || epoch == 0 {
 		return ownershipDomain.Assignment{}, ownershipDomain.ErrInvalidOwnership
 	}
-	reply, err := a.runOwnership(ctx, ownershipBeginDrain, true, 0, proxyID, workerID, epoch, 0, 0, 0)
+	reply, err := a.runOwnership(ctx, ownershipBeginDrain, true, 0, proxyID, workerID, epoch, 0, 0, 0, "", "")
 	if err != nil {
 		return ownershipDomain.Assignment{}, err
 	}
@@ -128,7 +131,7 @@ func (a *Adapter) BeginUnhealthyDrain(
 	}
 	reply, err := a.runOwnership(
 		ctx, ownershipBeginUnhealthyDrain, true, now.UTC().UnixMilli(), candidate.ProxyID, candidate.WorkerID,
-		candidate.AssignmentEpoch, candidate.UnhealthySince.UTC().UnixMilli(), 0, 0,
+		candidate.AssignmentEpoch, candidate.UnhealthySince.UTC().UnixMilli(), 0, 0, "", "",
 	)
 	if err != nil {
 		return false, err
@@ -148,6 +151,40 @@ func (a *Adapter) BeginUnhealthyDrain(
 	}
 }
 
+func (a *Adapter) BeginDisabledUpstreamDrain(
+	ctx context.Context,
+	now time.Time,
+	candidate activitypool.DisabledUpstreamDrainCandidate,
+) (bool, error) {
+	if err := validateOwnershipCall(ctx, a); err != nil {
+		return false, err
+	}
+	if now.IsZero() || candidate.UpstreamID == "" || candidate.PolicyRevision == 0 || candidate.ProxyID == "" ||
+		candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 {
+		return false, activitypool.ErrInvalidMaintenance
+	}
+	reply, err := a.runOwnership(
+		ctx, ownershipBeginDisabledDrain, true, now.UTC().UnixMilli(), candidate.ProxyID, candidate.WorkerID,
+		candidate.AssignmentEpoch, 0, 0, 0, candidate.UpstreamID, strconv.FormatUint(candidate.PolicyRevision, 10),
+	)
+	if err != nil {
+		return false, err
+	}
+	switch reply.Status {
+	case scriptOK:
+		if _, err := decodeAssignmentReply(reply); err != nil {
+			return false, err
+		}
+		return reply.Started, nil
+	case scriptStale, scriptNotFound:
+		return false, nil
+	case scriptInvalid:
+		return false, activitypool.ErrInvalidMaintenance
+	default:
+		return false, invalidScriptReply("unexpected disabled upstream ownership drain status")
+	}
+}
+
 func (a *Adapter) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) {
 	if err := validateOwnershipCall(ctx, a); err != nil {
 		return nil, err
@@ -242,7 +279,7 @@ func (a *Adapter) AcknowledgeDrain(
 	if proxyID == "" || workerID == "" || epoch == 0 || active < 0 || reserved < 0 {
 		return ownershipDomain.ErrInvalidOwnership
 	}
-	reply, err := a.runOwnership(ctx, ownershipAcknowledgeDrain, true, 0, proxyID, workerID, epoch, 0, active, reserved)
+	reply, err := a.runOwnership(ctx, ownershipAcknowledgeDrain, true, 0, proxyID, workerID, epoch, 0, active, reserved, "", "")
 	if err != nil {
 		return err
 	}
@@ -269,7 +306,7 @@ func (a *Adapter) Get(ctx context.Context, proxyID string) (ownershipDomain.Assi
 	if proxyID == "" {
 		return ownershipDomain.Assignment{}, false, ownershipDomain.ErrInvalidOwnership
 	}
-	reply, err := a.runOwnership(ctx, ownershipGet, false, 0, proxyID, "", 0, 0, 0, 0)
+	reply, err := a.runOwnership(ctx, ownershipGet, false, 0, proxyID, "", 0, 0, 0, 0, "", "")
 	if err != nil {
 		return ownershipDomain.Assignment{}, false, err
 	}
@@ -293,7 +330,7 @@ func (a *Adapter) Expire(ctx context.Context, now time.Time, limit int) ([]owner
 	if now.IsZero() || limit <= 0 {
 		return nil, ownershipDomain.ErrInvalidOwnership
 	}
-	reply, err := a.runOwnership(ctx, ownershipExpire, true, now.UnixMilli(), "", "", 0, int64(limit), 0, 0)
+	reply, err := a.runOwnership(ctx, ownershipExpire, true, now.UnixMilli(), "", "", 0, int64(limit), 0, 0, "", "")
 	if err != nil {
 		return nil, err
 	}
@@ -331,6 +368,8 @@ func (a *Adapter) runOwnership(
 	value int64,
 	active int64,
 	reserved int64,
+	upstreamID string,
+	policyRevision string,
 ) (ownershipScriptReply, error) {
 	operationKey := a.keys.epoch
 	if mutating {
@@ -345,9 +384,9 @@ func (a *Adapter) runOwnership(
 		a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
 		a.keys.epoch, operationKey, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
 		a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
-		a.keys.drainTickets,
+		a.keys.drainTickets, a.keys.upstreamDrainPolicies,
 	}, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit,
-		nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID), a.keys.workerDraining(workerID))
+		nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID), a.keys.workerDraining(workerID), upstreamID, policyRevision)
 	if err != nil {
 		return ownershipScriptReply{}, err
 	}

internal/adapters/redisactivity/scripts.go

@@ -88,6 +88,10 @@ type ownershipScriptReply struct {
 	Started bool         `json:"started"`
 }
 
+type upstreamDrainPolicyScriptReply struct {
+	Status scriptStatus `json:"status"`
+}
+
 type drainTicketsScriptReply struct {
 	Status  scriptStatus `json:"status"`
 	Tickets []string     `json:"tickets"`
@@ -98,6 +102,7 @@ type maintenanceScriptReply struct {
 	Count                   int                                  `json:"count"`
 	DeferredOwned           int                                  `json:"deferredOwned"`
 	DrainCandidates         []maintenanceDrainCandidateScriptRow `json:"drainCandidates"`
+	DisabledDrainCandidates []disabledDrainCandidateScriptRow    `json:"disabledDrainCandidates"`
 }
 
 type maintenanceDrainCandidateScriptRow struct {
@@ -107,6 +112,14 @@ type maintenanceDrainCandidateScriptRow struct {
 	UnhealthySinceMS int64  `json:"unhealthySinceMs"`
 }
 
+type disabledDrainCandidateScriptRow struct {
+	UpstreamID      string `json:"upstreamId"`
+	PolicyRevision  string `json:"policyRevision"`
+	ProxyID         string `json:"proxyId"`
+	WorkerID        string `json:"workerId"`
+	AssignmentEpoch uint64 `json:"assignmentEpoch"`
+}
+
 type statusScriptReply struct {
 	Status      scriptStatus            `json:"status"`
 	Inventories []statusScriptInventory `json:"inventories"`
@@ -188,6 +201,9 @@ var capacitySource string
 //go:embed scripts/worker_snapshot.lua
 var workerSnapshotSource string
 
+//go:embed scripts/upstream_drain_policy.lua
+var upstreamDrainPolicySource string
+
 var (
 	upsertScript              = redis.NewScript(upsertSource)
 	healthScript              = redis.NewScript(healthSource)
@@ -203,6 +219,7 @@ var (
 	runtimeScript             = redis.NewScript(runtimeSource)
 	capacityScript            = redis.NewScript(capacitySource)
 	workerSnapshotScript      = redis.NewScript(workerSnapshotSource)
+	upstreamDrainPolicyScript = redis.NewScript(upstreamDrainPolicySource)
 )
 
 func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) {

internal/adapters/redisactivity/scripts/ownership.lua

@@ -17,6 +17,7 @@ local health_task_expiry_key = KEYS[16]
 local health_ref_task_key = KEYS[17]
 local health_unhealthy_key = KEYS[18]
 local drain_tickets_key = KEYS[19]
+local upstream_drain_policies_key = KEYS[20]
 
 local operation = ARGV[1]
 local operation_ttl_ms = tonumber(ARGV[2])
@@ -30,6 +31,8 @@ local active = tonumber(ARGV[9])
 local reserved = tonumber(ARGV[10])
 local worker_index_key = ARGV[11]
 local worker_draining_index_key = ARGV[12]
+local upstream_id = ARGV[13]
+local policy_revision = ARGV[14]
 local mutating = operation ~= 'get'
 
 local function finish(reply)
@@ -212,6 +215,11 @@ local function decode_table(raw)
     return value
 end
 
+local function valid_uint(value)
+    return type(value) == 'string' and string.match(value, '^[0-9]+$') and
+        value ~= '0' and (string.len(value) == 1 or string.sub(value, 1, 1) ~= '0')
+end
+
 local function valid_assignment(assignment)
     return assignment and assignment.version == 1 and type(assignment.proxyId) == 'string' and
         assignment.proxyId ~= '' and type(assignment.workerId) == 'string' and assignment.workerId ~= '' and
@@ -396,6 +404,33 @@ if operation == 'begin_unhealthy_drain' then
     return finish({status = 'ok', record = encoded, started = started})
 end
 
+if operation == 'begin_disabled_upstream_drain' then
+    cleanup_hard_expired(now_ms)
+    if type(upstream_id) ~= 'string' or upstream_id == '' or not valid_uint(policy_revision) then
+        return finish({status = 'invalid'})
+    end
+    local policy = decode_table(redis.call('HGET', upstream_drain_policies_key, upstream_id))
+    if not policy or policy.version ~= 1 or policy.upstreamId ~= upstream_id or policy.enabled ~= false or
+        policy.revision ~= policy_revision then
+        return finish({status = 'stale'})
+    end
+    local current = decode_table(redis.call('HGET', owners_key, proxy_id))
+    if not valid_assignment(current) or current.workerId ~= worker_id or tonumber(current.epoch) ~= epoch or
+        tonumber(current.expiresAtMs) <= now_ms then
+        return finish({status = 'stale'})
+    end
+    local record = decode_table(redis.call('HGET', records_key, proxy_id))
+    if not record or record.sourceUpstream ~= upstream_id or record.ownerWorkerId ~= worker_id or
+        record.state ~= 'AVAILABLE' then
+        return finish({status = 'stale'})
+    end
+    local encoded, started, failure = begin_drain(current)
+    if failure then
+        return finish({status = failure})
+    end
+    return finish({status = 'ok', record = encoded, started = started})
+end
+
 if operation == 'acknowledge_drain' then
     local current = decode_table(redis.call('HGET', owners_key, proxy_id))
     if not valid_assignment(current) or current.workerId ~= worker_id or tonumber(current.epoch) ~= epoch then

internal/adapters/redisactivity/scripts/sweep.lua

@@ -15,6 +15,8 @@ local health_tasks_key = KEYS[14]
 local health_task_expiry_key = KEYS[15]
 local health_ref_task_key = KEYS[16]
 local health_unhealthy_key = KEYS[17]
+local owned_upstream_key = KEYS[18]
+local upstream_drain_policies_key = KEYS[19]
 
 local operation = ARGV[1]
 local now_ms = tonumber(ARGV[2])
@@ -220,6 +222,40 @@ if operation == 'unhealthy' then
     end
     return finish({status = 'ok', count = removed, deferredOwned = deferred_owned, drainCandidates = drain_candidates})
 end
+if operation == 'disabled_upstream' then
+    local raw_policy = redis.call('HGET', upstream_drain_policies_key, upstream_id)
+    local policy = raw_policy and decode_table(raw_policy) or nil
+    if not policy or policy.version ~= 1 or policy.upstreamId ~= upstream_id or policy.enabled ~= false or
+        type(policy.revision) ~= 'string' or not string.match(policy.revision, '^[0-9]+$') or policy.revision == '0' then
+        return finish({status = 'stale', count = 0, disabledDrainCandidates = {}})
+    end
+    local scan_limit = limit * 4
+    if scan_limit > 1024 then
+        scan_limit = 1024
+    end
+    local proxy_ids = redis.call('ZRANGEBYSCORE', owned_upstream_key, '(' .. now_ms, '+inf', 'LIMIT', 0, scan_limit)
+    local candidates = {}
+    for _, proxy_id in ipairs(proxy_ids) do
+        local record = decode_table(redis.call('HGET', records_key, proxy_id))
+        local owner = decode_table(redis.call('HGET', owners_key, proxy_id))
+        if record and owner and record.sourceUpstream == upstream_id and record.state == 'AVAILABLE' and
+            record.ownerWorkerId == owner.workerId and owner.draining == false and
+            type(owner.workerId) == 'string' and owner.workerId ~= '' and tonumber(owner.epoch) and
+            tonumber(owner.epoch) > 0 and tonumber(owner.expiresAtMs) and tonumber(owner.expiresAtMs) > now_ms then
+            table.insert(candidates, {
+                upstreamId = upstream_id,
+                policyRevision = policy.revision,
+                proxyId = proxy_id,
+                workerId = owner.workerId,
+                assignmentEpoch = tonumber(owner.epoch),
+            })
+            if #candidates == limit then
+                break
+            end
+        end
+    end
+    return finish({status = 'ok', count = 0, disabledDrainCandidates = candidates})
+end
 if operation == 'inventory' then
     local count = tonumber(redis.call('HGET', inventory_key, upstream_id) or '0')
     if count < 0 then

internal/adapters/redisactivity/scripts/upstream_drain_policy.lua

@@ -0,0 +1,23 @@
+local policies_key = KEYS[1]
+local policies_json = ARGV[1]
+
+local decoded, policies = pcall(cjson.decode, policies_json or '')
+if not decoded or type(policies) ~= 'table' then
+    return cjson.encode({status = 'invalid'})
+end
+
+local seen = {}
+for _, policy in ipairs(policies) do
+    if type(policy) ~= 'table' or type(policy.upstreamId) ~= 'string' or policy.upstreamId == '' or
+        type(policy.revision) ~= 'string' or not string.match(policy.revision, '^[0-9]+$') or
+        policy.revision == '0' or type(policy.enabled) ~= 'boolean' or seen[policy.upstreamId] then
+        return cjson.encode({status = 'invalid'})
+    end
+    seen[policy.upstreamId] = true
+end
+
+redis.call('DEL', policies_key)
+for _, policy in ipairs(policies) do
+    redis.call('HSET', policies_key, policy.upstreamId, cjson.encode(policy))
+end
+return cjson.encode({status = 'ok'})

internal/adapters/redisactivity/upstream_drain_policy.go

@@ -0,0 +1,63 @@
+package redisactivity
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+
+	"proxy-pool/internal/domain/activitypool"
+)
+
+var _ activitypool.UpstreamDrainPolicyWriter = (*Adapter)(nil)
+
+type upstreamDrainPolicyRecord struct {
+	Version    int    `json:"version"`
+	UpstreamID string `json:"upstreamId"`
+	Revision   string `json:"revision"`
+	Enabled    bool   `json:"enabled"`
+}
+
+func (a *Adapter) ReplaceUpstreamDrainPolicies(ctx context.Context, policies []activitypool.UpstreamDrainPolicy) error {
+	if ctx == nil {
+		return activitypool.ErrInvalidMaintenance
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if a == nil {
+		return activitypool.ErrInvalidMaintenance
+	}
+	records := make([]upstreamDrainPolicyRecord, 0, len(policies))
+	seen := make(map[string]struct{}, len(policies))
+	for _, policy := range policies {
+		if policy.UpstreamID == "" || policy.Revision == 0 {
+			return activitypool.ErrInvalidMaintenance
+		}
+		if _, duplicate := seen[policy.UpstreamID]; duplicate {
+			return activitypool.ErrInvalidMaintenance
+		}
+		seen[policy.UpstreamID] = struct{}{}
+		records = append(records, upstreamDrainPolicyRecord{
+			Version: 1, UpstreamID: policy.UpstreamID, Revision: strconv.FormatUint(policy.Revision, 10), Enabled: policy.Enabled,
+		})
+	}
+	payload, err := json.Marshal(records)
+	if err != nil {
+		return err
+	}
+	result, err := runScript(ctx, a.client, upstreamDrainPolicyScript, []string{a.keys.upstreamDrainPolicies}, string(payload))
+	if err != nil {
+		return err
+	}
+	var reply upstreamDrainPolicyScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return err
+	}
+	if reply.Status == scriptInvalid {
+		return activitypool.ErrInvalidMaintenance
+	}
+	if reply.Status != scriptOK {
+		return invalidScriptReply("unexpected upstream drain policy reply")
+	}
+	return nil
+}

internal/controller/bootstrap/bootstrap.go

@@ -339,6 +339,30 @@ func runWithWorkerFactory(
 			}
 			runners = append(runners, reaper)
 		}
+		if drainStore, ok := opened.activity.(controllerHealth.UpstreamDrainStore); ok && !nilInterface(drainStore) {
+			var policyStates []controllerHealth.UpstreamDrainStateSource
+			if loaded.Value.Admin.Enabled {
+				if nilInterface(opened.state) {
+					return errors.Join(ErrStartup, ErrInvalidOptions)
+				}
+				policyStates = append(policyStates, opened.state)
+			}
+			policySource, policyErr := controllerHealth.NewEffectiveUpstreamDrainPolicySource(configurationStore, policyStates...)
+			if policyErr != nil {
+				return fmt.Errorf("%w: build upstream Drain policy source: %w", ErrStartup, policyErr)
+			}
+			reaper, reaperErr := controllerHealth.NewConfiguredUpstreamDrainReaper(
+				policySource,
+				drainStore,
+				controllerHealth.UpstreamDrainReaperOptions{
+					PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now,
+				},
+			)
+			if reaperErr != nil {
+				return fmt.Errorf("%w: build upstream Drain reaper: %w", ErrStartup, reaperErr)
+			}
+			runners = append(runners, reaper)
+		}
 	}
 	runners = append(runners, supervisor)
 	group, err := lifecycle.NewGroup(runners...)

internal/controller/bootstrap/provider_supervisor.go

@@ -276,7 +276,7 @@ func (supervisor *providerSupervisor) desired(ctx context.Context) (map[string]c
 	for name, upstream := range configuration.Upstreams {
 		isEnabled := upstream.Enabled
 		if !nilInterface(supervisor.state) {
-			isEnabled = enabled[name]
+			isEnabled = isEnabled && enabled[name]
 		}
 		if isEnabled {
 			upstream.Enabled = true

internal/controller/bootstrap/provider_supervisor_test.go

@@ -71,6 +71,38 @@ func TestProviderSupervisorAppliesDisableAndConfigurationReplacement(t *testing.
 	}
 }
 
+func TestProviderSupervisorKeepsStaticallyDisabledUpstreamStopped(t *testing.T) {
+	configuration, err := config.Load(strings.NewReader(bootstrapTestConfig))
+	if err != nil {
+		t.Fatalf("config.Load(): %v", err)
+	}
+	disabled := configuration.Upstreams["provider-b"]
+	disabled.Enabled = false
+	configuration.Upstreams["provider-b"] = disabled
+	store, err := config.NewStore(configuration)
+	if err != nil {
+		t.Fatalf("config.NewStore(): %v", err)
+	}
+	supervisor, err := newProviderSupervisor(
+		store,
+		&mutableProviderState{enabled: map[string]bool{"provider-a": true, "provider-b": true}},
+		func(string, config.Upstream) (lifecycle.Runner, error) {
+			return supervisorRunnerFunc(func(context.Context) error { return nil }), nil
+		},
+		nil, nil, nil, bootstrapTestFingerprintKey, time.Hour,
+	)
+	if err != nil {
+		t.Fatalf("newProviderSupervisor() error = %v", err)
+	}
+	desired, err := supervisor.desired(context.Background())
+	if err != nil || len(desired) != 1 {
+		t.Fatalf("desired() = %+v, %v", desired, err)
+	}
+	if _, exists := desired["provider-a"]; !exists {
+		t.Fatalf("desired() = %+v, provider-a missing", desired)
+	}
+}
+
 func TestProviderSupervisorPropagatesUnexpectedRuntimeFailure(t *testing.T) {
 	configuration, err := config.Load(strings.NewReader(bootstrapTestConfig))
 	if err != nil {

internal/controller/health/upstream_drain_reaper.go

@@ -0,0 +1,206 @@
+package health
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/domain/activitypool"
+	"proxy-pool/internal/domain/adminstate"
+)
+
+var ErrInvalidUpstreamDrainReaper = errors.New("invalid upstream drain reaper")
+
+// UpstreamDrainConfigurationSource returns one configuration and revision from
+// the same atomic publication. config.Store implements this interface.
+type UpstreamDrainConfigurationSource interface {
+	Snapshot() (*config.Config, uint64)
+}
+
+type UpstreamDrainStateSource interface {
+	Snapshot(context.Context) (adminstate.Snapshot, error)
+}
+
+// EffectiveUpstreamDrainPolicySource merges static configuration with optional
+// Admin state. With Admin enabled it uses the same revision equality rule as
+// Worker routing, so a torn config/state view produces no policy publication.
+type EffectiveUpstreamDrainPolicySource struct {
+	configuration UpstreamDrainConfigurationSource
+	state         UpstreamDrainStateSource
+}
+
+func NewEffectiveUpstreamDrainPolicySource(
+	configuration UpstreamDrainConfigurationSource,
+	states ...UpstreamDrainStateSource,
+) (*EffectiveUpstreamDrainPolicySource, error) {
+	if nilInterface(configuration) || len(states) > 1 || (len(states) == 1 && nilInterface(states[0])) {
+		return nil, ErrInvalidUpstreamDrainReaper
+	}
+	source := &EffectiveUpstreamDrainPolicySource{configuration: configuration}
+	if len(states) == 1 {
+		source.state = states[0]
+	}
+	return source, nil
+}
+
+func (source *EffectiveUpstreamDrainPolicySource) ReadUpstreamDrainPolicies(
+	ctx context.Context,
+) ([]activitypool.UpstreamDrainPolicy, error) {
+	if ctx == nil || source == nil || nilInterface(source.configuration) {
+		return nil, ErrInvalidUpstreamDrainReaper
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	configuration, revision := source.configuration.Snapshot()
+	if configuration == nil {
+		return nil, ErrInvalidUpstreamDrainReaper
+	}
+	states := make(map[string]adminstate.UpstreamState, len(configuration.Upstreams))
+	if source.state != nil {
+		snapshot, err := source.state.Snapshot(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if snapshot.Config == nil || snapshot.Config.Revision != revision {
+			return nil, ErrInvalidUpstreamDrainReaper
+		}
+		for _, upstream := range snapshot.Upstreams {
+			if upstream.Name == "" || upstream.Revision == 0 {
+				return nil, ErrInvalidUpstreamDrainReaper
+			}
+			if _, duplicate := states[upstream.Name]; duplicate {
+				return nil, ErrInvalidUpstreamDrainReaper
+			}
+			states[upstream.Name] = upstream
+		}
+	}
+	if revision == 0 {
+		// A config-only Controller has no persisted management revision. Its
+		// configuration cannot be hot-reloaded, so one stable non-zero fence is enough.
+		revision = 1
+	}
+	names := make([]string, 0, len(configuration.Upstreams))
+	for name := range configuration.Upstreams {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	policies := make([]activitypool.UpstreamDrainPolicy, 0, len(names))
+	for _, name := range names {
+		upstream := configuration.Upstreams[name]
+		policy := activitypool.UpstreamDrainPolicy{UpstreamID: name, Revision: revision, Enabled: upstream.Enabled}
+		if source.state != nil {
+			admin, exists := states[name]
+			if !exists {
+				return nil, ErrInvalidUpstreamDrainReaper
+			}
+			policy.Enabled = policy.Enabled && admin.Enabled
+			policy.Revision = admin.Revision
+		}
+		policies = append(policies, policy)
+	}
+	return policies, nil
+}
+
+type UpstreamDrainPolicySource interface {
+	ReadUpstreamDrainPolicies(context.Context) ([]activitypool.UpstreamDrainPolicy, error)
+}
+
+// UpstreamDrainStore is the narrow activity-pool capability required by the
+// configuration-driven Drain reaper.
+type UpstreamDrainStore interface {
+	activitypool.UpstreamDrainPolicyWriter
+	activitypool.DisabledUpstreamDrainLister
+	activitypool.DisabledUpstreamDrainStarter
+}
+
+type UpstreamDrainReaperOptions struct {
+	PollInterval time.Duration
+	BatchSize    int
+	Now          func() time.Time
+}
+
+type UpstreamDrainResult struct {
+	Candidates int
+	Started    int
+}
+
+// ConfiguredUpstreamDrainReaper continually publishes an effective, complete
+// policy view before reading one bounded batch of disabled Upstream ownership.
+// The Redis ownership operation performs the final policy-revision fence.
+type ConfiguredUpstreamDrainReaper struct {
+	policies UpstreamDrainPolicySource
+	store    UpstreamDrainStore
+	options  UpstreamDrainReaperOptions
+}
+
+func NewConfiguredUpstreamDrainReaper(
+	policies UpstreamDrainPolicySource,
+	store UpstreamDrainStore,
+	options UpstreamDrainReaperOptions,
+) (*ConfiguredUpstreamDrainReaper, error) {
+	if nilInterface(policies) || nilInterface(store) || options.PollInterval <= 0 || options.BatchSize <= 0 || options.Now == nil {
+		return nil, ErrInvalidUpstreamDrainReaper
+	}
+	return &ConfiguredUpstreamDrainReaper{policies: policies, store: store, options: options}, nil
+}
+
+func (reaper *ConfiguredUpstreamDrainReaper) Tick(ctx context.Context) (UpstreamDrainResult, error) {
+	if ctx == nil || reaper == nil || nilInterface(reaper.policies) || nilInterface(reaper.store) ||
+		reaper.options.PollInterval <= 0 || reaper.options.BatchSize <= 0 || reaper.options.Now == nil {
+		return UpstreamDrainResult{}, ErrInvalidUpstreamDrainReaper
+	}
+	if err := ctx.Err(); err != nil {
+		return UpstreamDrainResult{}, err
+	}
+	now := reaper.options.Now()
+	if now.IsZero() {
+		return UpstreamDrainResult{}, ErrInvalidUpstreamDrainReaper
+	}
+	policies, err := reaper.policies.ReadUpstreamDrainPolicies(ctx)
+	if err != nil {
+		return UpstreamDrainResult{}, err
+	}
+	if err := reaper.store.ReplaceUpstreamDrainPolicies(ctx, policies); err != nil {
+		return UpstreamDrainResult{}, err
+	}
+	result := UpstreamDrainResult{}
+	remaining := reaper.options.BatchSize
+	utcNow := now.UTC()
+	for _, policy := range policies {
+		if policy.Enabled || remaining == 0 {
+			continue
+		}
+		candidates, err := reaper.store.ListDisabledUpstreamDrainCandidates(ctx, utcNow, policy, remaining)
+		if err != nil {
+			return result, err
+		}
+		if len(candidates) > remaining {
+			return result, ErrInvalidUpstreamDrainReaper
+		}
+		for _, candidate := range candidates {
+			started, err := reaper.store.BeginDisabledUpstreamDrain(ctx, utcNow, candidate)
+			if err != nil {
+				return result, err
+			}
+			result.Candidates++
+			if started {
+				result.Started++
+			}
+		}
+		remaining -= len(candidates)
+	}
+	return result, nil
+}
+
+func (reaper *ConfiguredUpstreamDrainReaper) Run(ctx context.Context) error {
+	if reaper == nil {
+		return ErrInvalidUpstreamDrainReaper
+	}
+	return runScheduler(ctx, reaper.options.PollInterval, func(tickCtx context.Context) (TickResult, error) {
+		_, err := reaper.Tick(tickCtx)
+		return TickResult{}, err
+	})
+}

internal/controller/health/upstream_drain_reaper_test.go

@@ -0,0 +1,114 @@
+package health
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/domain/activitypool"
+	"proxy-pool/internal/domain/adminstate"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+type upstreamDrainPolicySourceStub struct {
+	policies []activitypool.UpstreamDrainPolicy
+	err      error
+}
+
+func (source upstreamDrainPolicySourceStub) ReadUpstreamDrainPolicies(context.Context) ([]activitypool.UpstreamDrainPolicy, error) {
+	return append([]activitypool.UpstreamDrainPolicy(nil), source.policies...), source.err
+}
+
+type upstreamDrainConfigurationStub struct {
+	configuration *config.Config
+	revision      uint64
+}
+
+type upstreamDrainStateStub struct{ snapshot adminstate.Snapshot }
+
+func (source upstreamDrainStateStub) Snapshot(context.Context) (adminstate.Snapshot, error) {
+	return source.snapshot, nil
+}
+
+func (source upstreamDrainConfigurationStub) Snapshot() (*config.Config, uint64) {
+	return source.configuration, source.revision
+}
+
+func TestConfiguredUpstreamDrainReaperStartsBoundedDisabledAssignments(t *testing.T) {
+	now := time.Date(2026, 8, 2, 16, 0, 0, 0, time.UTC)
+	pool := activitypool.NewMemoryPool()
+	if _, err := pool.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
+		ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
+		Proxies: []proxyDomain.Proxy{{
+			ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, State: proxyDomain.StateAvailable,
+		}},
+	}); err != nil {
+		t.Fatalf("UpsertFetched() error = %v", err)
+	}
+	assignment, err := pool.Assign(context.Background(), now, "proxy-a", "worker-a", time.Minute)
+	if err != nil {
+		t.Fatalf("Assign() error = %v", err)
+	}
+	policy := activitypool.UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 21}
+	reaper, err := NewConfiguredUpstreamDrainReaper(upstreamDrainPolicySourceStub{policies: []activitypool.UpstreamDrainPolicy{policy}}, pool,
+		UpstreamDrainReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }})
+	if err != nil {
+		t.Fatalf("NewConfiguredUpstreamDrainReaper() error = %v", err)
+	}
+	result, err := reaper.Tick(context.Background())
+	if err != nil || result != (UpstreamDrainResult{Candidates: 1, Started: 1}) {
+		t.Fatalf("Tick() = %+v, %v", result, err)
+	}
+	if current, found, err := pool.Get(context.Background(), "proxy-a"); err != nil || !found ||
+		current.Epoch != assignment.Epoch || !current.Draining {
+		t.Fatalf("Get(after drain) = %+v, %t, %v", current, found, err)
+	}
+}
+
+func TestEffectiveUpstreamDrainPolicySourceUsesStableConfigOnlyRevision(t *testing.T) {
+	configuration := &config.Config{Upstreams: map[string]config.Upstream{
+		"provider-b": {Enabled: true},
+		"provider-a": {Enabled: false},
+	}}
+	source, err := NewEffectiveUpstreamDrainPolicySource(upstreamDrainConfigurationStub{configuration: configuration})
+	if err != nil {
+		t.Fatalf("NewEffectiveUpstreamDrainPolicySource() error = %v", err)
+	}
+	policies, err := source.ReadUpstreamDrainPolicies(context.Background())
+	if err != nil || len(policies) != 2 || policies[0] != (activitypool.UpstreamDrainPolicy{
+		UpstreamID: "provider-a", Revision: 1,
+	}) || policies[1] != (activitypool.UpstreamDrainPolicy{
+		UpstreamID: "provider-b", Revision: 1, Enabled: true,
+	}) {
+		t.Fatalf("ReadUpstreamDrainPolicies() = %+v, %v", policies, err)
+	}
+}
+
+func TestEffectiveUpstreamDrainPolicySourceCombinesConfigurationAndAdminState(t *testing.T) {
+	configuration := &config.Config{Upstreams: map[string]config.Upstream{
+		"provider-a": {Enabled: false},
+		"provider-b": {Enabled: true},
+	}}
+	source, err := NewEffectiveUpstreamDrainPolicySource(
+		upstreamDrainConfigurationStub{configuration: configuration, revision: 9},
+		upstreamDrainStateStub{snapshot: adminstate.Snapshot{
+			Config: &adminstate.ConfigRevision{Revision: 9},
+			Upstreams: []adminstate.UpstreamState{
+				{Name: "provider-a", Enabled: true, Revision: 7},
+				{Name: "provider-b", Enabled: false, Revision: 8},
+			},
+		}},
+	)
+	if err != nil {
+		t.Fatalf("NewEffectiveUpstreamDrainPolicySource() error = %v", err)
+	}
+	policies, err := source.ReadUpstreamDrainPolicies(context.Background())
+	want := []activitypool.UpstreamDrainPolicy{
+		{UpstreamID: "provider-a", Revision: 7},
+		{UpstreamID: "provider-b", Revision: 8},
+	}
+	if err != nil || len(policies) != len(want) || policies[0] != want[0] || policies[1] != want[1] {
+		t.Fatalf("ReadUpstreamDrainPolicies() = %+v, %v; want %+v", policies, err, want)
+	}
+}

internal/domain/activitypool/pool.go

@@ -162,6 +162,45 @@ type UnhealthyDrainStarter interface {
 	BeginUnhealthyDrain(context.Context, time.Time, UnhealthyDrainCandidate) (bool, error)
 }
 
+// UpstreamDrainPolicy is the effective, versioned enablement state for one
+// Upstream. It is copied into the activity pool so a conditional Drain can
+// verify the same policy revision that selected its candidate.
+type UpstreamDrainPolicy struct {
+	UpstreamID string
+	Revision   uint64
+	Enabled    bool
+}
+
+// UpstreamDrainPolicyWriter replaces the complete effective policy view. A
+// complete replacement removes policies for deleted Upstreams as well.
+type UpstreamDrainPolicyWriter interface {
+	ReplaceUpstreamDrainPolicies(context.Context, []UpstreamDrainPolicy) error
+}
+
+// DisabledUpstreamDrainCandidate is a bounded, non-authoritative ownership
+// hint for an effectively disabled Upstream. PolicyRevision fences an old
+// candidate after the Upstream has been enabled again.
+type DisabledUpstreamDrainCandidate struct {
+	UpstreamID      string
+	PolicyRevision  uint64
+	ProxyID         string
+	WorkerID        string
+	AssignmentEpoch uint64
+}
+
+// DisabledUpstreamDrainLister finds candidates from one disabled policy. The
+// list is deliberately bounded and must be followed by BeginDisabledUpstreamDrain.
+type DisabledUpstreamDrainLister interface {
+	ListDisabledUpstreamDrainCandidates(context.Context, time.Time, UpstreamDrainPolicy, int) ([]DisabledUpstreamDrainCandidate, error)
+}
+
+// DisabledUpstreamDrainStarter conditionally begins a Drain only while the
+// candidate's effective policy revision, ownership and Proxy source still
+// agree. False means a normal concurrent change made the candidate stale.
+type DisabledUpstreamDrainStarter interface {
+	BeginDisabledUpstreamDrain(context.Context, time.Time, DisabledUpstreamDrainCandidate) (bool, error)
+}
+
 const maximumUnhealthySweepScan = 1024
 
 type Entry struct {
@@ -182,6 +221,7 @@ type MemoryPool struct {
 	drains     map[string]ownershipDomain.DrainTicket
 	targets    map[targetHealthKey]healthDomain.TargetState
 	unhealthy  map[string]time.Time
+	policies   map[string]UpstreamDrainPolicy
 	nextEpoch  uint64
 }
 
@@ -207,6 +247,9 @@ var (
 	_ Maintainer                       = (*MemoryPool)(nil)
 	_ UnhealthyRemover                 = (*MemoryPool)(nil)
 	_ UnhealthyDrainStarter            = (*MemoryPool)(nil)
+	_ UpstreamDrainPolicyWriter        = (*MemoryPool)(nil)
+	_ DisabledUpstreamDrainLister      = (*MemoryPool)(nil)
+	_ DisabledUpstreamDrainStarter     = (*MemoryPool)(nil)
 	_ extractionDomain.Store           = (*MemoryPool)(nil)
 	_ ownershipDomain.Repository       = (*MemoryPool)(nil)
 	_ ownershipDomain.DrainTicketStore = (*MemoryPool)(nil)
@@ -221,6 +264,7 @@ func NewMemoryPool() *MemoryPool {
 		drains:     make(map[string]ownershipDomain.DrainTicket),
 		targets:    make(map[targetHealthKey]healthDomain.TargetState),
 		unhealthy:  make(map[string]time.Time),
+		policies:   make(map[string]UpstreamDrainPolicy),
 	}
 }
 
@@ -708,6 +752,117 @@ func (p *MemoryPool) SweepUnhealthy(
 	return result, nil
 }
 
+// ReplaceUpstreamDrainPolicies publishes one detached, versioned policy view.
+// The pool deliberately owns no configuration data beyond this narrow Drain
+// fence, so proxy records remain entirely ephemeral.
+func (p *MemoryPool) ReplaceUpstreamDrainPolicies(ctx context.Context, policies []UpstreamDrainPolicy) error {
+	if ctx == nil {
+		return ErrInvalidMaintenance
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if p == nil || !validUpstreamDrainPolicies(policies) {
+		return ErrInvalidMaintenance
+	}
+	next := make(map[string]UpstreamDrainPolicy, len(policies))
+	for _, policy := range policies {
+		next[policy.UpstreamID] = policy
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	p.policies = next
+	return nil
+}
+
+// ListDisabledUpstreamDrainCandidates returns at most limit owned AVAILABLE
+// proxies for a policy that is still disabled at the supplied revision.
+func (p *MemoryPool) ListDisabledUpstreamDrainCandidates(
+	ctx context.Context,
+	now time.Time,
+	policy UpstreamDrainPolicy,
+	limit int,
+) ([]DisabledUpstreamDrainCandidate, error) {
+	if ctx == nil {
+		return nil, ErrInvalidMaintenance
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if p == nil || now.IsZero() || limit <= 0 || !validUpstreamDrainPolicy(policy) || policy.Enabled {
+		return nil, ErrInvalidMaintenance
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	p.purgeExpiredLocked(now)
+	if current, exists := p.policies[policy.UpstreamID]; !exists || current != policy {
+		return nil, nil
+	}
+	proxyIDs := make([]string, 0)
+	for proxyID, assignment := range p.ownership {
+		entry, exists := p.entryByIDLocked(proxyID)
+		if !exists || assignment.Draining || !assignment.ExpiresAt.After(now) ||
+			entry.Proxy.SourceUpstream != policy.UpstreamID || entry.State != proxyDomain.StateAvailable ||
+			entry.OwnerWorkerID != assignment.WorkerID || !entry.UsableUntil.After(now) {
+			continue
+		}
+		proxyIDs = append(proxyIDs, proxyID)
+	}
+	sort.Strings(proxyIDs)
+	if len(proxyIDs) > limit {
+		proxyIDs = proxyIDs[:limit]
+	}
+	result := make([]DisabledUpstreamDrainCandidate, 0, len(proxyIDs))
+	for _, proxyID := range proxyIDs {
+		assignment := p.ownership[proxyID]
+		result = append(result, DisabledUpstreamDrainCandidate{
+			UpstreamID: policy.UpstreamID, PolicyRevision: policy.Revision, ProxyID: proxyID,
+			WorkerID: assignment.WorkerID, AssignmentEpoch: assignment.Epoch,
+		})
+	}
+	return result, nil
+}
+
+// BeginDisabledUpstreamDrain is the ownership-side final decision for a
+// candidate. Policy, source upstream and assignment are checked under one
+// pool lock before the existing Drain ticket flow is entered.
+func (p *MemoryPool) BeginDisabledUpstreamDrain(
+	ctx context.Context,
+	now time.Time,
+	candidate DisabledUpstreamDrainCandidate,
+) (bool, error) {
+	if err := ownershipContextError(ctx); err != nil {
+		return false, err
+	}
+	if p == nil || now.IsZero() || !validDisabledUpstreamDrainCandidate(candidate) {
+		return false, ErrInvalidMaintenance
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return false, err
+	}
+	p.purgeExpiredLocked(now)
+	policy, active := p.policies[candidate.UpstreamID]
+	assignment, owned := p.ownership[candidate.ProxyID]
+	entry, exists := p.entryByIDLocked(candidate.ProxyID)
+	if !active || policy.Enabled || policy.Revision != candidate.PolicyRevision ||
+		!owned || !exists || assignment.Draining || !assignment.ExpiresAt.After(now) ||
+		assignment.WorkerID != candidate.WorkerID || assignment.Epoch != candidate.AssignmentEpoch ||
+		entry.Proxy.SourceUpstream != candidate.UpstreamID || entry.State != proxyDomain.StateAvailable ||
+		entry.OwnerWorkerID != candidate.WorkerID {
+		return false, nil
+	}
+	_, started := p.beginDrainLocked(assignment)
+	return started, nil
+}
+
 func (p *MemoryPool) Snapshot(now time.Time) []Entry {
 	if p == nil {
 		return nil
@@ -1192,6 +1347,29 @@ func validUnhealthySweepPolicies(policies map[string]time.Duration) bool {
 	return true
 }
 
+func validUpstreamDrainPolicies(policies []UpstreamDrainPolicy) bool {
+	seen := make(map[string]struct{}, len(policies))
+	for _, policy := range policies {
+		if !validUpstreamDrainPolicy(policy) {
+			return false
+		}
+		if _, duplicate := seen[policy.UpstreamID]; duplicate {
+			return false
+		}
+		seen[policy.UpstreamID] = struct{}{}
+	}
+	return true
+}
+
+func validUpstreamDrainPolicy(policy UpstreamDrainPolicy) bool {
+	return policy.UpstreamID != "" && policy.Revision > 0
+}
+
+func validDisabledUpstreamDrainCandidate(candidate DisabledUpstreamDrainCandidate) bool {
+	return candidate.UpstreamID != "" && candidate.PolicyRevision > 0 && candidate.ProxyID != "" &&
+		candidate.WorkerID != "" && candidate.AssignmentEpoch > 0
+}
+
 func (p *MemoryPool) entryByIDLocked(proxyID string) (Entry, bool) {
 	key, ok := p.keyByID[proxyID]
 	if !ok {

internal/domain/activitypool/pool_test.go

@@ -456,3 +456,52 @@ func poolWithOneProxy(t *testing.T, now time.Time) *MemoryPool {
 	}
 	return pool
 }
+
+func TestMemoryPoolDisabledUpstreamDrainFencesPolicyRevision(t *testing.T) {
+	now := time.Date(2026, 8, 2, 15, 0, 0, 0, time.UTC)
+	pool := poolWithOneProxy(t, now)
+	proxyID := pool.Snapshot(now)[0].Proxy.ID
+	assignment, err := pool.Assign(context.Background(), now, proxyID, "worker-a", time.Minute)
+	if err != nil {
+		t.Fatalf("Assign() error = %v", err)
+	}
+	disabled := UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 11}
+	if err := pool.ReplaceUpstreamDrainPolicies(context.Background(), []UpstreamDrainPolicy{disabled}); err != nil {
+		t.Fatalf("ReplaceUpstreamDrainPolicies(disabled) error = %v", err)
+	}
+	candidates, err := pool.ListDisabledUpstreamDrainCandidates(context.Background(), now, disabled, 1)
+	if err != nil || len(candidates) != 1 || candidates[0] != (DisabledUpstreamDrainCandidate{
+		UpstreamID: "provider-a", PolicyRevision: 11, ProxyID: proxyID, WorkerID: "worker-a", AssignmentEpoch: assignment.Epoch,
+	}) {
+		t.Fatalf("ListDisabledUpstreamDrainCandidates() = %+v, %v", candidates, err)
+	}
+	if err := pool.ReplaceUpstreamDrainPolicies(context.Background(), []UpstreamDrainPolicy{{
+		UpstreamID: "provider-a", Revision: 12, Enabled: true,
+	}}); err != nil {
+		t.Fatalf("ReplaceUpstreamDrainPolicies(enabled) error = %v", err)
+	}
+	if started, err := pool.BeginDisabledUpstreamDrain(context.Background(), now, candidates[0]); err != nil || started {
+		t.Fatalf("BeginDisabledUpstreamDrain(stale policy) = %t, %v", started, err)
+	}
+	if _, found, err := pool.Get(context.Background(), proxyID); err != nil || !found {
+		t.Fatalf("Get(after stale candidate) = found:%t, %v", found, err)
+	}
+
+	if err := pool.ReplaceUpstreamDrainPolicies(context.Background(), []UpstreamDrainPolicy{{
+		UpstreamID: "provider-a", Revision: 13,
+	}}); err != nil {
+		t.Fatalf("ReplaceUpstreamDrainPolicies(disabled again) error = %v", err)
+	}
+	candidates, err = pool.ListDisabledUpstreamDrainCandidates(context.Background(), now, UpstreamDrainPolicy{
+		UpstreamID: "provider-a", Revision: 13,
+	}, 1)
+	if err != nil || len(candidates) != 1 {
+		t.Fatalf("ListDisabledUpstreamDrainCandidates(disabled again) = %+v, %v", candidates, err)
+	}
+	if started, err := pool.BeginDisabledUpstreamDrain(context.Background(), now, candidates[0]); err != nil || !started {
+		t.Fatalf("BeginDisabledUpstreamDrain() = %t, %v", started, err)
+	}
+	if assignment, found, err := pool.Get(context.Background(), proxyID); err != nil || !found || !assignment.Draining {
+		t.Fatalf("Get(after drain) = %+v, found:%t, %v", assignment, found, err)
+	}
+}

progress.md

@@ -2,6 +2,13 @@
 
 ## 2026-08-02
 
+- 已实现上游停用的自动 Drain：有效策略由静态配置与 PostgreSQL Admin 管理态共同计算，
+  Controller 持续发布到 Redis；每个候选绑定上游 revision，只从 `owned(upstream)` 有界索引
+  读取。Redis 在创建 Drain Ticket 前原子复核策略 revision、Proxy 上游、Worker、assignment
+  epoch 和租约，重启用或重新分配后的旧候选不会排空当前归属。
+- 新增 Memory/Redis 契约与 Controller Reaper 测试，并修复 Provider Supervisor 只看管理态
+  而可能重新启动静态停用上游的问题。`go test -count=1 -timeout 60s ./...` 和 Redis 集成标签
+  测试已通过。
 - 已提交并推送 `2af504f feat: expose gateway outcome metrics`：Gateway 请求路径使用
   领域 Metrics Observer，将固定阶段的成功/失败聚合到
   `proxy_pool_gateway_outcomes_total{stage,result}`，并暴露本地 Outcome 队列满时的

task_plan.md

@@ -43,7 +43,8 @@
 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照
     刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引、ownership epoch 推进和
     完整排除 Snapshot 屏障绑定；已在 ACK 与 Runtime 零计数之间完成 Redis 原子自动收尾。
-    后续将配置停用接入同一 `BeginDrain` 编排，并补齐可观测闭环。
+    配置与管理态共同决定的上游停用已接入同一 `BeginDrain` 编排：候选绑定上游 revision，
+    Redis 原子复核策略、归属与 assignment epoch。后续补齐可观测闭环。
 
 ## 串并行关系