feat: build worker proxy snapshots from ownership

README.md

@@ -42,8 +42,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
   审计与 Outbox；不保存 Proxy 明细或逐次提取记录。
 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、
   目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/
-  Runtime 会话组件已有实现与定向测试；`proxy-gateway` 命令、凭据分发和权威
+  Runtime 会话组件已有实现与定向测试。Controller 可向 Worker 下发无凭据引用的
-  Proxy/Routing Snapshot 发布仍待装配。
+  已归属 Proxy 快照；`proxy-gateway` 命令、凭据分发和 Routing Snapshot 发布仍待装配。
 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持
   CIDR、可信代理、严格请求解析和敏感信息最小化。
 
@@ -82,7 +82,7 @@ flowchart LR
   Redis 会话栅栏。
 - **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership
   与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。
-- **待完成**：Worker 权威 Proxy/Routing Snapshot 发布、Gateway 进程装配、Outcome
+- **待完成**：带凭据 Proxy 与 Routing 的权威 Snapshot 发布、Gateway 进程装配、Outcome
   上报、Checker 调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性集群压测。
 
 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表

docs/api/control-plane.md

@@ -26,10 +26,11 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和
 `controlPlane` 配置限定。
 
 `WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照，
-Gateway 校验后 ACK 并开始 Runtime 心跳。当前基础快照不包含权威 Proxy 或 Routing
+Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership
-内容，流会保持等待后续发布；权威快照发布器、增量、Gateway 进程装配、Outcome 与
+索引构建无凭据引用的已归属 Proxy 内容，并将租约到期收紧到 Proxy 的
-Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；`100,000 QPS`
+`usable_until`；带凭据 Proxy 会 fail-closed，直到凭据材料分发完成。Routing 内容、
-仍是未验证的设计目标。
+增量、Gateway 进程装配、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确
+返回 `Unimplemented`；`100,000 QPS` 仍是未验证的设计目标。
 
 ## 2. Worker 会话
 

docs/development/implementation-plan.md

@@ -210,10 +210,10 @@ Prometheus 运行时指标，三监听器隔离已通过测试；业务指标仍
 
 WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Runtime
 报告均经 Redis 服务端 TTL 的 session/issued-snapshot/ACK 栅栏校验；mTLS SPIFFE
-身份、消息/流限制和有界停机已实现。`WatchSnapshots` 会持续发送并保持当前 epoch
+身份、消息/流限制和有界停机已实现。`WatchSnapshots` 会发送当前 epoch 的基础完整
-的基础完整 Snapshot；Gateway 已具备 Register/Watch/ACK/Runtime 会话协调组件。
+Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话协调组件。
-按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本。权威 Proxy/Routing
+按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本，并可构建无凭据引用的
-payload 发布、凭据分发、Gateway 命令与 Outcome 上报仍未实现。
+已归属 Proxy payload。Routing payload、凭据分发、Gateway 命令与 Outcome 上报仍未实现。
 
 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider
 Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖

docs/requirements/completion-audit.md

@@ -51,8 +51,8 @@
   服务端 TTL，负向 ACK 会关闭 Runtime 写入栅栏。Controller 已装配 loopback
   plaintext fixture 与 SPIFFE mTLS 服务端；基础 Snapshot 流和 Gateway 的
   Register/Watch/ACK/Runtime 会话协调已实现。Redis 以 Worker 可下发 ownership
-  索引支持后续 payload 构建；权威 Proxy/Routing payload、凭据分发、Outcome 和
+  索引构建无凭据引用的已归属 Proxy payload，并以租约收紧可用期；Routing payload、
-  Checker 尚未闭环。
+  凭据分发、Outcome 和 Checker 尚未闭环。
 - `PostgreSQL 管理面`：已定义 `adminstate` 事务 seam、并发安全 MemoryStore、
   公用契约、100 并发 Routing CAS、租约 Outbox 和只含六张管理表的 Schema；pgx
   Adapter 已在真实 PostgreSQL 18 上通过同一契约、迁移幂等、审计/Outbox
@@ -94,9 +94,9 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt
 5. Worker Register/ACK/Runtime gRPC 接收路径、Redis 服务端 TTL 会话生命周期、
    SPIFFE mTLS 校验和 Controller 生命周期接线已完成；Redis Provider Leader、
    分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。
-6. Worker 基础网络快照流和 Gateway Snapshot 客户端已完成；权威 Proxy/Routing
+6. Worker 基础网络快照流、无凭据引用 Proxy payload 和 Gateway Snapshot 客户端已
-   payload、凭据分发与 Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收
+   完成；Routing payload、凭据分发与 Outcome 上报仍待完成。Redis ownership
-   及按 Worker 的可下发索引已完成。
+   drain/ACK/过期回收及按 Worker 的可下发索引已完成。
 7. Checker 调度、探测器和健康 reducer。
 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。
 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。

internal/adapters/redisactivity/ownership_integration_test.go

@@ -398,6 +398,51 @@ func TestRedisOwnershipAndMaintenanceValidateInputs(t *testing.T) {
 	}
 }
 
+func TestRedisReadWorkerSnapshotUsesWorkerOwnedIndex(t *testing.T) {
+	fixture := newRedisTestFixture(t)
+	now := redisTestNow()
+	candidate := testProxy("proxy-a", "192.0.2.10")
+	candidate.MaxConcurrency = 7
+	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*time.Minute, candidate)
+	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)
+	}
+	proxies, err := fixture.Adapter.ReadWorkerSnapshot(context.Background(), "worker-a", 1)
+	if err != nil || len(proxies) != 1 {
+		t.Fatalf("ReadWorkerSnapshot() = %+v, %v", proxies, err)
+	}
+	if proxies[0].Proxy.ID != "proxy-a" || proxies[0].OwnershipEpoch != assignment.Epoch ||
+		!proxies[0].LeaseExpiresAt.Equal(assignment.ExpiresAt) {
+		t.Fatalf("snapshot proxy = %+v, want assignment %+v", proxies[0], assignment)
+	}
+	other, err := fixture.Adapter.ReadWorkerSnapshot(context.Background(), "worker-b", 1)
+	if err != nil || len(other) != 0 {
+		t.Fatalf("ReadWorkerSnapshot(other) = %+v, %v", other, err)
+	}
+	if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(3 * time.Second), NextState: proxyDomain.StateSuspect,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(suspect): %v", err)
+	}
+	proxies, err = fixture.Adapter.ReadWorkerSnapshot(context.Background(), "worker-a", 1)
+	if err != nil || len(proxies) != 0 {
+		t.Fatalf("ReadWorkerSnapshot(suspect) = %+v, %v", proxies, err)
+	}
+	if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(4 * time.Second), NextState: proxyDomain.StateAvailable,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(available): %v", err)
+	}
+	proxies, err = fixture.Adapter.ReadWorkerSnapshot(context.Background(), "worker-a", 1)
+	if err != nil || len(proxies) != 1 {
+		t.Fatalf("ReadWorkerSnapshot(available) = %+v, %v", proxies, err)
+	}
+	if _, err := fixture.Adapter.ReadWorkerSnapshot(context.Background(), "worker-a", 0); !errors.Is(err, ownershipDomain.ErrInvalidOwnership) {
+		t.Fatalf("ReadWorkerSnapshot(zero limit) error = %v", err)
+	}
+}
+
 func assertRedisInventory(t *testing.T, adapter *Adapter, upstreamID string, now time.Time, want int) {
 	t.Helper()
 	inventory, err := adapter.Inventory(context.Background(), upstreamID, now)

internal/adapters/redisactivity/scripts.go

@@ -88,6 +88,17 @@ type capacityScriptReply struct {
 	AvailableSlots int64        `json:"availableSlots,string"`
 }
 
+type workerSnapshotScriptReply struct {
+	Status  scriptStatus              `json:"status"`
+	Proxies []workerSnapshotProxyWire `json:"proxies"`
+}
+
+type workerSnapshotProxyWire struct {
+	Record           string `json:"record"`
+	OwnershipEpoch   string `json:"ownershipEpoch"`
+	LeaseExpiresAtMS int64  `json:"leaseExpiresAtMs"`
+}
+
 //go:embed scripts/upsert.lua
 var upsertSource string
 
@@ -112,6 +123,9 @@ var runtimeSource string
 //go:embed scripts/capacity.lua
 var capacitySource string
 
+//go:embed scripts/worker_snapshot.lua
+var workerSnapshotSource string
+
 var (
 	upsertScript         = redis.NewScript(upsertSource)
 	healthScript         = redis.NewScript(healthSource)
@@ -121,6 +135,7 @@ var (
 	statusScript         = redis.NewScript(statusSource)
 	runtimeScript        = redis.NewScript(runtimeSource)
 	capacityScript       = redis.NewScript(capacitySource)
+	workerSnapshotScript = redis.NewScript(workerSnapshotSource)
 )
 
 func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) {

internal/adapters/redisactivity/scripts/health.lua

@@ -88,6 +88,29 @@ local function remove_worker_owned(id)
     end
 end
 
+local function sync_worker_owned(id, record)
+    local raw = redis.call('HGET', owners_key, id)
+    if not raw then
+        return
+    end
+    local decoded, assignment = pcall(cjson.decode, raw)
+    if not decoded or type(assignment) ~= 'table' or
+        type(assignment.workerIndexKey) ~= 'string' or assignment.workerIndexKey == '' then
+        return
+    end
+    if assignment.workerId == record.ownerWorkerId and not assignment.draining and
+        type(assignment.expiresAtMs) == 'number' and assignment.expiresAtMs > checked_at_ms and
+        record.state == 'AVAILABLE' and tonumber(record.usableUntilMs) > checked_at_ms then
+        redis.call('ZADD', assignment.workerIndexKey, assignment.expiresAtMs, id)
+        local current = redis.call('PEXPIRETIME', assignment.workerIndexKey)
+        if current < tonumber(record.expiresAtMs) then
+            redis.call('PEXPIREAT', assignment.workerIndexKey, tonumber(record.expiresAtMs))
+        end
+        return
+    end
+    redis.call('ZREM', assignment.workerIndexKey, id)
+end
+
 local function remove_proxy(id)
     local raw = redis.call('HGET', records_key, id)
     local record = nil
@@ -220,6 +243,7 @@ if next_state == 'AVAILABLE' and tonumber(record.usableUntilMs) > checked_at_ms
         end
     end
 end
+sync_worker_owned(proxy_id, record)
 touch(records_key, tonumber(record.expiresAtMs))
 touch(unique_key, tonumber(record.expiresAtMs))
 touch(idkeys_key, tonumber(record.expiresAtMs))

internal/adapters/redisactivity/scripts/upsert.lua

@@ -88,6 +88,29 @@ local function remove_worker_owned(proxy_id)
     end
 end
 
+local function sync_worker_owned(proxy_id, record)
+    local raw = redis.call('HGET', owners_key, proxy_id)
+    if not raw then
+        return
+    end
+    local decoded, assignment = pcall(cjson.decode, raw)
+    if not decoded or type(assignment) ~= 'table' or
+        type(assignment.workerIndexKey) ~= 'string' or assignment.workerIndexKey == '' then
+        return
+    end
+    if assignment.workerId == record.ownerWorkerId and not assignment.draining and
+        type(assignment.expiresAtMs) == 'number' and assignment.expiresAtMs > now_ms and
+        record.state == 'AVAILABLE' and tonumber(record.usableUntilMs) > now_ms then
+        redis.call('ZADD', assignment.workerIndexKey, assignment.expiresAtMs, proxy_id)
+        local current = redis.call('PEXPIRETIME', assignment.workerIndexKey)
+        if current < tonumber(record.expiresAtMs) then
+            redis.call('PEXPIREAT', assignment.workerIndexKey, tonumber(record.expiresAtMs))
+        end
+        return
+    end
+    redis.call('ZREM', assignment.workerIndexKey, proxy_id)
+end
+
 local function remove_proxy(proxy_id)
     local raw = redis.call('HGET', records_key, proxy_id)
     local record = nil
@@ -210,6 +233,7 @@ for _, candidate in ipairs(candidates) do
             redis.call('HSET', records_key, incumbent_id, encoded)
             redis.call('ZADD', expiry_key, incoming.expiresAtMs, incumbent_id)
             sync_owned(incumbent_id, incoming)
+            sync_worker_owned(incumbent_id, incoming)
             add_available(incumbent_id, incoming)
             if tonumber(incoming.expiresAtMs) > max_expiry_ms then
                 max_expiry_ms = tonumber(incoming.expiresAtMs)
@@ -229,6 +253,7 @@ for _, candidate in ipairs(candidates) do
             redis.call('HSET', idkeys_key, candidate.proxyId, candidate.uniqueDigest)
             redis.call('ZADD', expiry_key, incoming.expiresAtMs, candidate.proxyId)
             sync_owned(candidate.proxyId, incoming)
+            sync_worker_owned(candidate.proxyId, incoming)
             if is_managed(incoming.state) then
                 redis.call('HINCRBY', inventory_key, candidate.upstream, 1)
             end

internal/adapters/redisactivity/scripts/worker_snapshot.lua

@@ -0,0 +1,75 @@
+local records_key = KEYS[1]
+local owners_key = KEYS[2]
+local worker_owned_key = KEYS[3]
+
+local worker_id = ARGV[1]
+local limit = tonumber(ARGV[2])
+
+local function reply(status, proxies)
+    if not proxies or #proxies == 0 then
+        return '{"status":' .. cjson.encode(status) .. ',"proxies":[]}'
+    end
+    return cjson.encode({status = status, proxies = proxies})
+end
+
+local function now_ms()
+    local value = redis.call('TIME')
+    return tonumber(value[1]) * 1000 + math.floor(tonumber(value[2]) / 1000)
+end
+
+local function decode_table(value)
+    if not value then
+        return nil
+    end
+    local ok, decoded = pcall(cjson.decode, value)
+    if not ok or type(decoded) ~= 'table' then
+        return nil
+    end
+    return decoded
+end
+
+local function valid_assignment(value, now)
+    return value and value.version == 1 and type(value.proxyId) == 'string' and value.proxyId ~= '' and
+        type(value.workerId) == 'string' and value.workerId == worker_id and type(value.epoch) == 'number' and
+        value.epoch > 0 and type(value.expiresAtMs) == 'number' and value.expiresAtMs > now and
+        type(value.draining) == 'boolean' and not value.draining and
+        type(value.workerIndexKey) == 'string' and value.workerIndexKey == worker_owned_key
+end
+
+local function valid_record(value, proxy_id, now)
+    return value and value.version == 1 and value.id == proxy_id and value.state == 'AVAILABLE' and
+        value.ownerWorkerId == worker_id and type(value.scheme) == 'string' and
+        (value.scheme == 'http' or value.scheme == 'https' or value.scheme == 'socks5') and
+        type(value.host) == 'string' and value.host ~= '' and type(value.port) == 'number' and
+        value.port > 0 and value.port <= 65535 and type(value.sourceUpstream) == 'string' and
+        value.sourceUpstream ~= '' and type(value.expiresAtMs) == 'number' and value.expiresAtMs > now and
+        type(value.usableUntilMs) == 'number' and value.usableUntilMs > now and
+        type(value.maxConcurrency) == 'number' and value.maxConcurrency > 0
+end
+
+if type(worker_id) ~= 'string' or worker_id == '' or not limit or limit <= 0 then
+    return reply('invalid')
+end
+
+if redis.call('ZCARD', worker_owned_key) > limit then
+    return reply('unavailable')
+end
+
+local now = now_ms()
+local ids = redis.call('ZRANGEBYSCORE', worker_owned_key, now + 1, '+inf', 'LIMIT', 0, limit)
+local proxies = cjson.decode('[]')
+for _, proxy_id in ipairs(ids) do
+    local assignment = decode_table(redis.call('HGET', owners_key, proxy_id))
+    local raw_record = redis.call('HGET', records_key, proxy_id)
+    local record = decode_table(raw_record)
+    if valid_assignment(assignment, now) and valid_record(record, proxy_id, now) then
+        proxies[#proxies + 1] = {
+            record = raw_record,
+            ownershipEpoch = tostring(assignment.epoch),
+            leaseExpiresAtMs = assignment.expiresAtMs,
+        }
+    else
+        redis.call('ZREM', worker_owned_key, proxy_id)
+    end
+end
+return reply('ok', proxies)

internal/adapters/redisactivity/worker_snapshot.go

@@ -0,0 +1,69 @@
+package redisactivity
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"time"
+
+	ownershipDomain "proxy-pool/internal/domain/ownership"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+var _ ownershipDomain.SnapshotReader = (*Adapter)(nil)
+
+// ReadWorkerSnapshot returns every currently assignable proxy owned by one
+// Worker. The Lua script rejects an oversized view rather than returning a
+// partial snapshot that could silently withdraw still-owned proxies.
+func (a *Adapter) ReadWorkerSnapshot(
+	ctx context.Context,
+	workerID string,
+	limit int,
+) ([]ownershipDomain.SnapshotProxy, error) {
+	if ctx == nil || a == nil || !runtimeClean(workerID) || limit <= 0 {
+		return nil, ownershipDomain.ErrInvalidOwnership
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	result, err := runScript(ctx, a.client, workerSnapshotScript, []string{
+		a.keys.records, a.keys.owners, a.keys.workerOwned(workerID),
+	}, workerID, limit)
+	if err != nil {
+		return nil, err
+	}
+	var reply workerSnapshotScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return nil, err
+	}
+	switch reply.Status {
+	case scriptOK:
+	case scriptInvalid:
+		return nil, ownershipDomain.ErrInvalidOwnership
+	case scriptUnavailable:
+		return nil, errors.Join(ownershipDomain.ErrOwnershipUnavailable, errors.New("worker snapshot limit exceeded"))
+	default:
+		return nil, invalidScriptReply("unexpected worker snapshot reply")
+	}
+	proxies := make([]ownershipDomain.SnapshotProxy, 0, len(reply.Proxies))
+	for _, item := range reply.Proxies {
+		record, err := decodeProxyRecord(item.Record)
+		if err != nil {
+			return nil, invalidScriptReply("worker snapshot contained invalid proxy record")
+		}
+		epoch, err := strconv.ParseUint(item.OwnershipEpoch, 10, 64)
+		if err != nil || epoch == 0 || item.LeaseExpiresAtMS <= 0 {
+			return nil, invalidScriptReply("worker snapshot contained invalid ownership")
+		}
+		proxy := proxyRecordEntry(record).Proxy
+		if proxy.State != proxyDomain.StateAvailable || proxy.MaxConcurrency <= 0 || proxy.ExpiresAt == nil ||
+			proxy.UsableUntil == nil {
+			return nil, invalidScriptReply("worker snapshot contained unavailable proxy")
+		}
+		proxies = append(proxies, ownershipDomain.SnapshotProxy{
+			Proxy: proxy, OwnershipEpoch: epoch,
+			LeaseExpiresAt: time.UnixMilli(item.LeaseExpiresAtMS).UTC(),
+		})
+	}
+	return proxies, nil
+}

internal/controller/bootstrap/bootstrap.go

@@ -22,6 +22,7 @@ import (
 	"proxy-pool/internal/controller/worker"
 	"proxy-pool/internal/domain/activitypool"
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	ownershipDomain "proxy-pool/internal/domain/ownership"
 	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
 	"proxy-pool/internal/platform/credentials"
@@ -232,12 +233,18 @@ func runWithWorkerFactory(
 		if nilInterface(opened.workerStore) {
 			return errors.Join(ErrStartup, ErrInvalidOptions)
 		}
+		var snapshotReader ownershipDomain.SnapshotReader
+		if reader, ok := opened.workerStore.(ownershipDomain.SnapshotReader); ok {
+			snapshotReader = reader
+		}
 		service, serviceErr := worker.NewService(opened.workerStore, worker.Options{
 			ProtocolVersion:    loaded.Value.ControlPlane.ProtocolVersion,
 			HeartbeatInterval:  loaded.Value.ControlPlane.HeartbeatInterval.Value(),
 			SessionTTL:         loaded.Value.ControlPlane.SessionTTL.Value(),
 			MaxStaleAge:        loaded.Value.ControlPlane.MaxStaleAge.Value(),
 			MaxRuntimeCounters: loaded.Value.ControlPlane.MaxRuntimeCounters,
+			MaxSnapshotBytes:   loaded.Value.ControlPlane.MaxMessageBytes,
+			SnapshotReader:     snapshotReader,
 		})
 		if serviceErr != nil {
 			return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr)

internal/controller/worker/owned_snapshot_source.go

@@ -0,0 +1,142 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"math"
+	"time"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/controlplane/snapshotwire"
+	ownershipDomain "proxy-pool/internal/domain/ownership"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+	"proxy-pool/internal/domain/workerruntime"
+
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+var ErrSnapshotCredentialsUnavailable = errors.New("worker snapshot credential material is unavailable")
+
+// OwnedSnapshotSource builds a complete Worker view from the Redis ownership
+// index. It only accepts a complete bounded result from the reader.
+type OwnedSnapshotSource struct {
+	epochs     OwnershipEpochReader
+	reader     ownershipDomain.SnapshotReader
+	validFor   time.Duration
+	maxProxies int
+	maxBytes   int
+	now        func() time.Time
+}
+
+func NewOwnedSnapshotSource(
+	epochs OwnershipEpochReader,
+	reader ownershipDomain.SnapshotReader,
+	validFor time.Duration,
+	maxProxies int,
+	maxBytes int,
+	now func() time.Time,
+) (*OwnedSnapshotSource, error) {
+	if epochs == nil || reader == nil || validFor <= 0 || maxProxies <= 0 || maxBytes <= 0 || now == nil {
+		return nil, ErrSnapshotsUnavailable
+	}
+	return &OwnedSnapshotSource{
+		epochs: epochs, reader: reader, validFor: validFor, maxProxies: maxProxies, maxBytes: maxBytes, now: now,
+	}, nil
+}
+
+func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWatchRequest) (<-chan *controlplanev1.WorkerSnapshot, error) {
+	if source == nil || ctx == nil || !workerruntime.ValidIdentifier(request.WorkerID) ||
+		!workerruntime.ValidIdentifier(request.SessionID) || request.LastAppliedVersion == math.MaxUint64 {
+		return nil, ErrSnapshotsUnavailable
+	}
+	epoch, err := source.epochs.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if epoch == 0 {
+		return nil, ErrSnapshotsUnavailable
+	}
+	now := source.now().UTC()
+	if now.IsZero() {
+		return nil, ErrSnapshotsUnavailable
+	}
+	proxies, err := source.reader.ReadWorkerSnapshot(ctx, request.WorkerID, source.maxProxies)
+	if err != nil {
+		return nil, err
+	}
+	full := &controlplanev1.WorkerSnapshot{
+		Version: request.LastAppliedVersion + 1, OwnershipEpoch: epoch,
+		GeneratedAt: timestamppb.New(now), ValidUntil: timestamppb.New(now.Add(source.validFor)),
+		Proxies: make([]*controlplanev1.OwnedProxy, 0, len(proxies)),
+	}
+	for _, item := range proxies {
+		owned, validUntil, err := wireOwnedProxy(item, now)
+		if err != nil {
+			return nil, err
+		}
+		if validUntil.Before(full.ValidUntil.AsTime()) {
+			full.ValidUntil = timestamppb.New(validUntil)
+		}
+		full.Proxies = append(full.Proxies, owned)
+	}
+	checksum, err := snapshotwire.Checksum(full)
+	if err != nil {
+		return nil, err
+	}
+	full.Checksum = append([]byte(nil), checksum[:]...)
+	if proto.Size(&controlplanev1.SnapshotEnvelope{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: full}}) > source.maxBytes {
+		return nil, ErrSnapshotsUnavailable
+	}
+	updates := make(chan *controlplanev1.WorkerSnapshot, 1)
+	updates <- full
+	return updates, nil
+}
+
+func wireOwnedProxy(item ownershipDomain.SnapshotProxy, now time.Time) (*controlplanev1.OwnedProxy, time.Time, error) {
+	proxy := item.Proxy
+	if item.OwnershipEpoch == 0 || item.LeaseExpiresAt.IsZero() || proxy.CredentialVersion != "" || proxy.SecretRef != "" {
+		if proxy.CredentialVersion != "" || proxy.SecretRef != "" {
+			return nil, time.Time{}, ErrSnapshotCredentialsUnavailable
+		}
+		return nil, time.Time{}, ErrSnapshotsUnavailable
+	}
+	protocol, ok := wireProtocol(proxy.Scheme)
+	if !ok || !workerruntime.ValidIdentifier(proxy.ID) || !workerruntime.ValidIdentifier(proxy.SourceUpstream) ||
+		proxy.Host == "" || proxy.Port == 0 || proxy.MaxConcurrency <= 0 || proxy.MaxConcurrency > math.MaxUint32 ||
+		proxy.ExpiresAt == nil || proxy.UsableUntil == nil {
+		return nil, time.Time{}, ErrSnapshotsUnavailable
+	}
+	validUntil := proxy.UsableUntil.UTC()
+	if item.LeaseExpiresAt.Before(validUntil) {
+		validUntil = item.LeaseExpiresAt.UTC()
+	}
+	if !validUntil.After(now) || !proxy.ExpiresAt.After(now) {
+		return nil, time.Time{}, ErrSnapshotsUnavailable
+	}
+	tags := make(map[string]string, len(proxy.Tags))
+	for key, value := range proxy.Tags {
+		tags[key] = value
+	}
+	return &controlplanev1.OwnedProxy{
+		Id: proxy.ID, Upstream: proxy.SourceUpstream, Protocol: protocol, Host: proxy.Host, Port: uint32(proxy.Port),
+		Username: proxy.Username, ExpiresAt: timestamppb.New(proxy.ExpiresAt.UTC()),
+		MaxConcurrency: uint32(proxy.MaxConcurrency), Tags: tags, OwnershipEpoch: item.OwnershipEpoch,
+		UsableUntil: timestamppb.New(validUntil),
+	}, validUntil, nil
+}
+
+func wireProtocol(scheme proxyDomain.Scheme) (controlplanev1.ProxyProtocol, bool) {
+	switch scheme {
+	case proxyDomain.SchemeHTTP:
+		return controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, true
+	case proxyDomain.SchemeHTTPS:
+		return controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTPS, true
+	case proxyDomain.SchemeSOCKS5:
+		return controlplanev1.ProxyProtocol_PROXY_PROTOCOL_SOCKS5, true
+	default:
+		return controlplanev1.ProxyProtocol_PROXY_PROTOCOL_UNSPECIFIED, false
+	}
+}
+
+var _ SnapshotSource = (*OwnedSnapshotSource)(nil)

internal/controller/worker/server.go

@@ -52,6 +52,11 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server
 		return nil, err
 	}
 	snapshots := options.Snapshots
+	if snapshots == nil {
+		if provider, ok := service.(interface{ SnapshotSource() SnapshotSource }); ok {
+			snapshots = provider.SnapshotSource()
+		}
+	}
 	if snapshots == nil {
 		snapshots, err = NewInitialSnapshotSource(service, controlPlane.MaxStaleAge.Value(), time.Now)
 		if err != nil {

internal/controller/worker/service.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"time"
 
+	ownershipDomain "proxy-pool/internal/domain/ownership"
 	"proxy-pool/internal/domain/workerruntime"
 )
 
@@ -55,6 +56,8 @@ type Options struct {
 	SessionTTL         time.Duration
 	MaxStaleAge        time.Duration
 	MaxRuntimeCounters int
+	MaxSnapshotBytes   int
+	SnapshotReader     ownershipDomain.SnapshotReader
 	SessionID          func() (string, error)
 }
 
@@ -100,6 +103,7 @@ func (service *service) IssueSnapshot(ctx context.Context, reference workerrunti
 type service struct {
 	store     workerruntime.ControlStore
 	options   Options
+	snapshots SnapshotSource
 }
 
 func NewService(store workerruntime.ControlStore, options Options) (Service, error) {
@@ -111,7 +115,24 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err
 	if options.SessionID == nil {
 		options.SessionID = randomSessionID
 	}
-	return &service{store: store, options: options}, nil
+	result := &service{store: store, options: options}
+	if options.SnapshotReader != nil {
+		source, err := NewOwnedSnapshotSource(
+			result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now,
+		)
+		if err != nil {
+			return nil, errors.Join(ErrInvalidCommand, err)
+		}
+		result.snapshots = source
+	}
+	return result, nil
+}
+
+func (service *service) SnapshotSource() SnapshotSource {
+	if service == nil {
+		return nil
+	}
+	return service.snapshots
 }
 
 func (service *service) Register(ctx context.Context, command RegisterCommand) (Registration, error) {

internal/controller/worker/snapshot_source_test.go

@@ -2,10 +2,13 @@ package worker
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
 	"proxy-pool/internal/controlplane/snapshotwire"
+	ownershipDomain "proxy-pool/internal/domain/ownership"
+	proxyDomain "proxy-pool/internal/domain/proxy"
 )
 
 func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) {
@@ -30,6 +33,61 @@ func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) {
 	}
 }
 
+func TestOwnedSnapshotSourceBuildsBoundedProxySnapshot(t *testing.T) {
+	now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC)
+	expiresAt := now.Add(10 * time.Minute)
+	usableUntil := now.Add(5 * time.Minute)
+	source, err := NewOwnedSnapshotSource(epochReaderStub{epoch: 9}, snapshotReaderStub{proxies: []ownershipDomain.SnapshotProxy{{
+		Proxy: proxyDomain.Proxy{
+			ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
+			SourceUpstream: "upstream-a", ExpiresAt: &expiresAt, UsableUntil: &usableUntil,
+			MaxConcurrency: 7, State: proxyDomain.StateAvailable, Tags: map[string]string{"region": "cn"},
+		},
+		OwnershipEpoch: 4, LeaseExpiresAt: now.Add(time.Minute),
+	}}}, time.Minute*2, 10, 4096, func() time.Time { return now })
+	if err != nil {
+		t.Fatalf("NewOwnedSnapshotSource(): %v", err)
+	}
+	updates, err := source.Watch(context.Background(), SnapshotWatchRequest{WorkerID: "worker-a", SessionID: "session-a"})
+	if err != nil {
+		t.Fatalf("Watch(): %v", err)
+	}
+	full := <-updates
+	if full.GetVersion() != 1 || full.GetOwnershipEpoch() != 9 || len(full.GetProxies()) != 1 {
+		t.Fatalf("snapshot = %+v", full)
+	}
+	proxy := full.GetProxies()[0]
+	if proxy.GetOwnershipEpoch() != 4 || !proxy.GetUsableUntil().AsTime().Equal(now.Add(time.Minute)) ||
+		!full.GetValidUntil().AsTime().Equal(now.Add(time.Minute)) {
+		t.Fatalf("wire proxy = %+v, valid until = %s", proxy, full.GetValidUntil().AsTime())
+	}
+	checksum, err := snapshotwire.Checksum(full)
+	if err != nil || string(checksum[:]) != string(full.GetChecksum()) {
+		t.Fatalf("snapshot checksum = %x, %v; want %x", full.GetChecksum(), err, checksum)
+	}
+}
+
+func TestOwnedSnapshotSourceRejectsCredentialReferenceUntilMaterialIsAvailable(t *testing.T) {
+	now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC)
+	expiresAt := now.Add(time.Minute)
+	usableUntil := now.Add(time.Minute)
+	source, err := NewOwnedSnapshotSource(epochReaderStub{epoch: 1}, snapshotReaderStub{proxies: []ownershipDomain.SnapshotProxy{{
+		Proxy: proxyDomain.Proxy{
+			ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
+			SourceUpstream: "upstream-a", ExpiresAt: &expiresAt, UsableUntil: &usableUntil,
+			MaxConcurrency: 1, State: proxyDomain.StateAvailable, CredentialVersion: "v1",
+		},
+		OwnershipEpoch: 1, LeaseExpiresAt: now.Add(time.Minute),
+	}}}, time.Minute, 10, 4096, func() time.Time { return now })
+	if err != nil {
+		t.Fatalf("NewOwnedSnapshotSource(): %v", err)
+	}
+	_, err = source.Watch(context.Background(), SnapshotWatchRequest{WorkerID: "worker-a", SessionID: "session-a"})
+	if !errors.Is(err, ErrSnapshotCredentialsUnavailable) {
+		t.Fatalf("Watch() error = %v, want ErrSnapshotCredentialsUnavailable", err)
+	}
+}
+
 type epochReaderStub struct {
 	epoch uint64
 	err   error
@@ -38,3 +96,12 @@ type epochReaderStub struct {
 func (reader epochReaderStub) CurrentOwnershipEpoch(context.Context) (uint64, error) {
 	return reader.epoch, reader.err
 }
+
+type snapshotReaderStub struct {
+	proxies []ownershipDomain.SnapshotProxy
+	err     error
+}
+
+func (reader snapshotReaderStub) ReadWorkerSnapshot(context.Context, string, int) ([]ownershipDomain.SnapshotProxy, error) {
+	return reader.proxies, reader.err
+}

internal/domain/ownership/snapshot.go

@@ -0,0 +1,22 @@
+package ownership
+
+import (
+	"context"
+	"time"
+
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+// SnapshotProxy is one currently assignable proxy in a Worker-specific view.
+// LeaseExpiresAt bounds how long the view may remain valid without refresh.
+type SnapshotProxy struct {
+	Proxy          proxyDomain.Proxy
+	OwnershipEpoch uint64
+	LeaseExpiresAt time.Time
+}
+
+// SnapshotReader returns a complete bounded Worker view. Implementations must
+// never silently truncate the result when the requested limit is exceeded.
+type SnapshotReader interface {
+	ReadWorkerSnapshot(context.Context, string, int) ([]SnapshotProxy, error)
+}