feat: finalize drains from acknowledged runtime reports

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

@@ -196,8 +196,10 @@ Proxy 继续以 `draining=true` 上报，直到 Active/Reserved 同时归零。
 
 Drain Ticket 会绑定“已签发给当前 session、且不早于所需 epoch 的完整排除快照”引用，
 包括 version、epoch 和 checksum。
-后续会在 Runtime 替换 Lua 事务中同时验证 Snapshot ACK、Ticket 屏障和零计数，避免
+Runtime 替换 Lua 事务会同时验证当前 session ACK、Ticket 屏障与该 Proxy 的零计数，
-拆分为读 Runtime 再释放所有权产生竞态。
+仅在完整稀疏报告中 Active/Reserved 都为零时释放所有权、删除 Ticket 和 Worker Drain
+索引，并在记录仍为 AVAILABLE 时恢复可分配索引。ACK 或屏障不匹配、报告中仍有计数、
+或 session 已替换时都保留 Drain；这样避免拆分为读 Runtime 再释放所有权产生竞态。
 
 Redis Adapter 在单个 `{activity}` 原子边界内维护 Worker session 和运行态报告：
 

docs/api/control-plane.md

@@ -139,8 +139,12 @@ Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且
 Drain Ticket，并推进全局 ownership epoch，促使下一份完整 Snapshot 撤销该 Proxy。Ticket
 按 Worker 有界读取，供 Controller 在确认完整 Snapshot 确实不含该 Proxy 后绑定快照屏障。
 Worker Handler 先登记该 Snapshot 引用，再将 Ticket 绑定到 session、version、epoch 与
-checksum。当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放
+checksum。后续完整 Runtime 替换在同一 Redis Lua 事务中检查：Ticket 屏障是否属于当前
-所有权；`AcknowledgeDrain` 仍是已有的显式完成原语。
+session、该 session 是否已 ACK 不早于屏障的快照、以及本次完整稀疏报告中该 Proxy 的
+Active/Reserved 是否均为零（缺失项按零）。三者同时成立才清除 owner、Ticket 与
+Worker Drain 索引，并在记录仍为 AVAILABLE 时恢复可分配索引；因此不存在先读 Runtime
+再释放所有权的竞争窗口。`AcknowledgeDrain` 保留为显式完成原语，用于不经 Runtime
+上报的受控维护流程。
 
 Worker 崩溃时必须等待所有权 epoch/有效期失效后再转移，避免双主。Proto 中
 `ReportRuntimeResponse.revoke_proxy_ids` 是加速 Drain 的控制信号，不绕过

findings.md

@@ -121,8 +121,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
   checksum。短于全局 `maxStaleAge` 的 Proxy/ownership 租约会进一步缩短下一次刷新间隔。
 - gRPC Handler 每次成功发送完整 Snapshot 都重置服务端到期计时器；流中刷新失败或中断时
   Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot
-  消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报，但 Controller 尚未具备
+  消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报。Redis Runtime 替换已在当前
-  自动 `BeginDrain`/`AcknowledgeDrain` 编排或 Worker 分配器，不能视为自动 Drain 闭环。
+  session ACK、Ticket 屏障和完整零计数同时成立时自动释放 owner；Reaper 和配置停用尚未
+  调用 `BeginDrain`，因此全链路的 Drain 触发编排仍未完成。
 - Snapshot 的 `version` 是同一 Worker 流的连续序列，`ownership_epoch` 是独立且只能前进
   的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1，
   与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。
@@ -133,9 +134,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
   撤销”与“Gateway 已收到排除该 Proxy 的完整快照”。现已增加按 Worker 有界读取的持久化
   Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。Handler 在
   Snapshot 引用登记成功后才绑定屏障，绑定内容包含 session、version、epoch 与 checksum。
-- Ticket 不能单独成为释放依据。最终清理必须把 Ticket 屏障、当前 session ACK 和完整
+- Ticket 不能单独成为释放依据。`replace_report` Lua 已把 Ticket 屏障、当前 session ACK
-  Runtime 替换中的零计数置于同一 Redis Lua 原子边界；`ReadRuntime` 后再调用
+  和完整 Runtime 替换中的零计数置于同一原子边界；只有报告缺失该 Proxy 或其
-  `AcknowledgeDrain` 会保留竞争窗口。
+  Active/Reserved 均为零时，才删除 owner/Ticket/Drain 索引并恢复可分配状态。
 
 ## Git 同步事实（2026-07-29）
 

internal/adapters/redisactivity/ownership_integration_test.go

@@ -4,6 +4,7 @@ package redisactivity
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
 	"fmt"
 	"sync"
@@ -14,6 +15,7 @@ import (
 	extractionDomain "proxy-pool/internal/domain/extraction"
 	ownershipDomain "proxy-pool/internal/domain/ownership"
 	proxyDomain "proxy-pool/internal/domain/proxy"
+	"proxy-pool/internal/domain/workerruntime"
 )
 
 func TestRedisOwnershipLifecycle(t *testing.T) {
@@ -149,6 +151,76 @@ func TestRedisOwnershipRejectsInvalidDrainAndStaleAssignment(t *testing.T) {
 	}
 }
 
+func TestRedisRuntimeFinalizesAcknowledgedBoundDrain(t *testing.T) {
+	fixture := newRedisTestFixture(t)
+	now := redisTestNow()
+	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*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(): %v", err)
+	}
+	if _, err := fixture.Adapter.BeginDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch); err != nil {
+		t.Fatalf("BeginDrain(): %v", err)
+	}
+	pending, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1)
+	if err != nil || len(pending) != 1 {
+		t.Fatalf("PendingDrains() = %+v, %v", pending, err)
+	}
+	checksum := sha256.Sum256([]byte("drain-snapshot"))
+	reference := workerruntime.SnapshotReference{
+		WorkerID: "worker-a", Version: 1, OwnershipEpoch: pending[0].RequiredSnapshotEpoch, Checksum: checksum,
+	}
+	if err := fixture.Adapter.OpenSession(context.Background(), workerruntime.Session{
+		WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a", Zone: "zone-a", ProtocolVersion: 1,
+	}, time.Minute); err != nil {
+		t.Fatalf("OpenSession(): %v", err)
+	}
+	if err := fixture.Adapter.RecordIssuedSnapshot(context.Background(), "session-a", reference, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(): %v", err)
+	}
+	pending[0].Barrier = ownershipDomain.SnapshotBarrier{
+		SessionID: "session-a", Version: reference.Version, OwnershipEpoch: reference.OwnershipEpoch, Checksum: checksum,
+	}
+	if err := fixture.Adapter.BindDrainBarrier(context.Background(), pending[0]); err != nil {
+		t.Fatalf("BindDrainBarrier(): %v", err)
+	}
+	if err := fixture.Adapter.AcknowledgeSnapshot(context.Background(), workerruntime.SnapshotAcknowledgement{
+		WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true,
+	}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(): %v", err)
+	}
+	if err := fixture.Adapter.ReplaceRuntime(context.Background(), workerruntime.Report{
+		WorkerID: "worker-a", SessionID: "session-a", Sequence: 1,
+		SnapshotVersion: reference.Version, OwnershipEpoch: reference.OwnershipEpoch, ObservedAt: now.Add(3 * time.Second),
+		Counters: []workerruntime.Counter{{ProxyID: "proxy-a", Active: 1, Draining: true}},
+	}, time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(active drain): %v", err)
+	}
+	if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || !current.Draining {
+		t.Fatalf("Get(active drain) = %+v, %t, %v", current, ok, err)
+	}
+	if err := fixture.Adapter.ReplaceRuntime(context.Background(), workerruntime.Report{
+		WorkerID: "worker-a", SessionID: "session-a", Sequence: 2,
+		SnapshotVersion: reference.Version, OwnershipEpoch: reference.OwnershipEpoch, ObservedAt: now.Add(4 * time.Second),
+	}, time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(drained): %v", err)
+	}
+	if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || ok {
+		t.Fatalf("Get(after automatic drain) = %+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 automatic drain) = %+v, %v", pending, err)
+	}
+	result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{
+		RequestID: "req-after-auto-drain", ClientID: "client-a", Requested: 1,
+		Fulfillment: extractionDomain.Partial, Now: now.Add(5 * time.Second),
+	})
+	if err != nil || result.Returned != 1 || result.Items[0].ID != "proxy-a" {
+		t.Fatalf("Extract(after automatic drain) = %+v, %v", result, err)
+	}
+}
+
 func TestRedisOwnershipExpireIsLimitedAndAllowsTakeover(t *testing.T) {
 	fixture := newRedisTestFixture(t)
 	now := redisTestNow()

internal/adapters/redisactivity/runtime.go

@@ -454,7 +454,8 @@ func (a *Adapter) runRuntime(
 		a.keys.workerSessions, a.keys.workerSessionExpiry,
 		a.keys.workerSnapshots, a.keys.workerSnapshotExpiry,
 		a.keys.workerRuntime, a.keys.workerRuntimeExpiry, a.keys.owners, a.keys.epoch,
-		a.keys.workerOutcomes,
+		a.keys.workerOutcomes, a.keys.records, a.keys.available, a.keys.ownerExpiry,
+		a.keys.drainTickets, a.keys.workerDraining(runtimeWorkerID(payload)),
 	}, operation, ttlMS, a.options.CleanupLimit, string(payload), digest)
 	if err != nil {
 		return runtimeScriptReply{}, err
@@ -466,6 +467,16 @@ func (a *Adapter) runRuntime(
 	return reply, nil
 }
 
+func runtimeWorkerID(payload []byte) string {
+	var value struct {
+		WorkerID string `json:"workerId"`
+	}
+	if json.Unmarshal(payload, &value) != nil {
+		return ""
+	}
+	return value.WorkerID
+}
+
 func validateRuntimeCall(ctx context.Context, adapter *Adapter) error {
 	if ctx == nil || adapter == nil {
 		return workerruntime.ErrInvalidStore

internal/adapters/redisactivity/scripts/runtime.lua

@@ -7,6 +7,11 @@ local runtime_expiry_key = KEYS[6]
 local owners_key = KEYS[7]
 local epoch_key = KEYS[8]
 local outcomes_key = KEYS[9]
+local records_key = KEYS[10]
+local available_key = KEYS[11]
+local owner_expiry_key = KEYS[12]
+local drain_tickets_key = KEYS[13]
+local worker_draining_key = KEYS[14]
 
 local operation = ARGV[1]
 local ttl_ms = tonumber(ARGV[2])
@@ -130,6 +135,77 @@ local function valid_owner(value, worker_id, ownership_epoch, now)
         type(value.expiresAtMs) == 'number' and value.expiresAtMs > now
 end
 
+local function touch(key, expires_at_ms)
+    if redis.call('EXISTS', key) == 0 then
+        return
+    end
+    local current = redis.call('PEXPIRETIME', key)
+    if current < expires_at_ms then
+        redis.call('PEXPIREAT', key, expires_at_ms)
+    end
+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
+        return
+    end
+    redis.call('ZADD', available_key, usable_until_ms, id)
+    touch(available_key, tonumber(record.expiresAtMs))
+    if type(record.indexKeys) == 'table' then
+        for _, index_key in ipairs(record.indexKeys) do
+            if type(index_key) == 'string' and index_key ~= '' then
+                redis.call('ZADD', index_key, usable_until_ms, id)
+                touch(index_key, tonumber(record.expiresAtMs))
+            end
+        end
+    end
+end
+
+local function valid_drain_ticket(ticket, worker_id)
+    return ticket and ticket.version == 1 and type(ticket.proxyId) == 'string' and ticket.proxyId ~= '' and
+        ticket.workerId == worker_id and type(ticket.assignmentEpoch) == 'number' and ticket.assignmentEpoch > 0 and
+        type(ticket.requiredSnapshotEpoch) == 'number' and ticket.requiredSnapshotEpoch > ticket.assignmentEpoch and
+        type(ticket.workerIndexKey) == 'string' and ticket.workerIndexKey ~= '' and
+        type(ticket.sessionId) == 'string' and ticket.sessionId ~= '' and
+        type(ticket.snapshotVersion) == 'number' and ticket.snapshotVersion > 0 and
+        type(ticket.snapshotOwnershipEpoch) == 'number' and ticket.snapshotOwnershipEpoch >= ticket.requiredSnapshotEpoch and
+        type(ticket.snapshotChecksum) == 'string' and string.len(ticket.snapshotChecksum) == 64 and
+        string.match(ticket.snapshotChecksum, '^[0-9a-f]+$') ~= nil
+end
+
+local function clear_finalized_drain(proxy_id, owner, ticket, at_ms)
+    local record = decode_table(redis.call('HGET', records_key, proxy_id))
+    if record and record.ownerWorkerId == owner.workerId then
+        record.ownerWorkerId = nil
+        redis.call('HSET', records_key, proxy_id, cjson.encode(record))
+        add_available(proxy_id, record, at_ms)
+    end
+    redis.call('HDEL', owners_key, proxy_id)
+    redis.call('ZREM', owner_expiry_key, proxy_id)
+    redis.call('HDEL', drain_tickets_key, proxy_id)
+    redis.call('ZREM', ticket.workerIndexKey, proxy_id)
+end
+
+local function ticket_acknowledged(session, ticket)
+    if session.sessionId ~= ticket.sessionId then
+        return false
+    end
+    local barrier = {
+        ownershipEpoch = tostring(ticket.snapshotOwnershipEpoch),
+        snapshotVersion = tostring(ticket.snapshotVersion),
+    }
+    local acknowledged = {
+        ownershipEpoch = session.ackedOwnershipEpoch,
+        snapshotVersion = session.ackedSnapshotVersion,
+    }
+    local ordering = compare_reference(acknowledged, barrier)
+    if ordering < 0 then
+        return false
+    end
+    return ordering ~= 0 or session.ackedChecksum == ticket.snapshotChecksum
+end
+
 local now = now_ms()
 cleanup(now)
 
@@ -398,6 +474,7 @@ if operation == 'replace_report' then
         end
     end
     local seen = {}
+    local counters = {}
     for _, counter in pairs(report.counters) do
         if type(counter) ~= 'table' or type(counter.proxyId) ~= 'string' or counter.proxyId == '' or
             type(counter.active) ~= 'number' or counter.active < 0 or counter.active ~= math.floor(counter.active) or
@@ -406,6 +483,7 @@ if operation == 'replace_report' then
             return reply('invalid')
         end
         seen[counter.proxyId] = true
+        counters[counter.proxyId] = counter
         local owner = decode_table(redis.call('HGET', owners_key, counter.proxyId))
         if not valid_owner(owner, report.workerId, report.ownershipEpoch, now) then
             return reply('stale')
@@ -418,6 +496,23 @@ if operation == 'replace_report' then
     session.expiresAtMs = now + ttl_ms
     redis.call('HSET', sessions_key, report.workerId, cjson.encode(session))
     redis.call('ZADD', session_expiry_key, session.expiresAtMs, report.workerId)
+
+    local ticket_ids = redis.call('ZRANGE', worker_draining_key, 0, cleanup_limit - 1)
+    for _, proxy_id in ipairs(ticket_ids) do
+        local ticket = decode_table(redis.call('HGET', drain_tickets_key, proxy_id))
+        local owner = decode_table(redis.call('HGET', owners_key, proxy_id))
+        if not ticket then
+            redis.call('ZREM', worker_draining_key, proxy_id)
+        elseif valid_drain_ticket(ticket, report.workerId) and ticket.workerIndexKey == worker_draining_key and
+            owner and owner.draining == true and owner.workerId == report.workerId and
+            tonumber(owner.epoch) == tonumber(ticket.assignmentEpoch) and
+            valid_owner(owner, report.workerId, report.ownershipEpoch, now) and ticket_acknowledged(session, ticket) then
+            local counter = counters[proxy_id]
+            if not counter or (counter.active == 0 and counter.reserved == 0) then
+                clear_finalized_drain(proxy_id, owner, ticket, now)
+            end
+        end
+    end
     return reply('ok')
 end
 

progress.md

@@ -11,14 +11,14 @@
 - 已完成持续 `UNHEALTHY` 后的安全淘汰：首次异常时间写入 Redis TTL 活动池，
   `check.unhealthyRemoveAfter` 按 Upstream 生效；Controller 通过公用、有界的
   Reaper 仅回收未分配代理。仍持有 Worker ownership 的候选会延后，待既有
-  Drain/ACK 清空 ownership 后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。
+  Drain 收尾清空 ownership 后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。
 - Redis 集成标签测试在未设置 `PROXY_POOL_TEST_REDIS_URL` 时现能正确跳过所有契约
   子测试；`go test ./...`、`go test -tags=integration ./internal/adapters/redisactivity`
   与 `scripts/verify.ps1` 已通过，race 仍由 Linux CI 执行。
 - Worker 控制面改为持续完整 Snapshot 发布：Controller 在每份 Snapshot 有效期的一半
   前重建并下发递增版本，Gateway 在同一长连接内原子替换视图；构建失败或流中断时才由
   SessionSupervisor 按原有退避重连。该机制复用 `RefreshingSnapshotSource`，不降低
-  Delta 的 fail-closed 校验，也不把自动 Drain/ACK 记为已完成。
+  Delta 的 fail-closed 校验。
 - Health Scheduler 现在先建立 EGRESS 与 Routing TARGET 的独立 due 引用，再投递
   BASIC 任务；在单项小批次下，短 TTL Proxy 不会因 BASIC 先出队而错过首次出口/目标探测。
 - 修复 Snapshot epoch/version 栅栏：Worker 流的 `version` 在 ownership epoch 前进时
@@ -28,7 +28,9 @@
   epoch 与所需 Snapshot epoch 组成的待绑定 Ticket；Redis 使用 `drain-tickets` 与按
   Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。Worker Handler
   在持久登记完整 Snapshot 引用后，核验代理确实不在完整视图内，再绑定 Ticket 的
-  session/version/epoch/checksum 屏障；ACK 与 Runtime 零计数的原子自动完成仍在后续切片。
+  session/version/epoch/checksum 屏障；同一 Redis Runtime 替换事务验证当前 session ACK、
+  Ticket 屏障和完整报告零计数后自动删除 owner/Ticket/Drain 索引并恢复 AVAILABLE 代理的
+  可分配索引，避免读写分离竞态。
 
 ## 2026-07-30
 

task_plan.md

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