feat: bind drain tickets to full snapshots

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

@@ -194,7 +194,8 @@ Gateway 的 `Capacity` 使用一次打包原子读取取得同一时刻的 Activ
 `snapshot.Store` 周期生成完整稀疏报告。当前 Snapshot 已移除但仍有活动连接的
 Proxy 继续以 `draining=true` 上报，直到 Active/Reserved 同时归零。
 
-Drain Ticket 目前只记录“必须由不早于所需 epoch 的完整快照排除”的持久化前置条件。
+Drain Ticket 会绑定“已签发给当前 session、且不早于所需 epoch 的完整排除快照”引用，
+包括 version、epoch 和 checksum。
 后续会在 Runtime 替换 Lua 事务中同时验证 Snapshot ACK、Ticket 屏障和零计数，避免
 拆分为读 Runtime 再释放所有权产生竞态。
 

docs/api/control-plane.md

@@ -138,8 +138,9 @@ Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且
 首次 `BeginDrain` 会原子创建按 `proxy_id + worker_id + assignment_epoch` 栅栏的待绑定
 Drain Ticket，并推进全局 ownership epoch，促使下一份完整 Snapshot 撤销该 Proxy。Ticket
 按 Worker 有界读取，供 Controller 在确认完整 Snapshot 确实不含该 Proxy 后绑定快照屏障。
-当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放所有权；
+Worker Handler 先登记该 Snapshot 引用，再将 Ticket 绑定到 session、version、epoch 与
-`AcknowledgeDrain` 仍是已有的显式完成原语。
+checksum。当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放
+所有权；`AcknowledgeDrain` 仍是已有的显式完成原语。
 
 Worker 崩溃时必须等待所有权 epoch/有效期失效后再转移，避免双主。Proto 中
 `ReportRuntimeResponse.revoke_proxy_ids` 是加速 Drain 的控制信号，不绕过

findings.md

@@ -131,7 +131,8 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
 
 - 原有 `BeginDrain` 只会把 Proxy 从 Worker 下发索引移除，无法让后续步骤区分“已发起
   撤销”与“Gateway 已收到排除该 Proxy 的完整快照”。现已增加按 Worker 有界读取的持久化
-  Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。
+  Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。Handler 在
+  Snapshot 引用登记成功后才绑定屏障，绑定内容包含 session、version、epoch 与 checksum。
 - Ticket 不能单独成为释放依据。最终清理必须把 Ticket 屏障、当前 session ACK 和完整
   Runtime 替换中的零计数置于同一 Redis Lua 原子边界；`ReadRuntime` 后再调用
   `AcknowledgeDrain` 会保留竞争窗口。

internal/adapters/redisactivity/codec.go

@@ -64,6 +64,10 @@ type drainTicketRecord struct {
 	WorkerIndexKey         string `json:"workerIndexKey"`
 	AssignmentEpoch        uint64 `json:"assignmentEpoch"`
 	RequiredSnapshotEpoch  uint64 `json:"requiredSnapshotEpoch"`
+	SessionID              string `json:"sessionId,omitempty"`
+	SnapshotVersion        uint64 `json:"snapshotVersion,omitempty"`
+	SnapshotOwnershipEpoch uint64 `json:"snapshotOwnershipEpoch,omitempty"`
+	SnapshotChecksum       string `json:"snapshotChecksum,omitempty"`
 }
 
 type idempotencyRecord struct {
@@ -256,6 +260,14 @@ func validateDrainTicketRecord(record drainTicketRecord) error {
 		!strings.Contains(record.WorkerIndexKey, "{activity}") {
 		return ErrInvalidRecord
 	}
+	bound := record.SnapshotVersion != 0 || record.SnapshotOwnershipEpoch != 0 || record.SessionID != "" || record.SnapshotChecksum != ""
+	if !bound {
+		return nil
+	}
+	if record.SessionID == "" || record.SnapshotVersion == 0 || record.SnapshotOwnershipEpoch < record.RequiredSnapshotEpoch ||
+		!validDigest(record.SnapshotChecksum) {
+		return ErrInvalidRecord
+	}
 	return nil
 }
 

internal/adapters/redisactivity/ownership.go

@@ -2,6 +2,8 @@ package redisactivity
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"time"
@@ -139,14 +141,57 @@ func (a *Adapter) PendingDrains(ctx context.Context, workerID string, limit int)
 		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{
+		ticket := ownershipDomain.DrainTicket{
 			ProxyID: record.ProxyID, WorkerID: record.WorkerID, AssignmentEpoch: record.AssignmentEpoch,
 			RequiredSnapshotEpoch: record.RequiredSnapshotEpoch,
-		})
+		}
+		if record.SnapshotVersion != 0 {
+			checksum, err := hex.DecodeString(record.SnapshotChecksum)
+			if err != nil || len(checksum) != sha256.Size {
+				return nil, invalidScriptReply("pending drain ticket barrier checksum is invalid")
+			}
+			copy(ticket.Barrier.Checksum[:], checksum)
+			ticket.Barrier = ownershipDomain.SnapshotBarrier{
+				SessionID: record.SessionID, Version: record.SnapshotVersion,
+				OwnershipEpoch: record.SnapshotOwnershipEpoch, Checksum: ticket.Barrier.Checksum,
+			}
+		}
+		tickets = append(tickets, ticket)
 	}
 	return tickets, nil
 }
 
+func (a *Adapter) BindDrainBarrier(ctx context.Context, ticket ownershipDomain.DrainTicket) error {
+	if err := validateOwnershipCall(ctx, a); err != nil {
+		return err
+	}
+	if !validDrainTicket(ticket) || !validDrainBarrier(ticket.Barrier) ||
+		ticket.Barrier.OwnershipEpoch < ticket.RequiredSnapshotEpoch {
+		return ownershipDomain.ErrInvalidDrainTicket
+	}
+	result, err := runScript(ctx, a.client, bindDrainTicketScript, []string{
+		a.keys.drainTickets, a.keys.owners, a.keys.workerDraining(ticket.WorkerID),
+	}, ticket.ProxyID, ticket.WorkerID, ticket.AssignmentEpoch, ticket.RequiredSnapshotEpoch,
+		ticket.Barrier.SessionID, ticket.Barrier.Version, ticket.Barrier.OwnershipEpoch, hex.EncodeToString(ticket.Barrier.Checksum[:]))
+	if err != nil {
+		return err
+	}
+	var reply ownershipScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return err
+	}
+	switch reply.Status {
+	case scriptOK:
+		return nil
+	case scriptStale, scriptNotFound:
+		return ownershipDomain.ErrStaleAssignment
+	case scriptInvalid:
+		return ownershipDomain.ErrInvalidDrainTicket
+	default:
+		return invalidScriptReply("unexpected drain ticket barrier reply")
+	}
+}
+
 func (a *Adapter) AcknowledgeDrain(
 	ctx context.Context,
 	proxyID string,
@@ -308,3 +353,13 @@ func validateOwnershipCall(ctx context.Context, adapter *Adapter) error {
 	}
 	return nil
 }
+
+func validDrainTicket(ticket ownershipDomain.DrainTicket) bool {
+	return ticket.ProxyID != "" && ticket.WorkerID != "" && ticket.AssignmentEpoch > 0 &&
+		ticket.RequiredSnapshotEpoch > ticket.AssignmentEpoch
+}
+
+func validDrainBarrier(barrier ownershipDomain.SnapshotBarrier) bool {
+	return barrier.SessionID != "" && barrier.Version > 0 && barrier.OwnershipEpoch > 0 &&
+		barrier.Checksum != [32]byte{}
+}

internal/adapters/redisactivity/ownership_integration_test.go

@@ -72,6 +72,16 @@ func TestRedisOwnershipLifecycle(t *testing.T) {
 	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)
 	}
+	pending[0].Barrier = ownershipDomain.SnapshotBarrier{
+		SessionID: "session-a", Version: 1, OwnershipEpoch: pending[0].RequiredSnapshotEpoch,
+		Checksum: [32]byte{1},
+	}
+	if err := fixture.Adapter.BindDrainBarrier(context.Background(), pending[0]); err != nil {
+		t.Fatalf("BindDrainBarrier(): %v", err)
+	}
+	if bound, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1); err != nil || len(bound) != 1 || bound[0].Barrier != pending[0].Barrier {
+		t.Fatalf("PendingDrains(bound) = %+v, %v", bound, 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)

internal/adapters/redisactivity/scripts.go

@@ -161,6 +161,9 @@ var ownershipSource string
 //go:embed scripts/drain_tickets.lua
 var drainTicketsSource string
 
+//go:embed scripts/bind_drain_ticket.lua
+var bindDrainTicketSource string
+
 //go:embed scripts/sweep.lua
 var sweepSource string
 
@@ -185,6 +188,7 @@ var (
 	extractScript         = redis.NewScript(extractSource)
 	ownershipScript       = redis.NewScript(ownershipSource)
 	drainTicketsScript    = redis.NewScript(drainTicketsSource)
+	bindDrainTicketScript = redis.NewScript(bindDrainTicketSource)
 	sweepScript           = redis.NewScript(sweepSource)
 	statusScript          = redis.NewScript(statusSource)
 	runtimeScript         = redis.NewScript(runtimeSource)

internal/adapters/redisactivity/scripts/bind_drain_ticket.lua

@@ -0,0 +1,65 @@
+local tickets_key = KEYS[1]
+local owners_key = KEYS[2]
+local worker_index_key = KEYS[3]
+
+local proxy_id = ARGV[1]
+local worker_id = ARGV[2]
+local assignment_epoch = tonumber(ARGV[3])
+local required_epoch = tonumber(ARGV[4])
+local session_id = ARGV[5]
+local version = tonumber(ARGV[6])
+local snapshot_epoch = tonumber(ARGV[7])
+local checksum = ARGV[8]
+
+local function reply(status)
+    return cjson.encode({status = status})
+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_checksum(value)
+    return type(value) == 'string' and string.len(value) == 64 and string.match(value, '^[0-9a-f]+$') ~= nil
+end
+
+if type(proxy_id) ~= 'string' or proxy_id == '' or type(worker_id) ~= 'string' or worker_id == '' or
+    not assignment_epoch or assignment_epoch <= 0 or not required_epoch or required_epoch <= assignment_epoch or
+    type(session_id) ~= 'string' or session_id == '' or not version or version <= 0 or
+    not snapshot_epoch or snapshot_epoch < required_epoch or not valid_checksum(checksum) then
+    return reply('invalid')
+end
+
+local ticket = decode_table(redis.call('HGET', tickets_key, proxy_id))
+local assignment = decode_table(redis.call('HGET', owners_key, proxy_id))
+if not ticket or ticket.version ~= 1 or ticket.proxyId ~= proxy_id or ticket.workerId ~= worker_id or
+    ticket.workerIndexKey ~= worker_index_key or tonumber(ticket.assignmentEpoch) ~= assignment_epoch or
+    tonumber(ticket.requiredSnapshotEpoch) ~= required_epoch or not assignment or assignment.version ~= 1 or
+    assignment.draining ~= true or assignment.workerId ~= worker_id or tonumber(assignment.epoch) ~= assignment_epoch then
+    return reply('stale')
+end
+
+local current_epoch = tonumber(ticket.snapshotOwnershipEpoch) or 0
+local current_version = tonumber(ticket.snapshotVersion) or 0
+if current_epoch > snapshot_epoch or (current_epoch == snapshot_epoch and current_version > version) then
+    return reply('ok')
+end
+if current_epoch == snapshot_epoch and current_version == version then
+    if ticket.sessionId ~= session_id or ticket.snapshotChecksum ~= checksum then
+        return reply('stale')
+    end
+    return reply('ok')
+end
+ticket.sessionId = session_id
+ticket.snapshotVersion = version
+ticket.snapshotOwnershipEpoch = snapshot_epoch
+ticket.snapshotChecksum = checksum
+redis.call('HSET', tickets_key, proxy_id, cjson.encode(ticket))
+return reply('ok')

internal/controller/bootstrap/bootstrap.go

@@ -269,6 +269,10 @@ func runWithWorkerFactory(
 		if reader, ok := opened.workerStore.(ownershipDomain.SnapshotReader); ok {
 			snapshotReader = reader
 		}
+		var drainTickets ownershipDomain.DrainTicketStore
+		if tickets, ok := opened.workerStore.(ownershipDomain.DrainTicketStore); ok {
+			drainTickets = tickets
+		}
 		service, serviceErr := worker.NewService(opened.workerStore, worker.Options{
 			ProtocolVersion:    loaded.Value.ControlPlane.ProtocolVersion,
 			HeartbeatInterval:  loaded.Value.ControlPlane.HeartbeatInterval.Value(),
@@ -277,6 +281,7 @@ func runWithWorkerFactory(
 			MaxRuntimeCounters: loaded.Value.ControlPlane.MaxRuntimeCounters,
 			MaxSnapshotBytes:   loaded.Value.ControlPlane.MaxMessageBytes,
 			SnapshotReader:     snapshotReader,
+			DrainTickets:       drainTickets,
 			RoutingSource:      routingSource,
 			Credentials:        opened.credentials,
 		})

internal/controller/worker/grpc_handler.go

@@ -99,6 +99,14 @@ func (handler *GRPCHandler) WatchSnapshots(request *controlplanev1.WatchSnapshot
 			if err := handler.issueSnapshot(stream.Context(), request.GetWorkerId(), request.GetSessionId(), snapshot); err != nil {
 				return grpcError(err)
 			}
+			if binder, ok := handler.service.(interface {
+				BindDrainBarriers(context.Context, string, string, workerruntime.SnapshotReference, []string) error
+			}); ok {
+				reference := snapshotReference(request.GetWorkerId(), snapshot)
+				if err := binder.BindDrainBarriers(stream.Context(), request.GetWorkerId(), request.GetSessionId(), reference, snapshotProxyIDs(snapshot)); err != nil {
+					return grpcError(err)
+				}
+			}
 			if err := stream.Send(&controlplanev1.SnapshotEnvelope{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: snapshot}}); err != nil {
 				return err
 			}
@@ -114,11 +122,25 @@ func (handler *GRPCHandler) issueSnapshot(ctx context.Context, workerID, session
 		!snapshot.GetValidUntil().AsTime().After(time.Now().UTC()) {
 		return ErrInvalidCommand
 	}
+	return handler.service.IssueSnapshot(ctx, sessionID, snapshotReference(workerID, snapshot))
+}
+
+func snapshotReference(workerID string, snapshot *controlplanev1.WorkerSnapshot) workerruntime.SnapshotReference {
 	var checksum [sha256.Size]byte
 	copy(checksum[:], snapshot.GetChecksum())
-	return handler.service.IssueSnapshot(ctx, sessionID, workerruntime.SnapshotReference{
+	return workerruntime.SnapshotReference{
 		WorkerID: workerID, Version: snapshot.GetVersion(), OwnershipEpoch: snapshot.GetOwnershipEpoch(), Checksum: checksum,
-	})
+	}
+}
+
+func snapshotProxyIDs(snapshot *controlplanev1.WorkerSnapshot) []string {
+	proxyIDs := make([]string, 0, len(snapshot.GetProxies()))
+	for _, proxy := range snapshot.GetProxies() {
+		if proxy != nil {
+			proxyIDs = append(proxyIDs, proxy.GetId())
+		}
+	}
+	return proxyIDs
 }
 
 func resetSnapshotExpiryTimer(timer *time.Timer, deadline time.Time) (*time.Timer, <-chan time.Time) {

internal/controller/worker/grpc_handler_test.go

@@ -171,6 +171,28 @@ func TestGRPCHandlerStreamsIssuedFullSnapshots(t *testing.T) {
 	}
 }
 
+func TestGRPCHandlerBindsDrainBarriersAfterIssuingExcludedSnapshot(t *testing.T) {
+	checksum := make([]byte, 32)
+	checksum[0] = 1
+	service := &drainBindingServiceStub{grpcServiceStub: grpcServiceStub{}}
+	client, cleanup := grpcWorkerClient(t, service, allowIdentity{}, snapshotSourceStub{snapshots: []*controlplanev1.WorkerSnapshot{{
+		Version: 3, OwnershipEpoch: 9, Checksum: checksum,
+		GeneratedAt: timestamppb.New(time.Now()), ValidUntil: timestamppb.New(time.Now().Add(time.Minute)),
+		Proxies: []*controlplanev1.OwnedProxy{{Id: "proxy-present"}},
+	}}})
+	defer cleanup()
+	stream, err := client.WatchSnapshots(context.Background(), &controlplanev1.WatchSnapshotsRequest{WorkerId: "worker-a", SessionId: "session-a"})
+	if err != nil {
+		t.Fatalf("WatchSnapshots(): %v", err)
+	}
+	if _, err := stream.Recv(); err != nil {
+		t.Fatalf("Recv(): %v", err)
+	}
+	if !service.boundAfterIssue || service.boundReference.Version != 3 || len(service.presentProxyIDs) != 1 || service.presentProxyIDs[0] != "proxy-present" {
+		t.Fatalf("BindDrainBarriers() = afterIssue:%t reference:%+v present:%v", service.boundAfterIssue, service.boundReference, service.presentProxyIDs)
+	}
+}
+
 func TestGRPCHandlerClosesSnapshotStreamAtValidityDeadline(t *testing.T) {
 	checksum := make([]byte, 32)
 	checksum[0] = 1
@@ -255,6 +277,20 @@ type grpcServiceStub struct {
 	issueErr        error
 }
 
+type drainBindingServiceStub struct {
+	grpcServiceStub
+	boundAfterIssue bool
+	boundReference  workerruntime.SnapshotReference
+	presentProxyIDs []string
+}
+
+func (stub *drainBindingServiceStub) BindDrainBarriers(_ context.Context, _ string, _ string, reference workerruntime.SnapshotReference, proxyIDs []string) error {
+	stub.boundAfterIssue = stub.issued.Version == reference.Version
+	stub.boundReference = reference
+	stub.presentProxyIDs = append([]string(nil), proxyIDs...)
+	return nil
+}
+
 func (stub *grpcServiceStub) Register(context.Context, RegisterCommand) (Registration, error) {
 	return stub.registration, stub.registerErr
 }

internal/controller/worker/service.go

@@ -60,6 +60,7 @@ type Options struct {
 	MaxRuntimeCounters int
 	MaxSnapshotBytes   int
 	SnapshotReader     ownershipDomain.SnapshotReader
+	DrainTickets       ownershipDomain.DrainTicketStore
 	RoutingSource      RoutingSource
 	Credentials        platformCredentials.Store
 	SessionID          func() (string, error)
@@ -122,6 +123,7 @@ func (service *service) IssueSnapshot(ctx context.Context, sessionID string, ref
 type service struct {
 	store     workerruntime.ControlStore
 	outcomes  workerruntime.OutcomeWriter
+	drains    ownershipDomain.DrainTicketStore
 	options   Options
 	snapshots SnapshotSource
 }
@@ -135,7 +137,7 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err
 	if options.SessionID == nil {
 		options.SessionID = randomSessionID
 	}
-	result := &service{store: store, options: options}
+	result := &service{store: store, drains: options.DrainTickets, options: options}
 	result.outcomes, _ = store.(workerruntime.OutcomeWriter)
 	if options.SnapshotReader != nil {
 		var routing []RoutingSource
@@ -165,6 +167,54 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err
 	return result, nil
 }
 
+// BindDrainBarriers records the issued full snapshot which excludes each
+// pending proxy. The handler invokes it only after the snapshot reference is
+// durably issued for the current session.
+func (service *service) BindDrainBarriers(
+	ctx context.Context,
+	workerID, sessionID string,
+	reference workerruntime.SnapshotReference,
+	presentProxyIDs []string,
+) error {
+	if service == nil || service.drains == nil {
+		return nil
+	}
+	if ctx == nil || !workerruntime.ValidIdentifier(workerID) || !workerruntime.ValidIdentifier(sessionID) {
+		return ErrInvalidCommand
+	}
+	normalized, err := workerruntime.NormalizeSnapshotReference(reference)
+	if err != nil || normalized.WorkerID != workerID {
+		return errors.Join(ErrInvalidCommand, err)
+	}
+	present := make(map[string]struct{}, len(presentProxyIDs))
+	for _, proxyID := range presentProxyIDs {
+		if !workerruntime.ValidIdentifier(proxyID) {
+			return ErrInvalidCommand
+		}
+		if _, exists := present[proxyID]; exists {
+			return ErrInvalidCommand
+		}
+		present[proxyID] = struct{}{}
+	}
+	tickets, err := service.drains.PendingDrains(ctx, workerID, service.options.MaxRuntimeCounters)
+	if err != nil {
+		return classifyStoreError(err)
+	}
+	for _, ticket := range tickets {
+		if _, exists := present[ticket.ProxyID]; exists {
+			continue
+		}
+		ticket.Barrier = ownershipDomain.SnapshotBarrier{
+			SessionID: sessionID, Version: normalized.Version, OwnershipEpoch: normalized.OwnershipEpoch,
+			Checksum: normalized.Checksum,
+		}
+		if err := service.drains.BindDrainBarrier(ctx, ticket); err != nil {
+			return classifyStoreError(err)
+		}
+	}
+	return nil
+}
+
 func (service *service) SnapshotSource() SnapshotSource {
 	if service == nil {
 		return nil

internal/domain/activitypool/contracttest/contract.go

@@ -299,6 +299,17 @@ func runOwnershipContract(t *testing.T, factory Factory) {
 		pending[0].AssignmentEpoch != assigned.Epoch || pending[0].RequiredSnapshotEpoch <= assigned.Epoch {
 		t.Fatalf("PendingDrains() = %+v, %v", pending, err)
 	}
+	pending[0].Barrier = ownershipDomain.SnapshotBarrier{
+		SessionID: "session-a", Version: 1, OwnershipEpoch: pending[0].RequiredSnapshotEpoch,
+		Checksum: [32]byte{1},
+	}
+	if err := store.BindDrainBarrier(context.Background(), pending[0]); err != nil {
+		t.Fatalf("BindDrainBarrier(): %v", err)
+	}
+	bound, err := store.PendingDrains(context.Background(), "worker-a", 1)
+	if err != nil || len(bound) != 1 || bound[0].Barrier != pending[0].Barrier {
+		t.Fatalf("PendingDrains(bound) = %+v, %v", bound, 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)
 	}

internal/domain/activitypool/pool.go

@@ -929,6 +929,35 @@ func (p *MemoryPool) PendingDrains(ctx context.Context, workerID string, limit i
 	return tickets, nil
 }
 
+func (p *MemoryPool) BindDrainBarrier(ctx context.Context, ticket ownershipDomain.DrainTicket) error {
+	if err := ownershipContextError(ctx); err != nil {
+		return err
+	}
+	if p == nil || !validDrainTicket(ticket) || !validSnapshotBarrier(ticket.Barrier) ||
+		ticket.Barrier.OwnershipEpoch < ticket.RequiredSnapshotEpoch {
+		return ownershipDomain.ErrInvalidDrainTicket
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	current, exists := p.drains[ticket.ProxyID]
+	assignment, owned := p.ownership[ticket.ProxyID]
+	if !exists || !owned || !assignment.Draining || assignment.WorkerID != ticket.WorkerID ||
+		assignment.Epoch != ticket.AssignmentEpoch || current.ProxyID != ticket.ProxyID ||
+		current.WorkerID != ticket.WorkerID || current.AssignmentEpoch != ticket.AssignmentEpoch ||
+		current.RequiredSnapshotEpoch != ticket.RequiredSnapshotEpoch {
+		return ownershipDomain.ErrStaleAssignment
+	}
+	if barrierAfter(current.Barrier, ticket.Barrier) || barrierEqual(current.Barrier, ticket.Barrier) {
+		return nil
+	}
+	current.Barrier = ticket.Barrier
+	p.drains[ticket.ProxyID] = current
+	return 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
@@ -1025,6 +1054,27 @@ func ownershipContextError(ctx context.Context) error {
 	return ctx.Err()
 }
 
+func validDrainTicket(ticket ownershipDomain.DrainTicket) bool {
+	return ticket.ProxyID != "" && ticket.WorkerID != "" && ticket.AssignmentEpoch > 0 &&
+		ticket.RequiredSnapshotEpoch > ticket.AssignmentEpoch
+}
+
+func validSnapshotBarrier(barrier ownershipDomain.SnapshotBarrier) bool {
+	return barrier.SessionID != "" && barrier.Version > 0 && barrier.OwnershipEpoch > 0 &&
+		barrier.Checksum != [sha256.Size]byte{}
+}
+
+func barrierAfter(current, next ownershipDomain.SnapshotBarrier) bool {
+	if current.OwnershipEpoch != next.OwnershipEpoch {
+		return current.OwnershipEpoch > next.OwnershipEpoch
+	}
+	return current.Version > next.Version
+}
+
+func barrierEqual(left, right ownershipDomain.SnapshotBarrier) bool {
+	return left == right
+}
+
 func (p *MemoryPool) purgeExpiredLocked(now time.Time) int {
 	removed := 0
 	for key, entry := range p.entries {

internal/domain/ownership/ownership.go

@@ -2,6 +2,7 @@ package ownership
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
 	"time"
 )
@@ -33,6 +34,14 @@ type DrainTicket struct {
 	WorkerID              string
 	AssignmentEpoch       uint64
 	RequiredSnapshotEpoch uint64
+	Barrier               SnapshotBarrier
+}
+
+type SnapshotBarrier struct {
+	SessionID      string
+	Version        uint64
+	OwnershipEpoch uint64
+	Checksum       [sha256.Size]byte
 }
 
 // DrainTicketStore exposes bounded, Worker-local pending drain tickets. It
@@ -40,6 +49,7 @@ type DrainTicket struct {
 // prove both an acknowledged exclusion snapshot and zero live counters.
 type DrainTicketStore interface {
 	PendingDrains(context.Context, string, int) ([]DrainTicket, error)
+	BindDrainBarrier(context.Context, DrainTicket) error
 }
 
 // Repository is the shared authority for ownership changes. Implementations

progress.md

@@ -26,8 +26,9 @@
   有效刷新被错误拒绝。这是后续自动 Drain 编排的必要前提。
 - Drain 首次发起现在原子推进 ownership epoch，并写入由 Proxy、Worker、原 assignment
   epoch 与所需 Snapshot epoch 组成的待绑定 Ticket；Redis 使用 `drain-tickets` 与按
-  Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。完整快照排除、
+  Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。Worker Handler
-  ACK 与 Runtime 零计数的原子自动完成仍在后续切片，尚未标记为完成。
+  在持久登记完整 Snapshot 引用后，核验代理确实不在完整视图内，再绑定 Ticket 的
+  session/version/epoch/checksum 屏障；ACK 与 Runtime 零计数的原子自动完成仍在后续切片。
 
 ## 2026-07-30
 

task_plan.md

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