feat: persist drain tickets for worker snapshots

docs/adr/005-redis-activity-pool.md

@@ -100,6 +100,8 @@ pp:{activity}:carrier:<value>     ZSET   运营商候选索引
 pp:{activity}:upstream:<value>    ZSET   供应商候选索引
 pp:{activity}:owners              HASH   proxyID -> ownership assignment
 pp:{activity}:owner-expiry        ZSET   proxyID -> ownership expiry milliseconds
+pp:{activity}:drain-tickets       HASH   proxyID -> pending drain ticket
+pp:{activity}:worker-draining:<digest> ZSET 单 Worker 待绑定 Drain Ticket
 pp:{activity}:epoch               STRING ownership 全局递增代次
 pp:{activity}:inventory           HASH   upstreamID -> 当前未提取库存
 pp:{activity}:worker-sessions     HASH   workerID -> 当前 Worker session
@@ -180,6 +182,8 @@ Assign、Renew、BeginDrain 和 AcknowledgeDrain 分别使用有界小脚本，
 - Assign 只接受 AVAILABLE、无 owner 且未到 `usableUntil` 的 Proxy。
 - Renew 必须匹配 worker、epoch，并把 lease 截断到 `usableUntil`。
 - BeginDrain 对同一 assignment 幂等。
+- 首次 BeginDrain 原子写入按 Worker 索引的待绑定 Drain Ticket，并推进全局 ownership
+  epoch；`worker-owned` 已移除该 Proxy，下一份完整快照可据此撤销新预留。
 - AcknowledgeDrain 仅在 Active 和 Reserved 都为零时释放所有权。
 - Assign 与 Extract 并发竞争同一 Proxy 时，只允许一个操作成功。
 - Expire 使用 `limit` 分批回收过期 assignment，禁止无界返回。
@@ -190,6 +194,10 @@ Gateway 的 `Capacity` 使用一次打包原子读取取得同一时刻的 Activ
 `snapshot.Store` 周期生成完整稀疏报告。当前 Snapshot 已移除但仍有活动连接的
 Proxy 继续以 `draining=true` 上报，直到 Active/Reserved 同时归零。
 
+Drain Ticket 目前只记录“必须由不早于所需 epoch 的完整快照排除”的持久化前置条件。
+后续会在 Runtime 替换 Lua 事务中同时验证 Snapshot ACK、Ticket 屏障和零计数，避免
+拆分为读 Runtime 再释放所有权产生竞态。
+
 Redis Adapter 在单个 `{activity}` 原子边界内维护 Worker session 和运行态报告：
 
 1. 新 Worker session 替换旧 session，并隔离旧实例后续写入。

docs/api/control-plane.md

@@ -102,6 +102,10 @@ Worker 自报的超前版本或 epoch 也必须拒绝。
   和停止新分配的 `usable_until`。
 - 内容 `checksum`。
 
+`version` 在同一 Worker 流内严格连续递增，即使 `ownership_epoch` 因分配、Drain
+或所有权恢复而前进也不重置。Gateway 只接受比当前 `version` 恰好大一的完整快照，
+并拒绝倒退的 epoch；因此 epoch 变化不会让仍在长连接内的刷新快照被误判为缺口。
+
 `usable_until = expires_at - allocationSafetyMargin`。Worker 必须以
 `usable_until` 作为最后可分配时刻；达到该时间后即使尚未到 `expires_at`，
 也不得再为新请求选择该 Proxy。
@@ -131,6 +135,12 @@ Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且
 3. 两个计数都归零后 Controller 清除所有权。
 4. 无所有权 Proxy 才能进入 Distribution 的 Redis 原子提取操作。
 
+首次 `BeginDrain` 会原子创建按 `proxy_id + worker_id + assignment_epoch` 栅栏的待绑定
+Drain Ticket，并推进全局 ownership epoch，促使下一份完整 Snapshot 撤销该 Proxy。Ticket
+按 Worker 有界读取，供 Controller 在确认完整 Snapshot 确实不含该 Proxy 后绑定快照屏障。
+当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放所有权；
+`AcknowledgeDrain` 仍是已有的显式完成原语。
+
 Worker 崩溃时必须等待所有权 epoch/有效期失效后再转移，避免双主。Proto 中
 `ReportRuntimeResponse.revoke_proxy_ids` 是加速 Drain 的控制信号，不绕过
 Snapshot 版本和 Redis 中的权威所有权状态。

findings.md

@@ -123,6 +123,18 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
   Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot
   消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报，但 Controller 尚未具备
   自动 `BeginDrain`/`AcknowledgeDrain` 编排或 Worker 分配器，不能视为自动 Drain 闭环。
+- Snapshot 的 `version` 是同一 Worker 流的连续序列，`ownership_epoch` 是独立且只能前进
+  的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1，
+  与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。
+
+## Drain Ticket（2026-08-02）
+
+- 原有 `BeginDrain` 只会把 Proxy 从 Worker 下发索引移除，无法让后续步骤区分“已发起
+  撤销”与“Gateway 已收到排除该 Proxy 的完整快照”。现已增加按 Worker 有界读取的持久化
+  Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。
+- Ticket 不能单独成为释放依据。最终清理必须把 Ticket 屏障、当前 session ACK 和完整
+  Runtime 替换中的零计数置于同一 Redis Lua 原子边界；`ReadRuntime` 后再调用
+  `AcknowledgeDrain` 会保留竞争窗口。
 
 ## Git 同步事实（2026-07-29）
 

internal/adapters/redisactivity/codec.go

@@ -57,6 +57,15 @@ type ownershipRecord struct {
 	Draining          bool   `json:"draining"`
 }
 
+type drainTicketRecord struct {
+	Version               int    `json:"version"`
+	ProxyID               string `json:"proxyId"`
+	WorkerID              string `json:"workerId"`
+	WorkerIndexKey        string `json:"workerIndexKey"`
+	AssignmentEpoch       uint64 `json:"assignmentEpoch"`
+	RequiredSnapshotEpoch uint64 `json:"requiredSnapshotEpoch"`
+}
+
 type idempotencyRecord struct {
 	Version       int
 	RequestDigest string
@@ -240,6 +249,16 @@ func validateOwnershipRecord(record ownershipRecord) error {
 	return nil
 }
 
+func validateDrainTicketRecord(record drainTicketRecord) error {
+	if record.Version != recordVersion || record.ProxyID == "" || record.WorkerID == "" ||
+		record.AssignmentEpoch == 0 || record.RequiredSnapshotEpoch == 0 ||
+		record.RequiredSnapshotEpoch <= record.AssignmentEpoch || record.WorkerIndexKey == "" ||
+		!strings.Contains(record.WorkerIndexKey, "{activity}") {
+		return ErrInvalidRecord
+	}
+	return nil
+}
+
 func validateIdempotencyRecord(record idempotencyRecord) error {
 	if record.Version != recordVersion {
 		return invalidRecord("unsupported idempotency version")

internal/adapters/redisactivity/keys.go

@@ -17,6 +17,7 @@ type keyspace struct {
 	available            string
 	owners               string
 	ownerExpiry          string
+	drainTickets         string
 	epoch                string
 	inventory            string
 	stateInventory       string
@@ -49,6 +50,7 @@ func newKeyspace(namespace string) keyspace {
 		available:            prefix + ":available",
 		owners:               prefix + ":owners",
 		ownerExpiry:          prefix + ":owner-expiry",
+		drainTickets:         prefix + ":drain-tickets",
 		epoch:                prefix + ":epoch",
 		inventory:            prefix + ":inventory",
 		stateInventory:       prefix + ":state-inventory",
@@ -122,6 +124,10 @@ func (keys keyspace) workerOwned(workerID string) string {
 	return keys.facet("worker-owned", workerID)
 }
 
+func (keys keyspace) workerDraining(workerID string) string {
+	return keys.facet("worker-draining", workerID)
+}
+
 func (keys keyspace) facet(name, value string) string {
 	return keys.prefix + ":" + name + ":" + digestToken(value)
 }

internal/adapters/redisactivity/ownership.go

@@ -19,6 +19,7 @@ const (
 )
 
 var _ ownershipDomain.Repository = (*Adapter)(nil)
+var _ ownershipDomain.DrainTicketStore = (*Adapter)(nil)
 
 func (a *Adapter) Assign(
 	ctx context.Context,
@@ -109,6 +110,43 @@ func (a *Adapter) BeginDrain(
 	}
 }
 
+func (a *Adapter) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) {
+	if err := validateOwnershipCall(ctx, a); err != nil {
+		return nil, err
+	}
+	if workerID == "" || limit <= 0 {
+		return nil, ownershipDomain.ErrInvalidDrainTicket
+	}
+	result, err := runScript(ctx, a.client, drainTicketsScript, []string{
+		a.keys.drainTickets, a.keys.workerDraining(workerID), a.keys.owners,
+	}, workerID, limit)
+	if err != nil {
+		return nil, err
+	}
+	var reply drainTicketsScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return nil, err
+	}
+	if reply.Status == scriptInvalid {
+		return nil, ownershipDomain.ErrInvalidDrainTicket
+	}
+	if reply.Status != scriptOK {
+		return nil, invalidScriptReply("unexpected pending drain tickets reply")
+	}
+	tickets := make([]ownershipDomain.DrainTicket, 0, len(reply.Tickets))
+	for _, raw := range reply.Tickets {
+		var record drainTicketRecord
+		if err := decodeJSON(raw, &record); err != nil || validateDrainTicketRecord(record) != nil {
+			return nil, invalidScriptReply("pending drain tickets reply contained an invalid ticket")
+		}
+		tickets = append(tickets, ownershipDomain.DrainTicket{
+			ProxyID: record.ProxyID, WorkerID: record.WorkerID, AssignmentEpoch: record.AssignmentEpoch,
+			RequiredSnapshotEpoch: record.RequiredSnapshotEpoch,
+		})
+	}
+	return tickets, nil
+}
+
 func (a *Adapter) AcknowledgeDrain(
 	ctx context.Context,
 	proxyID string,
@@ -226,8 +264,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,
 	}, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit,
-		nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID))
+		nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID), a.keys.workerDraining(workerID))
 	if err != nil {
 		return ownershipScriptReply{}, err
 	}

internal/adapters/redisactivity/ownership_integration_test.go

@@ -64,6 +64,14 @@ func TestRedisOwnershipLifecycle(t *testing.T) {
 		t.Fatalf("BeginDrain() = %+v, %v", draining, err)
 	}
 	assertRedisWorkerOwned(t, fixture, "worker-a")
+	pending, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1)
+	if err != nil || len(pending) != 1 || pending[0].ProxyID != "proxy-a" ||
+		pending[0].AssignmentEpoch != assigned.Epoch || pending[0].RequiredSnapshotEpoch <= assigned.Epoch {
+		t.Fatalf("PendingDrains() = %+v, %v", pending, err)
+	}
+	if tickets, err := fixture.Client.ZRange(context.Background(), fixture.Adapter.keys.workerDraining("worker-a"), 0, -1).Result(); err != nil || len(tickets) != 1 || tickets[0] != "proxy-a" {
+		t.Fatalf("worker-draining = %v, %v", tickets, err)
+	}
 	replayed, err := fixture.Adapter.BeginDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch)
 	if err != nil || replayed != draining {
 		t.Fatalf("BeginDrain(replay) = %+v, %v", replayed, err)
@@ -88,6 +96,9 @@ func TestRedisOwnershipLifecycle(t *testing.T) {
 	if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || ok {
 		t.Fatalf("Get(after ACK) = %+v, %t, %v", current, ok, err)
 	}
+	if pending, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1); err != nil || len(pending) != 0 {
+		t.Fatalf("PendingDrains(after ACK) = %+v, %v", pending, err)
+	}
 	restored, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
 		RequestID: "req-restored", ClientID: "client-a", Requested: 1,
 		Fulfillment: extractionDomain.Partial, Now: now.Add(40 * time.Second),

internal/adapters/redisactivity/scripts.go

@@ -87,6 +87,11 @@ type ownershipScriptReply struct {
 	Record string       `json:"record,omitempty"`
 }
 
+type drainTicketsScriptReply struct {
+	Status  scriptStatus `json:"status"`
+	Tickets []string     `json:"tickets"`
+}
+
 type maintenanceScriptReply struct {
 	Status        scriptStatus `json:"status"`
 	Count         int          `json:"count"`
@@ -153,6 +158,9 @@ var extractSource string
 //go:embed scripts/ownership.lua
 var ownershipSource string
 
+//go:embed scripts/drain_tickets.lua
+var drainTicketsSource string
+
 //go:embed scripts/sweep.lua
 var sweepSource string
 
@@ -176,6 +184,7 @@ var (
 	upstreamLookupScript = redis.NewScript(upstreamLookupSource)
 	extractScript        = redis.NewScript(extractSource)
 	ownershipScript      = redis.NewScript(ownershipSource)
+	drainTicketsScript   = redis.NewScript(drainTicketsSource)
 	sweepScript          = redis.NewScript(sweepSource)
 	statusScript         = redis.NewScript(statusSource)
 	runtimeScript        = redis.NewScript(runtimeSource)

internal/adapters/redisactivity/scripts/drain_tickets.lua

@@ -0,0 +1,55 @@
+local tickets_key = KEYS[1]
+local worker_index_key = KEYS[2]
+local owners_key = KEYS[3]
+
+local worker_id = ARGV[1]
+local limit = tonumber(ARGV[2])
+
+local function reply(status, tickets)
+    return cjson.encode({status = status, tickets = tickets or {}})
+end
+
+local function decode_table(raw)
+    if not raw then
+        return nil
+    end
+    local ok, value = pcall(cjson.decode, raw)
+    if not ok or type(value) ~= 'table' then
+        return nil
+    end
+    return value
+end
+
+local function valid_ticket(ticket, proxy_id)
+    return ticket and ticket.version == 1 and ticket.proxyId == proxy_id and
+        ticket.workerId == worker_id and type(ticket.workerIndexKey) == 'string' and
+        ticket.workerIndexKey == worker_index_key and tonumber(ticket.assignmentEpoch) and
+        tonumber(ticket.assignmentEpoch) > 0 and tonumber(ticket.requiredSnapshotEpoch) and
+        tonumber(ticket.requiredSnapshotEpoch) > tonumber(ticket.assignmentEpoch)
+end
+
+local function valid_assignment(assignment, ticket)
+    return assignment and assignment.version == 1 and assignment.draining == true and
+        assignment.proxyId == ticket.proxyId and assignment.workerId == ticket.workerId and
+        tonumber(assignment.epoch) == tonumber(ticket.assignmentEpoch)
+end
+
+if type(worker_id) ~= 'string' or worker_id == '' or not limit or limit <= 0 then
+    return reply('invalid')
+end
+
+local scan_limit = limit * 4
+local ids = redis.call('ZRANGE', worker_index_key, 0, scan_limit - 1)
+local tickets = {}
+for _, proxy_id in ipairs(ids) do
+    local raw = redis.call('HGET', tickets_key, proxy_id)
+    local ticket = decode_table(raw)
+    local assignment = decode_table(redis.call('HGET', owners_key, proxy_id))
+    if valid_ticket(ticket, proxy_id) and valid_assignment(assignment, ticket) and #tickets < limit then
+        tickets[#tickets + 1] = raw
+    else
+        redis.call('HDEL', tickets_key, proxy_id)
+        redis.call('ZREM', worker_index_key, proxy_id)
+    end
+end
+return reply('ok', tickets)

internal/adapters/redisactivity/scripts/ownership.lua

@@ -16,6 +16,7 @@ local health_tasks_key = KEYS[15]
 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 operation = ARGV[1]
 local operation_ttl_ms = tonumber(ARGV[2])
@@ -28,6 +29,7 @@ local value = tonumber(ARGV[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 mutating = operation ~= 'get'
 
 local function finish(reply)
@@ -108,6 +110,17 @@ local function remove_owned(id, record)
     end
 end
 
+local function clear_drain(id)
+    local raw = redis.call('HGET', drain_tickets_key, id)
+    if raw then
+        local decoded, ticket = pcall(cjson.decode, raw)
+        if decoded and type(ticket) == 'table' and type(ticket.workerIndexKey) == 'string' and ticket.workerIndexKey ~= '' then
+            redis.call('ZREM', ticket.workerIndexKey, id)
+        end
+    end
+    redis.call('HDEL', drain_tickets_key, id)
+end
+
 local function add_available(id, record, at_ms)
     local usable_until_ms = record and tonumber(record.usableUntilMs)
     if not usable_until_ms or record.state ~= 'AVAILABLE' or usable_until_ms <= at_ms then
@@ -176,6 +189,7 @@ local function remove_proxy(id)
     redis.call('ZREM', expiry_key, id)
     redis.call('HDEL', owners_key, id)
     redis.call('ZREM', owner_expiry_key, id)
+    clear_drain(id)
     redis.call('ZREM', health_unhealthy_key, id)
     remove_health_task(id)
 end
@@ -220,6 +234,7 @@ local function clear_owner(id, assignment, at_ms, restore)
     end
     redis.call('HDEL', owners_key, id)
     redis.call('ZREM', owner_expiry_key, id)
+    clear_drain(id)
     if assignment and type(assignment.workerIndexKey) == 'string' and assignment.workerIndexKey ~= '' then
         redis.call('ZREM', assignment.workerIndexKey, id)
     end
@@ -322,8 +337,14 @@ if operation == 'begin_drain' then
         return finish({status = 'stale'})
     end
     if not current.draining then
+		if type(worker_draining_index_key) ~= 'string' or worker_draining_index_key == '' or
+			not string.find(worker_draining_index_key, '{activity}', 1, true) then
+			return finish({status = 'invalid'})
+		end
         current.draining = true
         current.assignmentVersion = tonumber(current.assignmentVersion) + 1
+		local drain_epoch = redis.call('INCR', epoch_key)
+		redis.call('PERSIST', epoch_key)
         local encoded = cjson.encode(current)
         redis.call('HSET', owners_key, proxy_id, encoded)
         local record = decode_table(redis.call('HGET', records_key, proxy_id))
@@ -331,6 +352,18 @@ if operation == 'begin_drain' then
         if type(current.workerIndexKey) == 'string' and current.workerIndexKey ~= '' then
             redis.call('ZREM', current.workerIndexKey, proxy_id)
         end
+		local ticket = {
+			version = 1,
+			proxyId = proxy_id,
+			workerId = worker_id,
+			workerIndexKey = worker_draining_index_key,
+			assignmentEpoch = current.epoch,
+			requiredSnapshotEpoch = drain_epoch,
+		}
+		redis.call('HSET', drain_tickets_key, proxy_id, cjson.encode(ticket))
+		redis.call('ZADD', worker_draining_index_key, current.expiresAtMs, proxy_id)
+		touch(drain_tickets_key, tonumber(current.expiresAtMs))
+		touch(worker_draining_index_key, tonumber(current.expiresAtMs))
         return finish({status = 'ok', record = encoded})
     end
     return finish({status = 'ok', record = cjson.encode(current)})

internal/domain/activitypool/contracttest/contract.go

@@ -22,6 +22,7 @@ type Store interface {
 	activitypool.Maintainer
 	extractionDomain.Store
 	ownershipDomain.Repository
+	ownershipDomain.DrainTicketStore
 }
 
 type Factory func(*testing.T) (Store, func())
@@ -293,6 +294,11 @@ func runOwnershipContract(t *testing.T, factory Factory) {
 	if replayed, err := store.BeginDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch); err != nil || replayed != draining {
 		t.Fatalf("BeginDrain(replay) = %+v, %v", replayed, err)
 	}
+	pending, err := store.PendingDrains(context.Background(), "worker-a", 1)
+	if err != nil || len(pending) != 1 || pending[0].ProxyID != "proxy-a" ||
+		pending[0].AssignmentEpoch != assigned.Epoch || pending[0].RequiredSnapshotEpoch <= assigned.Epoch {
+		t.Fatalf("PendingDrains() = %+v, %v", pending, err)
+	}
 	if err := store.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch, 1, 0); !errors.Is(err, ownershipDomain.ErrDrainNotReady) {
 		t.Fatalf("AcknowledgeDrain(active) error = %v", err)
 	}
@@ -302,6 +308,9 @@ func runOwnershipContract(t *testing.T, factory Factory) {
 	if current, ok, err := store.Get(context.Background(), "proxy-a"); err != nil || ok {
 		t.Fatalf("Get(after ACK) = %+v, %t, %v", current, ok, err)
 	}
+	if pending, err := store.PendingDrains(context.Background(), "worker-a", 1); err != nil || len(pending) != 0 {
+		t.Fatalf("PendingDrains(after ACK) = %+v, %v", pending, err)
+	}
 	if result, err := store.Extract(context.Background(), extractionDomain.Command{
 		RequestID: "req-after-ack", ClientID: "client-a", Requested: 1,
 		Fulfillment: extractionDomain.Partial, Now: now.Add(20 * time.Second),
@@ -537,6 +546,7 @@ func runCancellationContract(t *testing.T, store Store) {
 		{name: "assign", call: func() error { _, err := store.Assign(ctx, now, "proxy-a", "worker-a", time.Minute); return err }},
 		{name: "renew", call: func() error { _, err := store.Renew(ctx, now, "proxy-a", "worker-a", 1, time.Minute); return err }},
 		{name: "begin drain", call: func() error { _, err := store.BeginDrain(ctx, "proxy-a", "worker-a", 1); return err }},
+		{name: "pending drains", call: func() error { _, err := store.PendingDrains(ctx, "worker-a", 1); return err }},
 		{name: "acknowledge drain", call: func() error { return store.AcknowledgeDrain(ctx, "proxy-a", "worker-a", 1, 0, 0) }},
 		{name: "get", call: func() error { _, _, err := store.Get(ctx, "proxy-a"); return err }},
 		{name: "expire", call: func() error { _, err := store.Expire(ctx, now, 1); return err }},

internal/domain/activitypool/pool.go

@@ -159,6 +159,7 @@ type MemoryPool struct {
 	keyByID    map[string]string
 	idempotent map[string]idempotencyEntry
 	ownership  map[string]ownershipDomain.Assignment
+	drains     map[string]ownershipDomain.DrainTicket
 	targets    map[targetHealthKey]healthDomain.TargetState
 	unhealthy  map[string]time.Time
 	nextEpoch  uint64
@@ -187,6 +188,7 @@ var (
 	_ UnhealthyRemover                 = (*MemoryPool)(nil)
 	_ extractionDomain.Store           = (*MemoryPool)(nil)
 	_ ownershipDomain.Repository       = (*MemoryPool)(nil)
+	_ ownershipDomain.DrainTicketStore = (*MemoryPool)(nil)
 )
 
 func NewMemoryPool() *MemoryPool {
@@ -195,6 +197,7 @@ func NewMemoryPool() *MemoryPool {
 		keyByID:    make(map[string]string),
 		idempotent: make(map[string]idempotencyEntry),
 		ownership:  make(map[string]ownershipDomain.Assignment),
+		drains:     make(map[string]ownershipDomain.DrainTicket),
 		targets:    make(map[targetHealthKey]healthDomain.TargetState),
 		unhealthy:  make(map[string]time.Time),
 	}
@@ -819,6 +822,7 @@ func (p *MemoryPool) Assign(ctx context.Context, now time.Time, proxyID, workerI
 			p.setEntryByIDLocked(proxyID, entry)
 		}
 		delete(p.ownership, proxyID)
+		delete(p.drains, proxyID)
 	}
 	entry, ok := p.entryByIDLocked(proxyID)
 	if !ok || entry.State != proxyDomain.StateAvailable || entry.OwnerWorkerID != "" || !entry.UsableUntil.After(now) {
@@ -882,10 +886,49 @@ func (p *MemoryPool) BeginDrain(ctx context.Context, proxyID, workerID string, e
 		assignment.Draining = true
 		assignment.Version++
 		p.ownership[proxyID] = assignment
+		p.nextEpoch++
+		p.drains[proxyID] = ownershipDomain.DrainTicket{
+			ProxyID: proxyID, WorkerID: workerID, AssignmentEpoch: assignment.Epoch,
+			RequiredSnapshotEpoch: p.nextEpoch,
+		}
 	}
 	return assignment, nil
 }
 
+func (p *MemoryPool) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) {
+	if err := ownershipContextError(ctx); err != nil {
+		return nil, err
+	}
+	if p == nil || workerID == "" || limit <= 0 {
+		return nil, ownershipDomain.ErrInvalidDrainTicket
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	proxyIDs := make([]string, 0)
+	for proxyID, ticket := range p.drains {
+		assignment, exists := p.ownership[proxyID]
+		if !exists || !assignment.Draining || assignment.WorkerID != ticket.WorkerID || assignment.Epoch != ticket.AssignmentEpoch {
+			delete(p.drains, proxyID)
+			continue
+		}
+		if ticket.WorkerID == workerID {
+			proxyIDs = append(proxyIDs, proxyID)
+		}
+	}
+	sort.Strings(proxyIDs)
+	if len(proxyIDs) > limit {
+		proxyIDs = proxyIDs[:limit]
+	}
+	tickets := make([]ownershipDomain.DrainTicket, 0, len(proxyIDs))
+	for _, proxyID := range proxyIDs {
+		tickets = append(tickets, p.drains[proxyID])
+	}
+	return tickets, nil
+}
+
 func (p *MemoryPool) AcknowledgeDrain(ctx context.Context, proxyID, workerID string, epoch uint64, active, reserved int64) error {
 	if err := ownershipContextError(ctx); err != nil {
 		return err
@@ -913,6 +956,7 @@ func (p *MemoryPool) AcknowledgeDrain(ctx context.Context, proxyID, workerID str
 		p.setEntryByIDLocked(proxyID, entry)
 	}
 	delete(p.ownership, proxyID)
+	delete(p.drains, proxyID)
 	return nil
 }
 
@@ -969,6 +1013,7 @@ func (p *MemoryPool) Expire(ctx context.Context, now time.Time, limit int) ([]ow
 		}
 		expired = append(expired, assignment)
 		delete(p.ownership, proxyID)
+		delete(p.drains, proxyID)
 	}
 	return expired, nil
 }
@@ -1012,6 +1057,7 @@ func (p *MemoryPool) removeEntryByIDLocked(proxyID string) {
 
 func (p *MemoryPool) removeEntryLocked(key string, entry Entry) {
 	delete(p.ownership, entry.Proxy.ID)
+	delete(p.drains, entry.Proxy.ID)
 	delete(p.unhealthy, entry.Proxy.ID)
 	for target := range p.targets {
 		if target.proxyID == entry.Proxy.ID {

internal/domain/ownership/ownership.go

@@ -13,6 +13,7 @@ var (
 	ErrStaleAssignment      = errors.New("proxy ownership assignment is stale")
 	ErrNotDraining          = errors.New("proxy ownership is not draining")
 	ErrDrainNotReady        = errors.New("proxy still has active or reserved runtime")
+	ErrInvalidDrainTicket   = errors.New("invalid drain ticket request")
 )
 
 type Assignment struct {
@@ -24,6 +25,23 @@ type Assignment struct {
 	Draining  bool
 }
 
+// DrainTicket records a pending removal after an ownership assignment enters
+// draining. RequiredSnapshotEpoch fences the first complete Worker snapshot
+// that may prove the proxy was excluded.
+type DrainTicket struct {
+	ProxyID               string
+	WorkerID              string
+	AssignmentEpoch       uint64
+	RequiredSnapshotEpoch uint64
+}
+
+// DrainTicketStore exposes bounded, Worker-local pending drain tickets. It
+// deliberately does not finalize ownership: a later runtime transaction must
+// prove both an acknowledged exclusion snapshot and zero live counters.
+type DrainTicketStore interface {
+	PendingDrains(context.Context, string, int) ([]DrainTicket, error)
+}
+
 // Repository is the shared authority for ownership changes. Implementations
 // that also support extraction must serialize both operations transactionally.
 type Repository interface {

internal/gateway/dispatch/dispatcher_test.go

@@ -354,11 +354,11 @@ func TestAcquireSurvivesConcurrentSnapshotApply(t *testing.T) {
 			{epoch: 1, version: 2},
 			{epoch: 1, version: 3},
 			{epoch: 1, version: 4},
-			{epoch: 2, version: 1},
+			{epoch: 2, version: 5},
-			{epoch: 2, version: 2},
+			{epoch: 2, version: 6},
-			{epoch: 2, version: 3},
+			{epoch: 2, version: 7},
-			{epoch: 3, version: 1},
+			{epoch: 3, version: 8},
-			{epoch: 3, version: 2},
+			{epoch: 3, version: 9},
 		}
 		for _, step := range steps {
 			next := snapshot.Envelope{

internal/gateway/snapshot/store.go

@@ -267,9 +267,7 @@ func (s *Store) Apply(envelope Envelope) error {
 		switch {
 		case envelope.Epoch < current.Epoch:
 			return ErrResyncRequired
-		case envelope.Epoch == current.Epoch && envelope.Version != current.Version+1:
+		case envelope.Version != current.Version+1:
-			return ErrResyncRequired
-		case envelope.Epoch > current.Epoch && envelope.Version != 1:
 			return ErrResyncRequired
 		}
 	}

internal/gateway/snapshot/store_test.go

@@ -78,6 +78,12 @@ func TestStoreRejectsWrongWorkerVersionGapAndChecksum(t *testing.T) {
 		t.Fatalf("version gap error = %v, want ErrResyncRequired", err)
 	}
 
+	epochReset := base
+	epochReset.Epoch = 2
+	if err := store.Apply(epochReset); !errors.Is(err, ErrResyncRequired) {
+		t.Fatalf("epoch reset version error = %v, want ErrResyncRequired", err)
+	}
+
 	badChecksum := base
 	badChecksum.Version = 2
 	badChecksum.Checksum = "bad"
@@ -272,7 +278,7 @@ func TestStoreApplyReusesCapacityAcrossVersionsAndEpochs(t *testing.T) {
 
 	third := second
 	third.Epoch = 2
-	third.Version = 1
+	third.Version = 3
 	third.Checksum = Checksum(third.Proxies)
 	if err := store.Apply(third); err != nil {
 		t.Fatalf("Apply(third): %v", err)

progress.md

@@ -21,6 +21,13 @@
   Delta 的 fail-closed 校验，也不把自动 Drain/ACK 记为已完成。
 - Health Scheduler 现在先建立 EGRESS 与 Routing TARGET 的独立 due 引用，再投递
   BASIC 任务；在单项小批次下，短 TTL Proxy 不会因 BASIC 先出队而错过首次出口/目标探测。
+- 修复 Snapshot epoch/version 栅栏：Worker 流的 `version` 在 ownership epoch 前进时
+  仍严格递增，Gateway 以连续版本加非倒退 epoch 进行完整快照原子替换，避免动态分配后的
+  有效刷新被错误拒绝。这是后续自动 Drain 编排的必要前提。
+- Drain 首次发起现在原子推进 ownership epoch，并写入由 Proxy、Worker、原 assignment
+  epoch 与所需 Snapshot epoch 组成的待绑定 Ticket；Redis 使用 `drain-tickets` 与按
+  Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。完整快照排除、
+  ACK 与 Runtime 零计数的原子自动完成仍在后续切片，尚未标记为完成。
 
 ## 2026-07-30
 

task_plan.md

@@ -41,7 +41,9 @@
     通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy；拥有
     Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。
 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照
-    刷新；后续补齐自动 Drain 编排，使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。
+    刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引和 ownership epoch 推进；
+    后续在完整快照排除、ACK 与 Runtime 零计数之间补齐原子自动完成编排，使配置停用、
+    健康淘汰与 Snapshot 撤销形成可观测闭环。
 
 ## 串并行关系