feat: add gateway runtime control plane reporter

README.md

@@ -71,18 +71,18 @@ flowchart LR
 
 ## 当前完成度
 
-截至 **2026-07-30**，实施计划检查项为 **52 / 74（70.3%）**。详情见
+截至 **2026-07-31**，实施计划检查项为 **53 / 74（71.6%）**。详情见
 [实施计划](docs/development/implementation-plan.md)和
 [交付完成度审计](docs/requirements/completion-audit.md)。
 
 - **已完成**：严格配置、Provider 获取与协调、Redis 活动池、Distribution 原子
   提取与限流、Controller 的 Admin/Distribution/Metrics 监听，以及 PostgreSQL
-  管理状态。
+  管理状态；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和
+  Redis 会话栅栏。
 - **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership
   与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。
-- **待完成**：WorkerControlPlane gRPC session/snapshot/ACK 闭环、Checker 调度与
+- **待完成**：Worker Snapshot 下发流、Gateway 快照客户端、Outcome 上报、Checker
-  健康状态链、Gateway 进程与快照客户端、完整 Routing 运行链，以及 loadgen 和
+  调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性集群压测。
-  代表性集群压测。
 
 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表
 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。

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

@@ -66,6 +66,7 @@ var _ ownership.Repository = (*Adapter)(nil)
 var _ workerruntime.SessionWriter = (*Adapter)(nil)
 var _ workerruntime.ReportWriter = (*Adapter)(nil)
 var _ workerruntime.RuntimeReader = (*Adapter)(nil)
+var _ workerruntime.ControlStore = (*Adapter)(nil)
 var _ pool.InventoryReader = (*Adapter)(nil)
 ```
 
@@ -103,6 +104,8 @@ pp:{activity}:epoch               STRING ownership 全局递增代次
 pp:{activity}:inventory           HASH   upstreamID -> 当前未提取库存
 pp:{activity}:worker-sessions     HASH   workerID -> 当前 Worker session
 pp:{activity}:worker-session-expiry ZSET workerID -> session expiry milliseconds
+pp:{activity}:worker-snapshots    HASH   workerID -> 最近签发且待确认的 Snapshot 引用
+pp:{activity}:worker-snapshot-expiry ZSET workerID -> Snapshot 引用 expiry milliseconds
 pp:{activity}:worker-runtime      HASH   workerID -> 完整稀疏运行态报告
 pp:{activity}:worker-runtime-expiry ZSET workerID -> report expiry milliseconds
 pp:{activity}:owned:<digest>      ZSET   单 Upstream 已分配 AVAILABLE Proxy

docs/api/control-plane.md

@@ -12,6 +12,23 @@ Proto 源文件位于 `api/proto/controlplane/v1/controlplane.proto`，包含两
 该协议不承载 Client 的独占提取，也没有 extraction lease/release。Proxy 的
 `AVAILABLE -> EXTRACTED` 只在 Controller 调用的 Redis 原子操作中完成。
 
+## 当前实现状态
+
+Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和
+`ReportRuntime` 的一元 RPC。Register 创建带 Redis 服务端 TTL 的 session；ACK 只
+接受 Controller 已签发的 `(version, ownership_epoch, checksum)`；Runtime 的空
+`counters` 是完整稀疏替换，也续期 session。负向 ACK 会关闭该 session 的 Runtime
+写入栅栏，直到收到新的正向 ACK，延迟的旧报告不能重新开启它。
+
+生产配置使用 mTLS，并将叶子证书 SPIFFE URI 约束为
+`spiffe://<trust-domain>/<environment>/worker/<worker-id>`；仅经配置校验的回环
+监听允许明文 fixture 模式。单消息大小、并发流数和 gRPC keepalive 策略由
+`controlPlane` 配置限定。
+
+`WatchSnapshots` 和 `ReportOutcomes` 当前明确返回 `Unimplemented`。
+Snapshot payload/stream、Gateway 客户端、Outcome 与 Checker 闭环尚未实现；
+`100,000 QPS` 仍是未验证的设计目标。
+
 ## 2. Worker 会话
 
 ```mermaid

docs/development/implementation-plan.md

@@ -176,7 +176,7 @@ Upstream、`endBehavior` 默认 `stop`，并覆盖列表末端停止；disabled
   sweep primitives with a monotonic global epoch.
 - [x] Implement Redis Provider leader, distributed request quota, Client limit and
   automatic Provider inventory rebuild after Redis state loss.
-- [ ] Implement the Worker heartbeat receiving path and session lifecycle.
+- [x] Implement the Worker heartbeat receiving path and session lifecycle.
 - [x] Keep Provider output in Redis TTL activity state and node memory only; keep the
   Gateway request path on immutable local snapshots with no Redis/PostgreSQL calls.
 - [x] Expose Distribution extraction/status and Admin status/enable/disable/switch/reload
@@ -184,7 +184,7 @@ Upstream、`endBehavior` 默认 `stop`，并覆盖列表末端停止；disabled
 - [x] Add Compose-backed Redis 8.2 integration and shared Adapter contract tests.
 - [x] Add PostgreSQL management Adapter and Compose-backed integration tests.
 
-当前进度（2026-07-30）：已实现共享 `platform/httpapi`、Distribution
+当前进度（2026-07-31）：已实现共享 `platform/httpapi`、Distribution
 extract/live/ready Handler 与 Admin status/enable/disable/switch/reload Handler；
 定向契约测试已覆盖严格 JSON、Body 上限、Request ID、幂等 Header、DTO 映射、
 404/405 及业务错误映射。共享 `platform/httpsecurity` 已补齐 Basic/API Key/
@@ -203,11 +203,16 @@ Distribution/Admin 服务构造、错误合并和资源关闭。生产 Provider
 权威管理状态动态装配 Upstream，并与 HTTP Runtime 通过公用 lifecycle Group 联动
 停机；Admin disable 会取消 Runtime，reload 在提交前预检并在发布后替换运行实例。
 组合 fixture 已验证隔离 Redis namespace 下的选主、Provider HTTP 调用、模板解析
-和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、
+和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、`/readyz` 与基础
-`/readyz` 与基础 Prometheus 运行时指标，三监听器隔离已通过测试；业务指标仍待
+Prometheus 运行时指标，三监听器隔离已通过测试；业务指标仍待
 实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖
 迁移、启动配置提交、Readiness、Admin Status 和 Metrics 探针。
 
+WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Runtime
+报告均经 Redis 服务端 TTL 的 session/issued-snapshot/ACK 栅栏校验；mTLS SPIFFE
+身份、消息/流限制和有界停机已实现。Snapshot 下发流、Gateway 客户端与 Outcome
+上报仍未实现。
+
 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider
 Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖
 `usableUntil` 向 Worker Snapshot 的传播与 Gateway 本地截止过滤、

docs/requirements/completion-audit.md

@@ -46,6 +46,11 @@
 - `Redis Activity Adapter`：真实 Redis 8.2 已覆盖 Provider Upsert、健康更新、
   原子独占提取、短期幂等、Worker ownership、库存和有界过期清理，Memory/Redis
   运行同一公用契约。
+- `WorkerControlPlane`：Register、已签发 Snapshot 的 ACK 和完整稀疏 Runtime
+  报告已通过 gRPC、Memory 与 Redis 8.2 契约验证；session、ACK、报告均使用 Redis
+  服务端 TTL，负向 ACK 会关闭 Runtime 写入栅栏。Controller 已装配 loopback
+  plaintext fixture 与 SPIFFE mTLS 服务端；Snapshot 流、Gateway 客户端、Outcome 和
+  Checker 尚未闭环。
 - `PostgreSQL 管理面`：已定义 `adminstate` 事务 seam、并发安全 MemoryStore、
   公用契约、100 并发 Routing CAS、租约 Outbox 和只含六张管理表的 Schema；pgx
   Adapter 已在真实 PostgreSQL 18 上通过同一契约、迁移幂等、审计/Outbox
@@ -84,9 +89,11 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt
 4. Controller 的 PostgreSQL 连接池、迁移和 pgx Adapter 启动装配已完成；
    公用 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 双存储集成，Controller
    三监听器与探针集成已完成；可选聚合指标和完整容器进程部署验证仍待实现。
-5. Worker heartbeat gRPC 接收路径；Redis Provider Leader、分布式请求额度、
+5. Worker Register/ACK/Runtime gRPC 接收路径、Redis 服务端 TTL 会话生命周期、
-   Distribution Client 限制和 Provider 状态丢失重建已完成。
+   SPIFFE mTLS 校验和 Controller 生命周期接线已完成；Redis Provider Leader、
-6. Worker 网络快照流；Redis ownership drain/ACK/过期回收已完成。
+   分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。
+6. Worker 网络快照流、Gateway Snapshot 客户端和 Outcome 上报仍待完成；Redis
+   ownership drain/ACK/过期回收已完成。
 7. Checker 调度、探测器和健康 reducer。
 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。
 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。

docs/requirements/traceability.md

@@ -7,7 +7,7 @@
 
 | ID | 最终需求 | 来源 | 验证证据 |
 |---|---|---|---|
-| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 命令已实现，Gateway/Checker/Loadgen 构建产物待实现 |
+| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/ACK/Runtime gRPC 接收，Gateway/Checker/Loadgen 构建产物及 Snapshot 下发待实现 |
 | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Snapshot/Dispatch 及依赖边界已验证；完整 Gateway 进程与代表性性能剖析待完成 |
 | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 生产入口待装配 |
 | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 |
@@ -46,7 +46,7 @@
 | PROXY-002 | 唯一键包含 scheme、host、port、username、credentialVersion | 6655-6727, 8605-8678 | 去重单测 |
 | PROXY-003 | TTL 来源优先级明确并统一 UTC | 681-747, 8655-8678 | TTL 表驱动测试 |
 | CAP-001 | Gateway 分配使用 Reserved -> Active 原子转换 | 1203-1467, 8530-8597 | 固定 Max 下打包 CAS 与 1,000 并发不超卖已完成；动态降容和完整生命周期证据待完成 |
-| CAP-002 | 补池依据 Available Slots，不只看 Proxy 数量 | 1203-1402, 8530-8597 | `AvailableSlots`、显式 minimum/target 水位、pending 槽位和迟滞 Reconciler 已测试；Gateway 打包 Active/Reserved 报告、Worker session/ACK/sequence/TTL/ownership fence、单 Upstream 索引及 Redis 权威 Managed/Slots 汇总已通过内存与真实 Redis 测试；WorkerControlPlane 接线、目标健康和 Gateway reserve 策略仍待完成 |
+| CAP-002 | 补池依据 Available Slots，不只看 Proxy 数量 | 1203-1402, 8530-8597 | `AvailableSlots`、显式 minimum/target 水位、pending 槽位和迟滞 Reconciler 已测试；Worker session/ACK/sequence/TTL/ownership fence 与 WorkerControlPlane Register/ACK/Runtime 接收已通过内存、Redis 8.2 和 gRPC 测试；Snapshot 下发、目标健康和 Gateway reserve 策略仍待完成 |
 | CAP-003 | pool.maxSize 包括 FETCHED/CHECKING/AVAILABLE/SUSPECT/DRAINING 与 pending expected | 3001-3533, 6642-6680 | `FetchBudget` 100 并发额度预占测试 |
 | CAP-004 | TTL safety margin 内禁止新分配 | 173-220, 6728-6741 | 时钟测试 |
 | CAP-005 | 多 Worker 不在热路径访问 Redis 计数 | 1403-1467 | Gateway 包依赖审计、Snapshot/Dispatch 测试 |

internal/controller/bootstrap/bootstrap.go

@@ -19,8 +19,10 @@ import (
 	"proxy-pool/internal/controller/pool"
 	"proxy-pool/internal/controller/provider"
 	controllerRuntime "proxy-pool/internal/controller/runtime"
+	"proxy-pool/internal/controller/worker"
 	"proxy-pool/internal/domain/activitypool"
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
 	"proxy-pool/internal/platform/credentials"
 	"proxy-pool/internal/platform/httpserver"
@@ -59,6 +61,7 @@ type ports struct {
 	coordinator      provider.Coordinator
 	credentials      credentials.Store
 	providerResults  provider.ResultRecorder
+	workerStore      workerruntime.ControlStore
 	close            func() error
 }
 
@@ -74,15 +77,29 @@ type runtimeFactory interface {
 	New(*config.Config, controllerRuntime.Dependencies, controllerRuntime.Options) (controllerRunner, error)
 }
 
+type workerRuntimeFactory interface {
+	New(config.ControlPlane, worker.Service) (controllerRunner, error)
+}
+
 func Run(ctx context.Context, options Options) error {
-	return run(ctx, options, &productionInfrastructure{
+	return runWithWorkerFactory(ctx, options, &productionInfrastructure{
 		holderID: options.HolderID, namespace: options.RedisNamespace,
-	}, productionRuntimeFactory{})
+	}, productionRuntimeFactory{}, productionWorkerRuntimeFactory{})
 }
 
 func run(ctx context.Context, options Options, infrastructure infrastructure, factory runtimeFactory) (resultErr error) {
+	return runWithWorkerFactory(ctx, options, infrastructure, factory, productionWorkerRuntimeFactory{})
+}
+
+func runWithWorkerFactory(
+	ctx context.Context,
+	options Options,
+	infrastructure infrastructure,
+	factory runtimeFactory,
+	workerFactory workerRuntimeFactory,
+) (resultErr error) {
 	if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" ||
-		nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) {
+		nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) || nilInterface(workerFactory) {
 		return ErrInvalidOptions
 	}
 	if err := ctx.Err(); err != nil {
@@ -200,7 +217,7 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa
 		dependencies.MetricsHandler = handler
 	}
 
-	runners := make([]lifecycle.Runner, 0, 2)
+	runners := make([]lifecycle.Runner, 0, 3)
 	if hasHTTPRuntime(loaded.Value) {
 		runner, err := factory.New(configurationStore.Current(), dependencies, controllerRuntime.Options{HTTP: options.HTTP})
 		if err != nil {
@@ -211,6 +228,29 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa
 		}
 		runners = append(runners, runner)
 	}
+	if loaded.Value.ControlPlane.Enabled {
+		if nilInterface(opened.workerStore) {
+			return errors.Join(ErrStartup, ErrInvalidOptions)
+		}
+		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,
+		})
+		if serviceErr != nil {
+			return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr)
+		}
+		runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service)
+		if runnerErr != nil {
+			return fmt.Errorf("%w: build Worker control server: %w", ErrStartup, runnerErr)
+		}
+		if nilInterface(runner) {
+			return errors.Join(ErrStartup, ErrInvalidOptions)
+		}
+		runners = append(runners, runner)
+	}
 	runners = append(runners, supervisor)
 	group, err := lifecycle.NewGroup(runners...)
 	if err != nil {
@@ -298,3 +338,9 @@ func (productionRuntimeFactory) New(
 ) (controllerRunner, error) {
 	return controllerRuntime.New(configuration, dependencies, options)
 }
+
+type productionWorkerRuntimeFactory struct{}
+
+func (productionWorkerRuntimeFactory) New(controlPlane config.ControlPlane, service worker.Service) (controllerRunner, error) {
+	return worker.NewServer(controlPlane, service, worker.DefaultServerOptions())
+}

internal/controller/bootstrap/bootstrap_integration_test.go

@@ -4,7 +4,9 @@ package bootstrap
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -22,7 +24,14 @@ import (
 	"proxy-pool/internal/controller/pool"
 	"proxy-pool/internal/controller/provider"
 	controllerRuntime "proxy-pool/internal/controller/runtime"
+	controllerWorker "proxy-pool/internal/controller/worker"
+	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/credentials"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
 )
 
 func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *testing.T) {
@@ -80,6 +89,95 @@ func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *tes
 	}
 }
 
+func TestProductionBootstrapServesWorkerControlPlane(t *testing.T) {
+	postgresURL := os.Getenv("PROXY_POOL_TEST_POSTGRES_URL")
+	redisURL := os.Getenv("PROXY_POOL_TEST_REDIS_URL")
+	if postgresURL == "" || redisURL == "" {
+		t.Skip("PROXY_POOL_TEST_POSTGRES_URL and PROXY_POOL_TEST_REDIS_URL are required")
+	}
+	providerServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		_, _ = writer.Write([]byte("http://192.0.2.10:8080"))
+	}))
+	defer providerServer.Close()
+	namespace := "controller-worker-it-" + strconv.FormatInt(time.Now().UnixNano(), 10)
+	source := strings.ReplaceAll(bootstrapTestConfig, "postgres://fixture", postgresURL)
+	source = strings.ReplaceAll(source, "redis://fixture", redisURL)
+	source = strings.ReplaceAll(source, "https://provider.invalid/proxies", providerServer.URL)
+	source += `
+controlPlane:
+  enabled: true
+  listen: 127.0.0.1:0
+  protocolVersion: 1
+  heartbeatInterval: 10s
+  sessionTTL: 30s
+  maxStaleAge: 10s
+  maxMessageBytes: 1048576
+  maxRuntimeCounters: 100
+  maxConcurrentStreams: 10
+  tls: {mode: disabled}
+`
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	workerFactory := &integrationWorkerRuntimeFactory{ready: make(chan struct{})}
+	result := make(chan error, 1)
+	go func() {
+		result <- runWithWorkerFactory(ctx, Options{
+			ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}},
+			Now: time.Now, FingerprintKey: bootstrapTestFingerprintKey,
+		}, &productionInfrastructure{namespace: namespace}, integrationBlockingRuntimeFactory{}, workerFactory)
+	}()
+	select {
+	case <-workerFactory.ready:
+	case <-time.After(5 * time.Second):
+		t.Fatal("worker control plane did not start")
+	}
+
+	connection, err := grpc.NewClient(workerFactory.address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(): %v", err)
+	}
+	defer connection.Close()
+	client := controlplanev1.NewWorkerControlPlaneClient(connection)
+	requestCtx, requestCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer requestCancel()
+	registration, err := client.RegisterWorker(requestCtx, &controlplanev1.RegisterWorkerRequest{
+		WorkerId: "worker-a", InstanceId: "instance-a", Zone: "zone-a", SupportedProtocolVersion: 1,
+	})
+	if err != nil {
+		t.Fatalf("RegisterWorker(): %v", err)
+	}
+	store := newIntegrationWorkerStore(t, redisURL, namespace)
+	checksum := sha256.Sum256([]byte("snapshot-1"))
+	reference := workerruntime.SnapshotReference{
+		WorkerID: "worker-a", Version: 1, OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum,
+	}
+	if err := store.RecordIssuedSnapshot(requestCtx, reference, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(): %v", err)
+	}
+	if _, err := client.AcknowledgeSnapshot(requestCtx, &controlplanev1.AcknowledgeSnapshotRequest{
+		WorkerId: "worker-a", SessionId: registration.GetSessionId(), Version: 1,
+		OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum[:], Applied: true,
+	}); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(): %v", err)
+	}
+	runtime, err := client.ReportRuntime(requestCtx, &controlplanev1.ReportRuntimeRequest{
+		WorkerId: "worker-a", SessionId: registration.GetSessionId(), SnapshotVersion: 1,
+		OwnershipEpoch: registration.GetOwnershipEpoch(), ReportSequence: 1, ObservedAt: timestamppb.Now(),
+	})
+	if err != nil || runtime.GetRequireFullSnapshot() || runtime.GetAcceptedOwnershipEpoch() != registration.GetOwnershipEpoch() {
+		t.Fatalf("ReportRuntime() = %+v, %v", runtime, err)
+	}
+	cancel()
+	select {
+	case err := <-result:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("run() error = %v, want context cancellation", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("controller did not stop")
+	}
+}
+
 type integrationRuntimeFactory struct {
 	status          admin.Status
 	readyStatus     int
@@ -206,3 +304,62 @@ type integrationRunner struct {
 }
 
 func (runner integrationRunner) Run(ctx context.Context) error { return runner.run(ctx) }
+
+type integrationBlockingRuntimeFactory struct{}
+
+func (integrationBlockingRuntimeFactory) New(
+	_ *config.Config,
+	_ controllerRuntime.Dependencies,
+	_ controllerRuntime.Options,
+) (controllerRunner, error) {
+	return integrationRunner{run: func(ctx context.Context) error {
+		<-ctx.Done()
+		return ctx.Err()
+	}}, nil
+}
+
+type integrationWorkerRuntimeFactory struct {
+	address string
+	ready   chan struct{}
+}
+
+func (factory *integrationWorkerRuntimeFactory) New(
+	controlPlane config.ControlPlane,
+	service controllerWorker.Service,
+) (controllerRunner, error) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return nil, err
+	}
+	server, err := controllerWorker.NewServer(controlPlane, service, controllerWorker.DefaultServerOptions())
+	if err != nil {
+		_ = listener.Close()
+		return nil, err
+	}
+	factory.address = listener.Addr().String()
+	close(factory.ready)
+	return integrationRunner{run: func(ctx context.Context) error { return server.Serve(ctx, listener) }}, nil
+}
+
+func newIntegrationWorkerStore(t *testing.T, redisURL, namespace string) *redisactivity.Adapter {
+	t.Helper()
+	options, err := redis.ParseURL(redisURL)
+	if err != nil {
+		t.Fatalf("redis.ParseURL(): %v", err)
+	}
+	client := redis.NewClient(options)
+	t.Cleanup(func() { _ = client.Close() })
+	credentialStore, err := credentials.NewMemoryStore(100)
+	if err != nil {
+		t.Fatalf("credentials.NewMemoryStore(): %v", err)
+	}
+	store, err := redisactivity.New(client, redisactivity.Options{
+		Namespace: namespace, Credentials: credentialStore, OperationTTL: redisOperationTTL,
+		MaxCandidateScan: redisMinimumScan, MaxRuntimeCounters: 100, MaxInventoryScan: 100,
+		CleanupLimit: redisCleanupLimit,
+	})
+	if err != nil {
+		t.Fatalf("redisactivity.New(): %v", err)
+	}
+	return store
+}

internal/controller/bootstrap/bootstrap_test.go

@@ -12,10 +12,12 @@ import (
 	"proxy-pool/internal/controller/pool"
 	"proxy-pool/internal/controller/provider"
 	controllerRuntime "proxy-pool/internal/controller/runtime"
+	controllerWorker "proxy-pool/internal/controller/worker"
 	"proxy-pool/internal/domain/activitypool"
 	"proxy-pool/internal/domain/adminstate"
 	extractionDomain "proxy-pool/internal/domain/extraction"
 	"proxy-pool/internal/domain/upstream"
+	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
 	"proxy-pool/internal/platform/credentials"
 )
@@ -170,6 +172,52 @@ func TestRunSupportsProviderOnlyConfigurationWithoutHTTPRuntime(t *testing.T) {
 	}
 }
 
+func TestRunRequiresWorkerControlStoreWhenControlPlaneEnabled(t *testing.T) {
+	source := bootstrapControlPlaneConfig()
+	credentialStore, err := credentials.NewMemoryStore(10)
+	if err != nil {
+		t.Fatalf("NewMemoryStore(): %v", err)
+	}
+	infrastructure := &stubInfrastructure{ports: ports{
+		activity: &stubActivityStore{}, coordinator: coordinatorStub{}, credentials: credentialStore, close: func() error { return nil },
+	}}
+	err = run(context.Background(), Options{
+		ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}}, Now: time.Now,
+	}, infrastructure, &recordingRuntimeFactory{})
+	if !errors.Is(err, ErrStartup) || !errors.Is(err, ErrInvalidOptions) {
+		t.Fatalf("run() error = %v, want startup invalid options", err)
+	}
+}
+
+func TestRunStartsWorkerControlPlaneWithoutHTTPRuntime(t *testing.T) {
+	store, err := workerruntime.NewMemoryStore(time.Now)
+	if err != nil {
+		t.Fatalf("NewMemoryStore(): %v", err)
+	}
+	credentialStore, err := credentials.NewMemoryStore(10)
+	if err != nil {
+		t.Fatalf("NewMemoryStore(): %v", err)
+	}
+	infrastructure := &stubInfrastructure{ports: ports{
+		activity: &stubActivityStore{}, workerStore: store, coordinator: coordinatorStub{}, credentials: credentialStore, close: func() error { return nil },
+	}}
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	workerFactory := &recordingWorkerRuntimeFactory{runner: runnerFunc(func(ctx context.Context) error {
+		<-ctx.Done()
+		return ctx.Err()
+	})}
+	err = runWithWorkerFactory(ctx, Options{
+		ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(bootstrapControlPlaneConfig())}}, Now: time.Now,
+	}, infrastructure, &recordingRuntimeFactory{}, workerFactory)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("run() error = %v, want context deadline exceeded", err)
+	}
+	if workerFactory.controlPlane.Listen != "127.0.0.1:0" || workerFactory.service == nil {
+		t.Fatalf("worker runtime factory = controlPlane:%+v service:%T", workerFactory.controlPlane, workerFactory.service)
+	}
+}
+
 func TestRunAdminDisableStopsActiveProviderRuntime(t *testing.T) {
 	state := adminstate.NewMemoryStore()
 	credentialStore, err := credentials.NewMemoryStore(200)
@@ -332,6 +380,22 @@ func (factory runtimeFactoryFunc) New(
 	return factory(configuration, dependencies, options)
 }
 
+type recordingWorkerRuntimeFactory struct {
+	controlPlane config.ControlPlane
+	service      controllerWorker.Service
+	runner       controllerRunner
+	err          error
+}
+
+func (factory *recordingWorkerRuntimeFactory) New(
+	controlPlane config.ControlPlane,
+	service controllerWorker.Service,
+) (controllerRunner, error) {
+	factory.controlPlane = controlPlane
+	factory.service = service
+	return factory.runner, factory.err
+}
+
 type readyStub struct{}
 
 func (readyStub) Ready(context.Context) error { return nil }
@@ -453,3 +517,22 @@ upstreams:
       urls: [https://example.invalid/health]
   provider-b: *upstream
 `
+
+func bootstrapControlPlaneConfig() string {
+	source := strings.ReplaceAll(bootstrapTestConfig, "distribution:\n  enabled: true", "distribution:\n  enabled: false")
+	source = strings.ReplaceAll(source, "admin:\n  enabled: true", "admin:\n  enabled: false")
+	source = strings.ReplaceAll(source, "metrics:\n  enabled: true", "metrics:\n  enabled: false")
+	return source + `
+controlPlane:
+  enabled: true
+  listen: 127.0.0.1:0
+  protocolVersion: 1
+  heartbeatInterval: 10s
+  sessionTTL: 30s
+  maxStaleAge: 10s
+  maxMessageBytes: 1048576
+  maxRuntimeCounters: 100
+  maxConcurrentStreams: 10
+  tls: {mode: disabled}
+`
+}

internal/controller/bootstrap/infrastructure.go

@@ -102,7 +102,7 @@ func (infrastructure *productionInfrastructure) Open(
 	}
 
 	providersEnabled := hasEnabledUpstream(configuration)
-	if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled {
+	if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled || configuration.ControlPlane.Enabled {
 		if strings.TrimSpace(configuration.Storage.RedisURL) == "" {
 			return ports{}, ErrRedisConfiguration
 		}
@@ -123,7 +123,7 @@ func (infrastructure *productionInfrastructure) Open(
 			Credentials:        credentialStore,
 			OperationTTL:       redisOperationTTL,
 			MaxCandidateScan:   candidateScan(configuration),
-			MaxRuntimeCounters: credentialCapacity(configuration),
+			MaxRuntimeCounters: runtimeCounterCapacity(configuration),
 			MaxInventoryScan:   maxInventoryScan(configuration),
 			CleanupLimit:       redisCleanupLimit,
 		})
@@ -131,6 +131,7 @@ func (infrastructure *productionInfrastructure) Open(
 			return ports{}, err
 		}
 		opened.activity = adapter
+		opened.workerStore = adapter
 		opened.readiness = redisReadiness{client: redisClient}
 		opened.credentials = credentialStore
 		if configuration.Distribution.Enabled {
@@ -203,7 +204,7 @@ func selectMetricsReadiness(
 	configuration *config.Config,
 	admin, activity platformMetrics.ReadinessChecker,
 ) platformMetrics.ReadinessChecker {
-	if configuration.Distribution.Enabled || hasEnabledUpstream(configuration) {
+	if configuration.Distribution.Enabled || configuration.ControlPlane.Enabled || hasEnabledUpstream(configuration) {
 		return activity
 	}
 	if configuration.Admin.Enabled {
@@ -275,6 +276,14 @@ func credentialCapacity(configuration *config.Config) int {
 	return capacity
 }
 
+func runtimeCounterCapacity(configuration *config.Config) int {
+	capacity := credentialCapacity(configuration)
+	if configuration != nil && configuration.ControlPlane.Enabled && configuration.ControlPlane.MaxRuntimeCounters > capacity {
+		return configuration.ControlPlane.MaxRuntimeCounters
+	}
+	return capacity
+}
+
 func providerCredentialCapacity(configuration *config.Config) int {
 	capacity := 0
 	maximum := int(^uint(0) >> 1)

internal/controller/bootstrap/infrastructure_test.go

@@ -50,6 +50,13 @@ func TestProductionInfrastructureRejectsInvalidStorageWithoutLeakingURLs(t *test
 	if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) {
 		t.Fatalf("Open(invalid Redis) error = %v", err)
 	}
+	_, err = (&productionInfrastructure{}).Open(context.Background(), &config.Config{
+		ControlPlane: config.ControlPlane{Enabled: true},
+		Storage:      config.Storage{RedisURL: "redis://user:" + redisSecret + "@%zz"},
+	})
+	if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) {
+		t.Fatalf("Open(control plane invalid Redis) error = %v", err)
+	}
 }
 
 func TestNewDistributionAdmitterPassesConfiguredLimits(t *testing.T) {
@@ -124,6 +131,19 @@ func TestSelectMetricsReadinessUsesAdminStoresWithoutDistribution(t *testing.T)
 	}
 }
 
+func TestSelectMetricsReadinessUsesRedisForControlPlane(t *testing.T) {
+	t.Parallel()
+	wantErr := errors.New("redis unavailable")
+	selected := selectMetricsReadiness(
+		&config.Config{ControlPlane: config.ControlPlane{Enabled: true}},
+		readinessFunc(func(context.Context) error { return nil }),
+		readinessFunc(func(context.Context) error { return wantErr }),
+	)
+	if err := selected.Ready(context.Background()); !errors.Is(err, wantErr) {
+		t.Fatalf("Ready() error = %v, want %v", err, wantErr)
+	}
+}
+
 type readinessFunc func(context.Context) error
 
 func (function readinessFunc) Ready(ctx context.Context) error { return function(ctx) }
@@ -143,6 +163,14 @@ func TestProductionRedisSizingUsesConfigurationBounds(t *testing.T) {
 	if got := credentialCapacity(configuration); got != 5_000 {
 		t.Fatalf("credentialCapacity() = %d, want 5000", got)
 	}
+	configuration.ControlPlane = config.ControlPlane{Enabled: true, MaxRuntimeCounters: 100_000}
+	if got := runtimeCounterCapacity(configuration); got != 100_000 {
+		t.Fatalf("runtimeCounterCapacity(control plane) = %d, want 100000", got)
+	}
+	configuration.ControlPlane.Enabled = false
+	if got := runtimeCounterCapacity(configuration); got != 5_000 {
+		t.Fatalf("runtimeCounterCapacity(disabled control plane) = %d, want 5000", got)
+	}
 	if got := providerCredentialCapacity(configuration); got != 158_000 {
 		t.Fatalf("providerCredentialCapacity() = %d, want 158000", got)
 	}

internal/controller/worker/grpc_handler.go

@@ -0,0 +1,136 @@
+package worker
+
+import (
+	"context"
+	"errors"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/domain/workerruntime"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+type IdentityAuthorizer interface {
+	Authorize(context.Context, string) error
+}
+
+type GRPCHandler struct {
+	controlplanev1.UnimplementedWorkerControlPlaneServer
+	service  Service
+	identity IdentityAuthorizer
+}
+
+func NewGRPCHandler(service Service, identity IdentityAuthorizer) *GRPCHandler {
+	return &GRPCHandler{service: service, identity: identity}
+}
+
+func (handler *GRPCHandler) RegisterWorker(ctx context.Context, request *controlplanev1.RegisterWorkerRequest) (*controlplanev1.RegisterWorkerResponse, error) {
+	if request == nil || handler == nil || handler.service == nil || handler.identity == nil {
+		return nil, grpcError(ErrInvalidCommand)
+	}
+	if err := handler.authorize(ctx, request.GetWorkerId()); err != nil {
+		return nil, err
+	}
+	registration, err := handler.service.Register(ctx, RegisterCommand{
+		WorkerID: request.GetWorkerId(), InstanceID: request.GetInstanceId(), Zone: request.GetZone(),
+		ProtocolVersion: request.GetSupportedProtocolVersion(), Labels: cloneLabels(request.GetLabels()),
+	})
+	if err != nil {
+		return nil, grpcError(err)
+	}
+	return &controlplanev1.RegisterWorkerResponse{
+		WorkerId: registration.WorkerID, SessionId: registration.SessionID, OwnershipEpoch: registration.OwnershipEpoch,
+		HeartbeatInterval: durationpb.New(registration.HeartbeatInterval), MaxStaleAge: durationpb.New(registration.MaxStaleAge),
+	}, nil
+}
+
+func (handler *GRPCHandler) AcknowledgeSnapshot(ctx context.Context, request *controlplanev1.AcknowledgeSnapshotRequest) (*emptypb.Empty, error) {
+	if request == nil || handler == nil || handler.service == nil || handler.identity == nil {
+		return nil, grpcError(ErrInvalidCommand)
+	}
+	if err := handler.authorize(ctx, request.GetWorkerId()); err != nil {
+		return nil, err
+	}
+	err := handler.service.Acknowledge(ctx, SnapshotAcknowledgement{
+		WorkerID: request.GetWorkerId(), SessionID: request.GetSessionId(), Version: request.GetVersion(),
+		OwnershipEpoch: request.GetOwnershipEpoch(), Checksum: append([]byte(nil), request.GetChecksum()...),
+		Applied: request.GetApplied(), ErrorCode: request.GetErrorCode(), ErrorMessage: request.GetErrorMessage(),
+	})
+	if err != nil {
+		return nil, grpcError(err)
+	}
+	return &emptypb.Empty{}, nil
+}
+
+func (handler *GRPCHandler) ReportRuntime(ctx context.Context, request *controlplanev1.ReportRuntimeRequest) (*controlplanev1.ReportRuntimeResponse, error) {
+	if request == nil || handler == nil || handler.service == nil || handler.identity == nil || request.GetObservedAt() == nil || request.GetObservedAt().CheckValid() != nil {
+		return nil, grpcError(ErrInvalidCommand)
+	}
+	if err := handler.authorize(ctx, request.GetWorkerId()); err != nil {
+		return nil, err
+	}
+	counters := make([]workerruntime.Counter, len(request.GetCounters()))
+	for index, counter := range request.GetCounters() {
+		if counter == nil {
+			return nil, grpcError(ErrInvalidCommand)
+		}
+		counters[index] = workerruntime.Counter{
+			ProxyID: counter.GetProxyId(), Active: int64(counter.GetActive()), Reserved: int64(counter.GetReserved()), Draining: counter.GetDraining(),
+		}
+	}
+	decision, err := handler.service.ReportRuntime(ctx, workerruntime.Report{
+		WorkerID: request.GetWorkerId(), SessionID: request.GetSessionId(), Sequence: request.GetReportSequence(),
+		SnapshotVersion: request.GetSnapshotVersion(), OwnershipEpoch: request.GetOwnershipEpoch(),
+		ObservedAt: request.GetObservedAt().AsTime(), Counters: counters,
+	})
+	if err != nil {
+		return nil, grpcError(err)
+	}
+	return &controlplanev1.ReportRuntimeResponse{
+		AcceptedOwnershipEpoch: decision.AcceptedOwnershipEpoch, RequireFullSnapshot: decision.RequireFullSnapshot,
+	}, nil
+}
+
+func (handler *GRPCHandler) authorize(ctx context.Context, workerID string) error {
+	if err := handler.identity.Authorize(ctx, workerID); err != nil {
+		return status.Error(codes.PermissionDenied, "worker identity is not authorized")
+	}
+	return nil
+}
+
+func grpcError(err error) error {
+	switch {
+	case errors.Is(err, context.Canceled):
+		return status.Error(codes.Canceled, "worker control request canceled")
+	case errors.Is(err, context.DeadlineExceeded):
+		return status.Error(codes.DeadlineExceeded, "worker control request deadline exceeded")
+	case errors.Is(err, ErrInvalidCommand), errors.Is(err, workerruntime.ErrInvalidReport),
+		errors.Is(err, workerruntime.ErrInvalidAcknowledgement), errors.Is(err, workerruntime.ErrInvalidSnapshotReference):
+		return status.Error(codes.InvalidArgument, "invalid worker control request")
+	case errors.Is(err, ErrProtocolVersion):
+		return status.Error(codes.FailedPrecondition, "unsupported worker protocol version")
+	case errors.Is(err, workerruntime.ErrStaleSession):
+		return status.Error(codes.FailedPrecondition, "worker session is stale")
+	case errors.Is(err, workerruntime.ErrSnapshotMismatch):
+		return status.Error(codes.FailedPrecondition, "worker snapshot does not match issued snapshot")
+	case errors.Is(err, workerruntime.ErrStaleAcknowledgement):
+		return status.Error(codes.Aborted, "worker snapshot acknowledgement is stale")
+	case errors.Is(err, workerruntime.ErrStaleReport):
+		return status.Error(codes.Aborted, "worker runtime sequence is stale")
+	case errors.Is(err, workerruntime.ErrConflictingReport):
+		return status.Error(codes.AlreadyExists, "worker runtime sequence conflicts")
+	default:
+		return status.Error(codes.Unavailable, "worker control plane unavailable")
+	}
+}
+
+func cloneLabels(labels map[string]string) map[string]string {
+	result := make(map[string]string, len(labels))
+	for key, value := range labels {
+		result[key] = value
+	}
+	return result
+}

internal/controller/worker/grpc_handler_test.go

@@ -0,0 +1,103 @@
+package worker
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/domain/workerruntime"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func TestGRPCHandlerMapsWorkerRequests(t *testing.T) {
+	service := &grpcServiceStub{registration: Registration{WorkerID: "worker-a", SessionID: "session-a", OwnershipEpoch: 9, HeartbeatInterval: time.Second, MaxStaleAge: 2 * time.Second}}
+	client, cleanup := grpcWorkerClient(t, service, allowIdentity{})
+	defer cleanup()
+	registered, err := client.RegisterWorker(context.Background(), &controlplanev1.RegisterWorkerRequest{
+		WorkerId: "worker-a", InstanceId: "instance-a", Zone: "zone-a", SupportedProtocolVersion: 1,
+	})
+	if err != nil || registered.GetSessionId() != "session-a" || registered.GetHeartbeatInterval().AsDuration() != time.Second {
+		t.Fatalf("RegisterWorker() = %+v, %v", registered, err)
+	}
+	if err := service.acknowledgeErr; err != nil {
+		t.Fatal(err)
+	}
+	_, err = client.AcknowledgeSnapshot(context.Background(), &controlplanev1.AcknowledgeSnapshotRequest{
+		WorkerId: "worker-a", SessionId: "session-a", Version: 7, OwnershipEpoch: 9, Checksum: make([]byte, 32),
+	})
+	if err != nil || service.acknowledgement.Version != 7 {
+		t.Fatalf("AcknowledgeSnapshot() error = %v; command=%+v", err, service.acknowledgement)
+	}
+	response, err := client.ReportRuntime(context.Background(), &controlplanev1.ReportRuntimeRequest{
+		WorkerId: "worker-a", SessionId: "session-a", SnapshotVersion: 7, OwnershipEpoch: 9, ReportSequence: 1,
+		ObservedAt: timestamppb.New(time.Now()), Counters: []*controlplanev1.ProxyRuntime{{ProxyId: "proxy-a", Active: 2, Reserved: 1}},
+	})
+	if err != nil || response.GetAcceptedOwnershipEpoch() != 9 || service.report.Counters[0].Active != 2 {
+		t.Fatalf("ReportRuntime() = %+v, %v; report=%+v", response, err, service.report)
+	}
+}
+
+func TestGRPCHandlerMapsErrorsAndLeavesStreamsUnimplemented(t *testing.T) {
+	service := &grpcServiceStub{registerErr: ErrProtocolVersion}
+	client, cleanup := grpcWorkerClient(t, service, allowIdentity{})
+	defer cleanup()
+	_, err := client.RegisterWorker(context.Background(), &controlplanev1.RegisterWorkerRequest{WorkerId: "worker-a"})
+	if status.Code(err) != codes.FailedPrecondition {
+		t.Fatalf("RegisterWorker() code = %s, want FailedPrecondition", status.Code(err))
+	}
+	stream, err := client.WatchSnapshots(context.Background(), &controlplanev1.WatchSnapshotsRequest{})
+	_, streamErr := stream.Recv()
+	if err != nil || status.Code(streamErr) != codes.Unimplemented {
+		t.Fatalf("WatchSnapshots() = %v, %v", err, streamErr)
+	}
+	outcomes, err := client.ReportOutcomes(context.Background())
+	_, outcomesErr := outcomes.CloseAndRecv()
+	if err != nil || status.Code(outcomesErr) != codes.Unimplemented {
+		t.Fatalf("ReportOutcomes() = %v, %v", err, outcomesErr)
+	}
+}
+
+type grpcServiceStub struct {
+	registration    Registration
+	registerErr     error
+	acknowledgement SnapshotAcknowledgement
+	acknowledgeErr  error
+	report          workerruntime.Report
+	reportErr       error
+}
+
+func (stub *grpcServiceStub) Register(context.Context, RegisterCommand) (Registration, error) {
+	return stub.registration, stub.registerErr
+}
+func (stub *grpcServiceStub) Acknowledge(_ context.Context, acknowledgement SnapshotAcknowledgement) error {
+	stub.acknowledgement = acknowledgement
+	return stub.acknowledgeErr
+}
+func (stub *grpcServiceStub) ReportRuntime(_ context.Context, report workerruntime.Report) (RuntimeDecision, error) {
+	stub.report = report
+	return RuntimeDecision{AcceptedOwnershipEpoch: 9}, stub.reportErr
+}
+
+type allowIdentity struct{}
+
+func (allowIdentity) Authorize(context.Context, string) error { return nil }
+
+func grpcWorkerClient(t *testing.T, service Service, identity IdentityAuthorizer) (controlplanev1.WorkerControlPlaneClient, func()) {
+	t.Helper()
+	listener := bufconn.Listen(1 << 20)
+	server := grpc.NewServer()
+	controlplanev1.RegisterWorkerControlPlaneServer(server, NewGRPCHandler(service, identity))
+	go func() { _ = server.Serve(listener) }()
+	connection, err := grpc.NewClient("passthrough:///bufnet", grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), grpc.WithInsecure())
+	if err != nil {
+		t.Fatalf("grpc.NewClient(): %v", err)
+	}
+	return controlplanev1.NewWorkerControlPlaneClient(connection), func() { _ = connection.Close(); server.Stop(); _ = listener.Close() }
+}

internal/controller/worker/identity.go

@@ -0,0 +1,72 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/peer"
+)
+
+var ErrUnauthorizedIdentity = errors.New("worker identity is not authorized")
+
+// AllowLoopbackIdentity is used only by the validated loopback plaintext mode.
+// It deliberately does not inspect transport credentials because that mode has no TLS peer.
+type AllowLoopbackIdentity struct{}
+
+func (AllowLoopbackIdentity) Authorize(context.Context, string) error {
+	return nil
+}
+
+type SPIFFEIdentityAuthorizer struct {
+	trustDomain string
+	environment string
+}
+
+func NewSPIFFEIdentityAuthorizer(trustDomain, environment string) (*SPIFFEIdentityAuthorizer, error) {
+	if strings.TrimSpace(trustDomain) == "" || strings.TrimSpace(environment) == "" {
+		return nil, fmt.Errorf("spiffe identity authorizer: trust domain and environment are required")
+	}
+	return &SPIFFEIdentityAuthorizer{trustDomain: trustDomain, environment: environment}, nil
+}
+
+func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, workerID string) error {
+	if authorizer == nil || workerID == "" {
+		return ErrUnauthorizedIdentity
+	}
+	peerInfo, ok := peer.FromContext(ctx)
+	if !ok || peerInfo.AuthInfo == nil {
+		return ErrUnauthorizedIdentity
+	}
+	tlsInfo, ok := peerInfo.AuthInfo.(credentials.TLSInfo)
+	if !ok || len(tlsInfo.State.VerifiedChains) == 0 {
+		return ErrUnauthorizedIdentity
+	}
+
+	expectedPath := "/" + authorizer.environment + "/worker/" + workerID
+	workerIdentityCount := 0
+	for _, chain := range tlsInfo.State.VerifiedChains {
+		if len(chain) == 0 || chain[0] == nil {
+			continue
+		}
+		for _, uri := range chain[0].URIs {
+			if uri == nil || uri.Scheme != "spiffe" || uri.Host != authorizer.trustDomain {
+				continue
+			}
+			segments := strings.Split(strings.Trim(uri.Path, "/"), "/")
+			if len(segments) != 3 || segments[1] != "worker" {
+				continue
+			}
+			workerIdentityCount++
+			if workerIdentityCount > 1 || uri.RawQuery != "" || uri.Fragment != "" || uri.Path != expectedPath {
+				return ErrUnauthorizedIdentity
+			}
+		}
+	}
+	if workerIdentityCount != 1 {
+		return ErrUnauthorizedIdentity
+	}
+	return nil
+}

internal/controller/worker/identity_test.go

@@ -0,0 +1,84 @@
+package worker
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"net/url"
+	"testing"
+
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/peer"
+)
+
+func TestSPIFFEIdentityAuthorizer(t *testing.T) {
+	authorizer, err := NewSPIFFEIdentityAuthorizer("proxy.example", "prod")
+	if err != nil {
+		t.Fatalf("NewSPIFFEIdentityAuthorizer(): %v", err)
+	}
+
+	tests := []struct {
+		name      string
+		workerID  string
+		uri       string
+		withTLS   bool
+		wantError bool
+	}{
+		{name: "matching worker uri", workerID: "worker-a", uri: "spiffe://proxy.example/prod/worker/worker-a", withTLS: true},
+		{name: "different worker", workerID: "worker-a", uri: "spiffe://proxy.example/prod/worker/worker-b", withTLS: true, wantError: true},
+		{name: "different environment", workerID: "worker-a", uri: "spiffe://proxy.example/staging/worker/worker-a", withTLS: true, wantError: true},
+		{name: "different trust domain", workerID: "worker-a", uri: "spiffe://other.example/prod/worker/worker-a", withTLS: true, wantError: true},
+		{name: "missing peer tls", workerID: "worker-a", wantError: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ctx := context.Background()
+			if test.withTLS {
+				ctx = tlsPeerContext(t, test.uri)
+			}
+			err := authorizer.Authorize(ctx, test.workerID)
+			if (err != nil) != test.wantError {
+				t.Fatalf("Authorize() error = %v, wantError %v", err, test.wantError)
+			}
+		})
+	}
+}
+
+func TestSPIFFEIdentityAuthorizerRejectsMultipleWorkerURIs(t *testing.T) {
+	authorizer, err := NewSPIFFEIdentityAuthorizer("proxy.example", "prod")
+	if err != nil {
+		t.Fatalf("NewSPIFFEIdentityAuthorizer(): %v", err)
+	}
+	first, err := url.Parse("spiffe://proxy.example/prod/worker/worker-a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := url.Parse("spiffe://proxy.example/prod/worker/worker-b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := peer.NewContext(context.Background(), &peer.Peer{AuthInfo: credentials.TLSInfo{
+		State: tls.ConnectionState{VerifiedChains: [][]*x509.Certificate{{{URIs: []*url.URL{first, second}}}}},
+	}})
+	if err := authorizer.Authorize(ctx, "worker-a"); err == nil {
+		t.Fatal("Authorize() error = nil, want rejection for multiple worker identities")
+	}
+}
+
+func TestAllowLoopbackIdentity(t *testing.T) {
+	if err := (AllowLoopbackIdentity{}).Authorize(context.Background(), "worker-a"); err != nil {
+		t.Fatalf("Authorize(): %v", err)
+	}
+}
+
+func tlsPeerContext(t *testing.T, identityURI string) context.Context {
+	t.Helper()
+	uri, err := url.Parse(identityURI)
+	if err != nil {
+		t.Fatalf("url.Parse(): %v", err)
+	}
+	return peer.NewContext(context.Background(), &peer.Peer{AuthInfo: credentials.TLSInfo{
+		State: tls.ConnectionState{VerifiedChains: [][]*x509.Certificate{{{URIs: []*url.URL{uri}}}}},
+	}})
+}

internal/controller/worker/server.go

@@ -0,0 +1,169 @@
+package worker
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"strings"
+	"time"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/config"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
+)
+
+var ErrInvalidServer = errors.New("invalid worker control server configuration")
+
+type ServerOptions struct {
+	ShutdownTimeout time.Duration
+}
+
+func DefaultServerOptions() ServerOptions {
+	return ServerOptions{ShutdownTimeout: 15 * time.Second}
+}
+
+type Server struct {
+	listen          string
+	grpcServer      *grpc.Server
+	shutdownTimeout time.Duration
+}
+
+func NewServer(controlPlane config.ControlPlane, service Service, options ServerOptions) (*Server, error) {
+	if service == nil || !controlPlane.Enabled || !validServerConfig(controlPlane) {
+		return nil, ErrInvalidServer
+	}
+	if options.ShutdownTimeout < 0 {
+		return nil, fmt.Errorf("%w: shutdown timeout must not be negative", ErrInvalidServer)
+	}
+	if options.ShutdownTimeout == 0 {
+		options = DefaultServerOptions()
+	}
+
+	identity, serverOptions, err := serverTransportOptions(controlPlane)
+	if err != nil {
+		return nil, err
+	}
+	serverOptions = append(serverOptions,
+		grpc.MaxRecvMsgSize(controlPlane.MaxMessageBytes),
+		grpc.MaxSendMsgSize(controlPlane.MaxMessageBytes),
+		grpc.MaxConcurrentStreams(controlPlane.MaxConcurrentStreams),
+		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
+			MinTime:             10 * time.Second,
+			PermitWithoutStream: false,
+		}),
+	)
+	grpcServer := grpc.NewServer(serverOptions...)
+	controlplanev1.RegisterWorkerControlPlaneServer(grpcServer, NewGRPCHandler(service, identity))
+	return &Server{listen: controlPlane.Listen, grpcServer: grpcServer, shutdownTimeout: options.ShutdownTimeout}, nil
+}
+
+func (server *Server) Run(ctx context.Context) error {
+	if server == nil || server.grpcServer == nil || server.listen == "" {
+		return ErrInvalidServer
+	}
+	listener, err := net.Listen("tcp", server.listen)
+	if err != nil {
+		return fmt.Errorf("listen worker control plane: %w", err)
+	}
+	return server.Serve(ctx, listener)
+}
+
+func (server *Server) Serve(ctx context.Context, listener net.Listener) error {
+	if server == nil || server.grpcServer == nil || listener == nil || ctx == nil {
+		return ErrInvalidServer
+	}
+	completed := make(chan struct{})
+	go func() {
+		select {
+		case <-ctx.Done():
+			server.gracefulStop()
+		case <-completed:
+		}
+	}()
+
+	err := server.grpcServer.Serve(listener)
+	close(completed)
+	if ctx.Err() != nil || errors.Is(err, grpc.ErrServerStopped) {
+		return nil
+	}
+	return err
+}
+
+func (server *Server) gracefulStop() {
+	stopped := make(chan struct{})
+	go func() {
+		server.grpcServer.GracefulStop()
+		close(stopped)
+	}()
+	timer := time.NewTimer(server.shutdownTimeout)
+	defer timer.Stop()
+	select {
+	case <-stopped:
+	case <-timer.C:
+		server.grpcServer.Stop()
+		<-stopped
+	}
+}
+
+func serverTransportOptions(controlPlane config.ControlPlane) (IdentityAuthorizer, []grpc.ServerOption, error) {
+	switch controlPlane.TLS.Mode {
+	case "disabled":
+		if !loopbackListen(controlPlane.Listen) {
+			return nil, nil, fmt.Errorf("%w: plaintext listener must be loopback", ErrInvalidServer)
+		}
+		return AllowLoopbackIdentity{}, nil, nil
+	case "mtls":
+		identity, err := NewSPIFFEIdentityAuthorizer(controlPlane.TLS.TrustDomain, controlPlane.TLS.Environment)
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: %v", ErrInvalidServer, err)
+		}
+		certificate, err := tls.LoadX509KeyPair(controlPlane.TLS.CertFile, controlPlane.TLS.KeyFile)
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: load server certificate: %v", ErrInvalidServer, err)
+		}
+		caPEM, err := os.ReadFile(controlPlane.TLS.ClientCAFile)
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: read client ca: %v", ErrInvalidServer, err)
+		}
+		clientCAs := x509.NewCertPool()
+		if !clientCAs.AppendCertsFromPEM(caPEM) {
+			return nil, nil, fmt.Errorf("%w: parse client ca", ErrInvalidServer)
+		}
+		transport := credentials.NewTLS(&tls.Config{
+			MinVersion:   tls.VersionTLS13,
+			Certificates: []tls.Certificate{certificate},
+			ClientAuth:   tls.RequireAndVerifyClientCert,
+			ClientCAs:    clientCAs,
+		})
+		return identity, []grpc.ServerOption{grpc.Creds(transport)}, nil
+	default:
+		return nil, nil, fmt.Errorf("%w: unsupported tls mode", ErrInvalidServer)
+	}
+}
+
+func validServerConfig(controlPlane config.ControlPlane) bool {
+	return controlPlane.Listen != "" && controlPlane.ProtocolVersion == 1 &&
+		controlPlane.HeartbeatInterval.Value() > 0 && controlPlane.SessionTTL.Value() > 0 &&
+		controlPlane.MaxStaleAge.Value() > 0 && controlPlane.MaxMessageBytes > 0 &&
+		controlPlane.MaxRuntimeCounters > 0 && controlPlane.MaxConcurrentStreams > 0
+}
+
+func loopbackListen(listen string) bool {
+	host, _, err := net.SplitHostPort(listen)
+	if err != nil {
+		return false
+	}
+	host = strings.Trim(host, "[]")
+	if strings.EqualFold(host, "localhost") {
+		return true
+	}
+	ip := net.ParseIP(host)
+	return ip != nil && ip.IsLoopback()
+}

internal/controller/worker/server_test.go

@@ -0,0 +1,86 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/config"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func TestNewServerRejectsInvalidOptions(t *testing.T) {
+	controlPlane := validServerControlPlane()
+	service := &grpcServiceStub{}
+	if _, err := NewServer(controlPlane, nil, DefaultServerOptions()); !errors.Is(err, ErrInvalidServer) {
+		t.Fatalf("NewServer(nil service) error = %v, want ErrInvalidServer", err)
+	}
+	if _, err := NewServer(controlPlane, service, ServerOptions{ShutdownTimeout: -time.Second}); !errors.Is(err, ErrInvalidServer) {
+		t.Fatalf("NewServer(negative shutdown timeout) error = %v, want ErrInvalidServer", err)
+	}
+	controlPlane.Enabled = false
+	if _, err := NewServer(controlPlane, service, DefaultServerOptions()); !errors.Is(err, ErrInvalidServer) {
+		t.Fatalf("NewServer(disabled) error = %v, want ErrInvalidServer", err)
+	}
+}
+
+func TestServerServesAndStopsOnContextCancellation(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("net.Listen(): %v", err)
+	}
+	controlPlane := validServerControlPlane()
+	controlPlane.Listen = listener.Addr().String()
+	service := &grpcServiceStub{registration: Registration{WorkerID: "worker-a", SessionID: "session-a", OwnershipEpoch: 5, HeartbeatInterval: time.Second, MaxStaleAge: 2 * time.Second}}
+	server, err := NewServer(controlPlane, service, ServerOptions{ShutdownTimeout: time.Second})
+	if err != nil {
+		t.Fatalf("NewServer(): %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	result := make(chan error, 1)
+	go func() { result <- server.Serve(ctx, listener) }()
+
+	connection, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(): %v", err)
+	}
+	client := controlplanev1.NewWorkerControlPlaneClient(connection)
+	requestCtx, requestCancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer requestCancel()
+	response, err := client.RegisterWorker(requestCtx, &controlplanev1.RegisterWorkerRequest{WorkerId: "worker-a", SupportedProtocolVersion: 1})
+	if err != nil || response.GetSessionId() != "session-a" {
+		t.Fatalf("RegisterWorker() = %+v, %v", response, err)
+	}
+	if err := connection.Close(); err != nil {
+		t.Fatalf("connection.Close(): %v", err)
+	}
+	cancel()
+	select {
+	case err := <-result:
+		if err != nil {
+			t.Fatalf("Serve() error = %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("Serve() did not stop after context cancellation")
+	}
+}
+
+func validServerControlPlane() config.ControlPlane {
+	return config.ControlPlane{
+		Enabled:              true,
+		Listen:               "127.0.0.1:8443",
+		ProtocolVersion:      1,
+		HeartbeatInterval:    config.Duration(10 * time.Second),
+		SessionTTL:           config.Duration(30 * time.Second),
+		MaxStaleAge:          config.Duration(10 * time.Second),
+		MaxMessageBytes:      1 << 20,
+		MaxRuntimeCounters:   100,
+		MaxConcurrentStreams: 10,
+		TLS:                  config.ControlPlaneTLS{Mode: "disabled"},
+	}
+}

internal/gateway/controlplane/reporter.go

@@ -0,0 +1,177 @@
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"sync"
+	"time"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/domain/workerruntime"
+	"proxy-pool/internal/gateway/snapshot"
+
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+var (
+	ErrInvalidOptions       = errors.New("invalid gateway control plane reporter options")
+	ErrNotRegistered        = errors.New("gateway worker session is not registered")
+	ErrFullSnapshotRequired = errors.New("controller requires a full snapshot")
+	ErrOwnershipChanged     = errors.New("controller accepted a different ownership epoch")
+)
+
+type Client interface {
+	RegisterWorker(context.Context, *controlplanev1.RegisterWorkerRequest) (*controlplanev1.RegisterWorkerResponse, error)
+	ReportRuntime(context.Context, *controlplanev1.ReportRuntimeRequest) (*controlplanev1.ReportRuntimeResponse, error)
+}
+
+type RuntimeSource interface {
+	RuntimeReport(string, uint64, time.Time) (workerruntime.Report, error)
+}
+
+type Options struct {
+	WorkerID        string
+	InstanceID      string
+	Zone            string
+	ProtocolVersion uint32
+	Labels          map[string]string
+	Now             func() time.Time
+}
+
+type Registration struct {
+	SessionID         string
+	OwnershipEpoch    uint64
+	HeartbeatInterval time.Duration
+	MaxStaleAge       time.Duration
+}
+
+type RuntimeReporter struct {
+	client    Client
+	snapshots RuntimeSource
+	options   Options
+
+	mu       sync.Mutex
+	session  Registration
+	sequence uint64
+}
+
+func NewRuntimeReporter(client Client, snapshots RuntimeSource, options Options) (*RuntimeReporter, error) {
+	if client == nil || snapshots == nil || options.ProtocolVersion == 0 || options.Now == nil ||
+		!workerruntime.ValidIdentifier(options.WorkerID) || !workerruntime.ValidIdentifier(options.InstanceID) ||
+		!workerruntime.ValidIdentifier(options.Zone) {
+		return nil, ErrInvalidOptions
+	}
+	if _, err := workerruntime.NormalizeLabels(options.Labels); err != nil {
+		return nil, errors.Join(ErrInvalidOptions, err)
+	}
+	options.Labels = cloneLabels(options.Labels)
+	return &RuntimeReporter{client: client, snapshots: snapshots, options: options}, nil
+}
+
+func (reporter *RuntimeReporter) Register(ctx context.Context) (Registration, error) {
+	if reporter == nil || ctx == nil {
+		return Registration{}, ErrInvalidOptions
+	}
+	if err := ctx.Err(); err != nil {
+		return Registration{}, err
+	}
+	response, err := reporter.client.RegisterWorker(ctx, &controlplanev1.RegisterWorkerRequest{
+		WorkerId: reporter.options.WorkerID, InstanceId: reporter.options.InstanceID, Zone: reporter.options.Zone,
+		SupportedProtocolVersion: reporter.options.ProtocolVersion, Labels: cloneLabels(reporter.options.Labels),
+	})
+	if err != nil {
+		return Registration{}, err
+	}
+	registration, err := validateRegistration(reporter.options.WorkerID, response)
+	if err != nil {
+		return Registration{}, err
+	}
+	reporter.mu.Lock()
+	reporter.session = registration
+	reporter.sequence = 0
+	reporter.mu.Unlock()
+	return registration, nil
+}
+
+func (reporter *RuntimeReporter) Report(ctx context.Context) error {
+	if reporter == nil || ctx == nil {
+		return ErrInvalidOptions
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	reporter.mu.Lock()
+	defer reporter.mu.Unlock()
+	if reporter.session.SessionID == "" {
+		return ErrNotRegistered
+	}
+	sequence := reporter.sequence + 1
+	report, err := reporter.snapshots.RuntimeReport(reporter.session.SessionID, sequence, reporter.options.Now().UTC())
+	if err != nil {
+		return fmt.Errorf("build gateway runtime report: %w", err)
+	}
+	request, err := runtimeRequest(report)
+	if err != nil {
+		return err
+	}
+	response, err := reporter.client.ReportRuntime(ctx, request)
+	if err != nil {
+		return err
+	}
+	if response == nil {
+		return ErrOwnershipChanged
+	}
+	if response.GetRequireFullSnapshot() {
+		reporter.sequence = sequence
+		return ErrFullSnapshotRequired
+	}
+	if response.GetAcceptedOwnershipEpoch() != reporter.session.OwnershipEpoch {
+		return ErrOwnershipChanged
+	}
+	reporter.sequence = sequence
+	return nil
+}
+
+func validateRegistration(workerID string, response *controlplanev1.RegisterWorkerResponse) (Registration, error) {
+	if response == nil || response.GetWorkerId() != workerID || !workerruntime.ValidIdentifier(response.GetSessionId()) ||
+		response.GetOwnershipEpoch() == 0 || response.GetHeartbeatInterval() == nil || response.GetMaxStaleAge() == nil ||
+		response.GetHeartbeatInterval().AsDuration() <= 0 || response.GetMaxStaleAge().AsDuration() <= 0 {
+		return Registration{}, ErrInvalidOptions
+	}
+	return Registration{
+		SessionID: response.GetSessionId(), OwnershipEpoch: response.GetOwnershipEpoch(),
+		HeartbeatInterval: response.GetHeartbeatInterval().AsDuration(), MaxStaleAge: response.GetMaxStaleAge().AsDuration(),
+	}, nil
+}
+
+func runtimeRequest(report workerruntime.Report) (*controlplanev1.ReportRuntimeRequest, error) {
+	counters := make([]*controlplanev1.ProxyRuntime, len(report.Counters))
+	for index, counter := range report.Counters {
+		if counter.Active < 0 || counter.Reserved < 0 || counter.Active > math.MaxUint32 || counter.Reserved > math.MaxUint32 {
+			return nil, ErrInvalidOptions
+		}
+		counters[index] = &controlplanev1.ProxyRuntime{
+			ProxyId: counter.ProxyID, Active: uint32(counter.Active), Reserved: uint32(counter.Reserved), Draining: counter.Draining,
+		}
+	}
+	return &controlplanev1.ReportRuntimeRequest{
+		WorkerId: report.WorkerID, SessionId: report.SessionID, SnapshotVersion: report.SnapshotVersion,
+		OwnershipEpoch: report.OwnershipEpoch, ReportSequence: report.Sequence, Counters: counters,
+		ObservedAt: timestamppb.New(report.ObservedAt),
+	}, nil
+}
+
+func cloneLabels(source map[string]string) map[string]string {
+	if source == nil {
+		return nil
+	}
+	result := make(map[string]string, len(source))
+	for key, value := range source {
+		result[key] = value
+	}
+	return result
+}
+
+var _ RuntimeSource = (*snapshot.Store)(nil)

internal/gateway/controlplane/reporter_test.go

@@ -0,0 +1,120 @@
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
+	"proxy-pool/internal/gateway/snapshot"
+
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+func TestRuntimeReporterRegistersAndReportsSnapshotCounters(t *testing.T) {
+	store := snapshot.NewStore("cluster-a", "worker-a")
+	applySnapshot(t, store)
+	client := &clientStub{registration: &controlplanev1.RegisterWorkerResponse{
+		WorkerId: "worker-a", SessionId: "session-a", OwnershipEpoch: 7,
+		HeartbeatInterval: durationpb.New(10 * time.Second), MaxStaleAge: durationpb.New(30 * time.Second),
+	}}
+	now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
+	reporter, err := NewRuntimeReporter(client, store, Options{
+		WorkerID: "worker-a", InstanceID: "instance-a", Zone: "zone-a", ProtocolVersion: 1,
+		Now: func() time.Time { return now },
+	})
+	if err != nil {
+		t.Fatalf("NewRuntimeReporter(): %v", err)
+	}
+	registration, err := reporter.Register(context.Background())
+	if err != nil || registration.SessionID != "session-a" || client.register.GetZone() != "zone-a" {
+		t.Fatalf("Register() = %+v, %v; request=%+v", registration, err, client.register)
+	}
+	if err := reporter.Report(context.Background()); err != nil {
+		t.Fatalf("Report(): %v", err)
+	}
+	if client.runtime.GetSessionId() != "session-a" || client.runtime.GetReportSequence() != 1 ||
+		client.runtime.GetSnapshotVersion() != 1 || !client.runtime.GetObservedAt().AsTime().Equal(now) {
+		t.Fatalf("ReportRuntime request = %+v", client.runtime)
+	}
+	if err := reporter.Report(context.Background()); err != nil {
+		t.Fatalf("second Report(): %v", err)
+	}
+	if client.runtime.GetReportSequence() != 2 {
+		t.Fatalf("second report sequence = %d, want 2", client.runtime.GetReportSequence())
+	}
+}
+
+func TestRuntimeReporterRetainsSequenceOnTransportErrorAndSurfacesResync(t *testing.T) {
+	store := snapshot.NewStore("cluster-a", "worker-a")
+	applySnapshot(t, store)
+	transportErr := errors.New("temporary transport failure")
+	client := &clientStub{registration: &controlplanev1.RegisterWorkerResponse{
+		WorkerId: "worker-a", SessionId: "session-a", OwnershipEpoch: 7,
+		HeartbeatInterval: durationpb.New(time.Second), MaxStaleAge: durationpb.New(3 * time.Second),
+	}, runtimeErr: transportErr}
+	reporter, err := NewRuntimeReporter(client, store, Options{WorkerID: "worker-a", InstanceID: "instance-a", Zone: "zone-a", ProtocolVersion: 1, Now: time.Now})
+	if err != nil {
+		t.Fatalf("NewRuntimeReporter(): %v", err)
+	}
+	if _, err := reporter.Register(context.Background()); err != nil {
+		t.Fatalf("Register(): %v", err)
+	}
+	if err := reporter.Report(context.Background()); !errors.Is(err, transportErr) {
+		t.Fatalf("Report() error = %v, want transport error", err)
+	}
+	client.runtimeErr = nil
+	client.runtimeResponse = &controlplanev1.ReportRuntimeResponse{RequireFullSnapshot: true}
+	if err := reporter.Report(context.Background()); !errors.Is(err, ErrFullSnapshotRequired) {
+		t.Fatalf("Report() error = %v, want ErrFullSnapshotRequired", err)
+	}
+	if client.runtime.GetReportSequence() != 1 {
+		t.Fatalf("retried report sequence = %d, want 1", client.runtime.GetReportSequence())
+	}
+}
+
+func TestRuntimeReporterRejectsInvalidState(t *testing.T) {
+	store := snapshot.NewStore("cluster-a", "worker-a")
+	if _, err := NewRuntimeReporter(&clientStub{}, store, Options{}); !errors.Is(err, ErrInvalidOptions) {
+		t.Fatalf("NewRuntimeReporter() error = %v, want ErrInvalidOptions", err)
+	}
+	reporter, err := NewRuntimeReporter(&clientStub{}, store, Options{WorkerID: "worker-a", InstanceID: "instance-a", Zone: "zone-a", ProtocolVersion: 1, Now: time.Now})
+	if err != nil {
+		t.Fatalf("NewRuntimeReporter(): %v", err)
+	}
+	if err := reporter.Report(context.Background()); !errors.Is(err, ErrNotRegistered) {
+		t.Fatalf("Report() error = %v, want ErrNotRegistered", err)
+	}
+}
+
+type clientStub struct {
+	register        *controlplanev1.RegisterWorkerRequest
+	registration    *controlplanev1.RegisterWorkerResponse
+	registerErr     error
+	runtime         *controlplanev1.ReportRuntimeRequest
+	runtimeResponse *controlplanev1.ReportRuntimeResponse
+	runtimeErr      error
+}
+
+func (client *clientStub) RegisterWorker(_ context.Context, request *controlplanev1.RegisterWorkerRequest) (*controlplanev1.RegisterWorkerResponse, error) {
+	client.register = request
+	return client.registration, client.registerErr
+}
+
+func (client *clientStub) ReportRuntime(_ context.Context, request *controlplanev1.ReportRuntimeRequest) (*controlplanev1.ReportRuntimeResponse, error) {
+	client.runtime = request
+	if client.runtimeResponse == nil {
+		client.runtimeResponse = &controlplanev1.ReportRuntimeResponse{AcceptedOwnershipEpoch: 7}
+	}
+	return client.runtimeResponse, client.runtimeErr
+}
+
+func applySnapshot(t *testing.T, store *snapshot.Store) {
+	t.Helper()
+	envelope := snapshot.Envelope{ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 7, Version: 1, Full: true}
+	envelope.Checksum = snapshot.Checksum(nil)
+	if err := store.Apply(envelope); err != nil {
+		t.Fatalf("Apply(): %v", err)
+	}
+}