feat: add worker session runtime domain store

.github/workflows/ci.yml

@@ -8,6 +8,17 @@ permissions:
   contents: read
 
 jobs:
+  proto:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+          cache: true
+      - shell: pwsh
+        run: ./scripts/verify-proto.ps1
+
   test:
     strategy:
       matrix:

configs/proxy-pool.yaml

@@ -81,6 +81,9 @@ admin:
   auth:
     mode: none
 
+controlPlane:
+  enabled: false
+
 metrics:
   enabled: true
   listen: 127.0.0.1:9090

deploy/config/local.yaml

@@ -58,6 +58,9 @@ admin:
     header: X-Admin-Key
     token: "${PROXY_POOL_ADMIN_TOKEN}"
 
+controlPlane:
+  enabled: false
+
 metrics:
   enabled: true
   listen: 0.0.0.0:9090

deploy/kubernetes/base/configmap.yaml

@@ -60,6 +60,8 @@ data:
         mode: apiKey
         header: X-Admin-Key
         token: "${PROXY_POOL_ADMIN_TOKEN}"
+    controlPlane:
+      enabled: false
     metrics:
       enabled: true
       listen: 0.0.0.0:9090

docs/configuration/reference.md

@@ -52,6 +52,7 @@ security: {}
 gateway: {}
 distribution: {}
 admin: {}
+controlPlane: {}
 metrics: {}
 storage: {}
 routing: []
@@ -64,6 +65,7 @@ upstreams: {}
 - `gateway`：HTTP/HTTPS CONNECT 数据面入口。
 - `distribution`：一次性独占提取入口。
 - `admin`：运维管理入口，必须与 Distribution 分端口。
+- `controlPlane`：Worker 注册、Snapshot ACK 和 Runtime 心跳的 gRPC 控制面，默认关闭。
 - `metrics`：Prometheus 入口。
 - `storage`：Controller 使用的 PostgreSQL 与 Redis 地址。
 - `routing`：有序 Routing 列表，自上而下首条命中停止。
@@ -122,7 +124,59 @@ Lua 可精确表示的整数范围内。
 Gateway、Distribution、Admin 与 Provider API 是独立认证边界。改变其中一套
 不得连带改变其他入口。
 
-## 4. Gateway
+## 4. Worker 控制面
+
+控制面默认关闭；默认配置中的 `8443` 端口预留不表示服务已监听。控制面 Session、
+Snapshot ACK 和 Runtime 报告是 Redis 的短效运行时状态，**不写入 PostgreSQL**。
+
+仅本机开发可以使用回环明文：
+
+```yaml
+controlPlane:
+  enabled: true
+  listen: 127.0.0.1:8443
+  protocolVersion: 1
+  heartbeatInterval: 10s
+  sessionTTL: 30s
+  maxStaleAge: 10s
+  maxMessageBytes: 1048576
+  maxRuntimeCounters: 100000
+  maxConcurrentStreams: 128
+  tls:
+    mode: disabled
+```
+
+任何非回环监听地址都必须使用 mTLS：
+
+```yaml
+controlPlane:
+  enabled: true
+  listen: 0.0.0.0:8443
+  protocolVersion: 1
+  heartbeatInterval: 10s
+  sessionTTL: 30s
+  maxStaleAge: 10s
+  maxMessageBytes: 1048576
+  maxRuntimeCounters: 100000
+  maxConcurrentStreams: 128
+  tls:
+    mode: mtls
+    certFile: /run/secrets/controller-cert.pem
+    keyFile: /run/secrets/controller-key.pem
+    clientCAFile: /run/secrets/worker-ca.pem
+    trustDomain: proxy.example
+    environment: production
+```
+
+- `protocolVersion` 当前固定为 `1`。
+- `sessionTTL` 至少为 `3 * heartbeatInterval`；`maxStaleAge` 不得短于心跳间隔。
+- `maxMessageBytes` 范围为 `1..67108864`；`maxRuntimeCounters` 范围为
+  `1..1000000`；`maxConcurrentStreams` 必须为正数。
+- `tls.mode` 只能为 `disabled` 或 `mtls`。`disabled` 只允许回环监听；`mtls` 必须
+  同时配置证书、私钥、客户端 CA、全小写 DNS `trustDomain` 和单 URI 路径段
+  `environment`。
+
+## 5. Gateway
 
 ```yaml
 gateway:
@@ -151,7 +205,7 @@ gateway:
 - 保留地址、CGNAT 与云元数据端点始终拒绝，不能通过私网/链路本地开关放行。
 - `maxConcurrentConnections` 是入口准入上限，不是 Proxy 容量上限。
 
-## 5. Distribution
+## 6. Distribution
 
 ```yaml
 distribution:
@@ -195,7 +249,7 @@ Extraction 是固定的一次性独占行为，**没有** `mode`、`leaseDuratio
   启用认证。
 - `authenticatedClientOrSourceIP`：优先认证主体，无主体时回退来源地址。
 
-## 6. Routing
+## 7. Routing
 
 ```yaml
 routing:
@@ -231,7 +285,7 @@ Sequential 的空计数属于 Upstream，当前索引属于 Routing。只有 Pro
 成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数；重复候选
 会重置空计数但增加独立 duplicate 指标。
 
-## 7. Upstream
+## 8. Upstream
 
 ```yaml
 upstreams:
@@ -281,7 +335,7 @@ upstreams:
       urls: [http://connect.rom.miui.com/generate_204]
 ```
 
-### 7.1 Provider 与代理认证
+### 8.1 Provider 与代理认证
 
 - `api.auth` 用于系统访问 Provider API。
 - `proxyAuth` 用于最终连接被获取的 Proxy。
@@ -315,7 +369,7 @@ proxyAuth:
   password: "${PROVIDER_PROXY_PASSWORD}"
 ```
 
-### 7.2 Pool 与累计额度
+### 8.2 Pool 与累计额度
 
 - `pool.maxSize`：当前系统维护且尚未 EXTRACTED 的 Proxy 硬上限，包括
   FETCHED、CHECKING、AVAILABLE、SUSPECT、DRAINING 和 pending expected。
@@ -344,7 +398,7 @@ proxyAuth:
 内存 lease 上限按所有配置 Upstream 的 `pool.maxSize * fetch.maxInFlight` 汇总，
 配置 reload 只提高上限，不预分配对应内存。
 
-### 7.3 Fetch 限制
+### 8.3 Fetch 限制
 
 - `estimatedIPsPerCall`：冷启动时每次 Provider 调用预计返回的合法 Proxy 数，
   同时用于 `pool.maxSize` 的 pending 预占；不得从任意 Query 或 Body 字段推断。
@@ -359,7 +413,7 @@ proxyAuth:
 大量缺池信号必须合并成 singleflight 或容量为 1 的通知，不能按 Gateway 请求
 数量线性触发 Provider API。
 
-### 7.4 Refill 水位
+### 8.4 Refill 水位
 
 - `reconcileInterval`：无事件时重新读取库存的兜底周期，启动时仍立即检查一次。
 - `minimumAvailableSlots`：可用并发槽位低于该值时进入补池。
@@ -370,7 +424,7 @@ proxyAuth:
 `requestInterval` 只限制外部 API 调用，不能兼任库存复核周期。pending Proxy
 按 `estimatedIPsPerCall * maxConcurrencyPerProxy` 折算槽位，避免并发补池超量。
 
-### 7.5 生命周期与健康
+### 8.5 生命周期与健康
 
 - 明确绝对过期时间优先于响应 TTL，响应 TTL 优先于配置 `lifecycle.ttl`。
 - 距离过期不足 `allocationSafetyMargin` 时停止新分配。
@@ -378,7 +432,7 @@ proxyAuth:
 - 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入
   UNHEALTHY。
 
-## 8. 存储、Admin 与 Metrics
+## 9. 存储、Admin 与 Metrics
 
 ```yaml
 admin:
@@ -415,7 +469,7 @@ Metrics 启用时 `listen` 必须是合法 `host:port`。该入口固定提供 `
 服务流量门槛，PostgreSQL 故障由 Admin 接口独立报告。Metrics 开关或监听地址
 变更需要重启 Controller。
 
-## 9. 启动前校验清单
+## 10. 启动前校验清单
 
 1. `version` 必须为 `1`，未知字段拒绝。
 2. 所有启用监听器具有合法 `host:port`。

gen/controlplane/v1/controlplane_grpc.pb.go

@@ -0,0 +1,429 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.2
+// - protoc             v7.35.0
+// source: controlplane/v1/controlplane.proto
+
+package controlplanev1
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+	emptypb "google.golang.org/protobuf/types/known/emptypb"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	WorkerControlPlane_RegisterWorker_FullMethodName      = "/proxy_pool.controlplane.v1.WorkerControlPlane/RegisterWorker"
+	WorkerControlPlane_WatchSnapshots_FullMethodName      = "/proxy_pool.controlplane.v1.WorkerControlPlane/WatchSnapshots"
+	WorkerControlPlane_AcknowledgeSnapshot_FullMethodName = "/proxy_pool.controlplane.v1.WorkerControlPlane/AcknowledgeSnapshot"
+	WorkerControlPlane_ReportOutcomes_FullMethodName      = "/proxy_pool.controlplane.v1.WorkerControlPlane/ReportOutcomes"
+	WorkerControlPlane_ReportRuntime_FullMethodName       = "/proxy_pool.controlplane.v1.WorkerControlPlane/ReportRuntime"
+)
+
+// WorkerControlPlaneClient is the client API for WorkerControlPlane service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// WorkerControlPlane distributes immutable, worker-specific snapshots. The
+// gateway hot path does not call this service for individual requests.
+type WorkerControlPlaneClient interface {
+	RegisterWorker(ctx context.Context, in *RegisterWorkerRequest, opts ...grpc.CallOption) (*RegisterWorkerResponse, error)
+	WatchSnapshots(ctx context.Context, in *WatchSnapshotsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SnapshotEnvelope], error)
+	AcknowledgeSnapshot(ctx context.Context, in *AcknowledgeSnapshotRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	ReportOutcomes(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[OutcomeBatch, ReportOutcomesResponse], error)
+	ReportRuntime(ctx context.Context, in *ReportRuntimeRequest, opts ...grpc.CallOption) (*ReportRuntimeResponse, error)
+}
+
+type workerControlPlaneClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewWorkerControlPlaneClient(cc grpc.ClientConnInterface) WorkerControlPlaneClient {
+	return &workerControlPlaneClient{cc}
+}
+
+func (c *workerControlPlaneClient) RegisterWorker(ctx context.Context, in *RegisterWorkerRequest, opts ...grpc.CallOption) (*RegisterWorkerResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(RegisterWorkerResponse)
+	err := c.cc.Invoke(ctx, WorkerControlPlane_RegisterWorker_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *workerControlPlaneClient) WatchSnapshots(ctx context.Context, in *WatchSnapshotsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SnapshotEnvelope], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &WorkerControlPlane_ServiceDesc.Streams[0], WorkerControlPlane_WatchSnapshots_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[WatchSnapshotsRequest, SnapshotEnvelope]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type WorkerControlPlane_WatchSnapshotsClient = grpc.ServerStreamingClient[SnapshotEnvelope]
+
+func (c *workerControlPlaneClient) AcknowledgeSnapshot(ctx context.Context, in *AcknowledgeSnapshotRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(emptypb.Empty)
+	err := c.cc.Invoke(ctx, WorkerControlPlane_AcknowledgeSnapshot_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *workerControlPlaneClient) ReportOutcomes(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[OutcomeBatch, ReportOutcomesResponse], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &WorkerControlPlane_ServiceDesc.Streams[1], WorkerControlPlane_ReportOutcomes_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[OutcomeBatch, ReportOutcomesResponse]{ClientStream: stream}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type WorkerControlPlane_ReportOutcomesClient = grpc.ClientStreamingClient[OutcomeBatch, ReportOutcomesResponse]
+
+func (c *workerControlPlaneClient) ReportRuntime(ctx context.Context, in *ReportRuntimeRequest, opts ...grpc.CallOption) (*ReportRuntimeResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ReportRuntimeResponse)
+	err := c.cc.Invoke(ctx, WorkerControlPlane_ReportRuntime_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// WorkerControlPlaneServer is the server API for WorkerControlPlane service.
+// All implementations must embed UnimplementedWorkerControlPlaneServer
+// for forward compatibility.
+//
+// WorkerControlPlane distributes immutable, worker-specific snapshots. The
+// gateway hot path does not call this service for individual requests.
+type WorkerControlPlaneServer interface {
+	RegisterWorker(context.Context, *RegisterWorkerRequest) (*RegisterWorkerResponse, error)
+	WatchSnapshots(*WatchSnapshotsRequest, grpc.ServerStreamingServer[SnapshotEnvelope]) error
+	AcknowledgeSnapshot(context.Context, *AcknowledgeSnapshotRequest) (*emptypb.Empty, error)
+	ReportOutcomes(grpc.ClientStreamingServer[OutcomeBatch, ReportOutcomesResponse]) error
+	ReportRuntime(context.Context, *ReportRuntimeRequest) (*ReportRuntimeResponse, error)
+	mustEmbedUnimplementedWorkerControlPlaneServer()
+}
+
+// UnimplementedWorkerControlPlaneServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedWorkerControlPlaneServer struct{}
+
+func (UnimplementedWorkerControlPlaneServer) RegisterWorker(context.Context, *RegisterWorkerRequest) (*RegisterWorkerResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method RegisterWorker not implemented")
+}
+func (UnimplementedWorkerControlPlaneServer) WatchSnapshots(*WatchSnapshotsRequest, grpc.ServerStreamingServer[SnapshotEnvelope]) error {
+	return status.Error(codes.Unimplemented, "method WatchSnapshots not implemented")
+}
+func (UnimplementedWorkerControlPlaneServer) AcknowledgeSnapshot(context.Context, *AcknowledgeSnapshotRequest) (*emptypb.Empty, error) {
+	return nil, status.Error(codes.Unimplemented, "method AcknowledgeSnapshot not implemented")
+}
+func (UnimplementedWorkerControlPlaneServer) ReportOutcomes(grpc.ClientStreamingServer[OutcomeBatch, ReportOutcomesResponse]) error {
+	return status.Error(codes.Unimplemented, "method ReportOutcomes not implemented")
+}
+func (UnimplementedWorkerControlPlaneServer) ReportRuntime(context.Context, *ReportRuntimeRequest) (*ReportRuntimeResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ReportRuntime not implemented")
+}
+func (UnimplementedWorkerControlPlaneServer) mustEmbedUnimplementedWorkerControlPlaneServer() {}
+func (UnimplementedWorkerControlPlaneServer) testEmbeddedByValue()                            {}
+
+// UnsafeWorkerControlPlaneServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to WorkerControlPlaneServer will
+// result in compilation errors.
+type UnsafeWorkerControlPlaneServer interface {
+	mustEmbedUnimplementedWorkerControlPlaneServer()
+}
+
+func RegisterWorkerControlPlaneServer(s grpc.ServiceRegistrar, srv WorkerControlPlaneServer) {
+	// If the following call panics, it indicates UnimplementedWorkerControlPlaneServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&WorkerControlPlane_ServiceDesc, srv)
+}
+
+func _WorkerControlPlane_RegisterWorker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(RegisterWorkerRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(WorkerControlPlaneServer).RegisterWorker(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: WorkerControlPlane_RegisterWorker_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(WorkerControlPlaneServer).RegisterWorker(ctx, req.(*RegisterWorkerRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _WorkerControlPlane_WatchSnapshots_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(WatchSnapshotsRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(WorkerControlPlaneServer).WatchSnapshots(m, &grpc.GenericServerStream[WatchSnapshotsRequest, SnapshotEnvelope]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type WorkerControlPlane_WatchSnapshotsServer = grpc.ServerStreamingServer[SnapshotEnvelope]
+
+func _WorkerControlPlane_AcknowledgeSnapshot_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(AcknowledgeSnapshotRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(WorkerControlPlaneServer).AcknowledgeSnapshot(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: WorkerControlPlane_AcknowledgeSnapshot_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(WorkerControlPlaneServer).AcknowledgeSnapshot(ctx, req.(*AcknowledgeSnapshotRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _WorkerControlPlane_ReportOutcomes_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(WorkerControlPlaneServer).ReportOutcomes(&grpc.GenericServerStream[OutcomeBatch, ReportOutcomesResponse]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type WorkerControlPlane_ReportOutcomesServer = grpc.ClientStreamingServer[OutcomeBatch, ReportOutcomesResponse]
+
+func _WorkerControlPlane_ReportRuntime_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReportRuntimeRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(WorkerControlPlaneServer).ReportRuntime(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: WorkerControlPlane_ReportRuntime_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(WorkerControlPlaneServer).ReportRuntime(ctx, req.(*ReportRuntimeRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// WorkerControlPlane_ServiceDesc is the grpc.ServiceDesc for WorkerControlPlane service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var WorkerControlPlane_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "proxy_pool.controlplane.v1.WorkerControlPlane",
+	HandlerType: (*WorkerControlPlaneServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "RegisterWorker",
+			Handler:    _WorkerControlPlane_RegisterWorker_Handler,
+		},
+		{
+			MethodName: "AcknowledgeSnapshot",
+			Handler:    _WorkerControlPlane_AcknowledgeSnapshot_Handler,
+		},
+		{
+			MethodName: "ReportRuntime",
+			Handler:    _WorkerControlPlane_ReportRuntime_Handler,
+		},
+	},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "WatchSnapshots",
+			Handler:       _WorkerControlPlane_WatchSnapshots_Handler,
+			ServerStreams: true,
+		},
+		{
+			StreamName:    "ReportOutcomes",
+			Handler:       _WorkerControlPlane_ReportOutcomes_Handler,
+			ClientStreams: true,
+		},
+	},
+	Metadata: "controlplane/v1/controlplane.proto",
+}
+
+const (
+	CheckerControlPlane_StreamCheckTasks_FullMethodName   = "/proxy_pool.controlplane.v1.CheckerControlPlane/StreamCheckTasks"
+	CheckerControlPlane_ReportObservations_FullMethodName = "/proxy_pool.controlplane.v1.CheckerControlPlane/ReportObservations"
+)
+
+// CheckerControlPlaneClient is the client API for CheckerControlPlane service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// CheckerControlPlane hands bounded check work to independently scalable
+// checker processes. Observations are facts; only the Controller reducer may
+// change authoritative proxy state.
+type CheckerControlPlaneClient interface {
+	StreamCheckTasks(ctx context.Context, in *StreamCheckTasksRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[CheckTask], error)
+	ReportObservations(ctx context.Context, in *ObservationBatch, opts ...grpc.CallOption) (*ReportObservationsResponse, error)
+}
+
+type checkerControlPlaneClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewCheckerControlPlaneClient(cc grpc.ClientConnInterface) CheckerControlPlaneClient {
+	return &checkerControlPlaneClient{cc}
+}
+
+func (c *checkerControlPlaneClient) StreamCheckTasks(ctx context.Context, in *StreamCheckTasksRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[CheckTask], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &CheckerControlPlane_ServiceDesc.Streams[0], CheckerControlPlane_StreamCheckTasks_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[StreamCheckTasksRequest, CheckTask]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type CheckerControlPlane_StreamCheckTasksClient = grpc.ServerStreamingClient[CheckTask]
+
+func (c *checkerControlPlaneClient) ReportObservations(ctx context.Context, in *ObservationBatch, opts ...grpc.CallOption) (*ReportObservationsResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ReportObservationsResponse)
+	err := c.cc.Invoke(ctx, CheckerControlPlane_ReportObservations_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// CheckerControlPlaneServer is the server API for CheckerControlPlane service.
+// All implementations must embed UnimplementedCheckerControlPlaneServer
+// for forward compatibility.
+//
+// CheckerControlPlane hands bounded check work to independently scalable
+// checker processes. Observations are facts; only the Controller reducer may
+// change authoritative proxy state.
+type CheckerControlPlaneServer interface {
+	StreamCheckTasks(*StreamCheckTasksRequest, grpc.ServerStreamingServer[CheckTask]) error
+	ReportObservations(context.Context, *ObservationBatch) (*ReportObservationsResponse, error)
+	mustEmbedUnimplementedCheckerControlPlaneServer()
+}
+
+// UnimplementedCheckerControlPlaneServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedCheckerControlPlaneServer struct{}
+
+func (UnimplementedCheckerControlPlaneServer) StreamCheckTasks(*StreamCheckTasksRequest, grpc.ServerStreamingServer[CheckTask]) error {
+	return status.Error(codes.Unimplemented, "method StreamCheckTasks not implemented")
+}
+func (UnimplementedCheckerControlPlaneServer) ReportObservations(context.Context, *ObservationBatch) (*ReportObservationsResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ReportObservations not implemented")
+}
+func (UnimplementedCheckerControlPlaneServer) mustEmbedUnimplementedCheckerControlPlaneServer() {}
+func (UnimplementedCheckerControlPlaneServer) testEmbeddedByValue()                             {}
+
+// UnsafeCheckerControlPlaneServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to CheckerControlPlaneServer will
+// result in compilation errors.
+type UnsafeCheckerControlPlaneServer interface {
+	mustEmbedUnimplementedCheckerControlPlaneServer()
+}
+
+func RegisterCheckerControlPlaneServer(s grpc.ServiceRegistrar, srv CheckerControlPlaneServer) {
+	// If the following call panics, it indicates UnimplementedCheckerControlPlaneServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&CheckerControlPlane_ServiceDesc, srv)
+}
+
+func _CheckerControlPlane_StreamCheckTasks_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(StreamCheckTasksRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(CheckerControlPlaneServer).StreamCheckTasks(m, &grpc.GenericServerStream[StreamCheckTasksRequest, CheckTask]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type CheckerControlPlane_StreamCheckTasksServer = grpc.ServerStreamingServer[CheckTask]
+
+func _CheckerControlPlane_ReportObservations_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ObservationBatch)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CheckerControlPlaneServer).ReportObservations(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: CheckerControlPlane_ReportObservations_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CheckerControlPlaneServer).ReportObservations(ctx, req.(*ObservationBatch))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// CheckerControlPlane_ServiceDesc is the grpc.ServiceDesc for CheckerControlPlane service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var CheckerControlPlane_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "proxy_pool.controlplane.v1.CheckerControlPlane",
+	HandlerType: (*CheckerControlPlaneServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "ReportObservations",
+			Handler:    _CheckerControlPlane_ReportObservations_Handler,
+		},
+	},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "StreamCheckTasks",
+			Handler:       _CheckerControlPlane_StreamCheckTasks_Handler,
+			ServerStreams: true,
+		},
+	},
+	Metadata: "controlplane/v1/controlplane.proto",
+}

go.mod

@@ -7,6 +7,8 @@ require (
 	github.com/prometheus/client_golang v1.24.1
 	github.com/redis/go-redis/v9 v9.19.0
 	go.yaml.in/yaml/v4 v4.0.0-rc.3
+	google.golang.org/grpc v1.83.0
+	google.golang.org/protobuf v1.36.11
 )
 
 require (
@@ -20,9 +22,16 @@ require (
 	github.com/prometheus/common v0.70.1 // indirect
 	github.com/prometheus/procfs v0.21.1 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
-	golang.org/x/crypto v0.17.0 // indirect
+	golang.org/x/crypto v0.54.0 // indirect
+	golang.org/x/net v0.57.0 // indirect
 	golang.org/x/sync v0.22.0 // indirect
 	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
-	google.golang.org/protobuf v1.36.11 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 // indirect
+)
+
+tool (
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc
+	google.golang.org/protobuf/cmd/protoc-gen-go
 )

go.sum

@@ -9,8 +9,16 @@ github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
@@ -46,6 +54,18 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
+go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
+go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
+go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
+go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
+go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
+go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
+go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
+go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
+go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
+go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
@@ -54,14 +74,24 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go=
 go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0=
-golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
+golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
-golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
 golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
 golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
 golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
 golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
+gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
+gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ=
+google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ=
+google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 h1:rgSNvqscFZ1JgV/4wH5GOsZFSFkR2Eua9As3KIr2LlM=
+google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2/go.mod h1:iMEtFwDlAhjDU9L5mY6U1XLwlIId/G3h+QcBHDIvrJ8=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/config/config.go

@@ -31,6 +31,7 @@ type Config struct {
 	Gateway      Listener            `yaml:"gateway"`
 	Distribution Distribution        `yaml:"distribution"`
 	Admin        Listener            `yaml:"admin"`
+	ControlPlane ControlPlane        `yaml:"controlPlane"`
 	Metrics      Metrics             `yaml:"metrics"`
 	Storage      Storage             `yaml:"storage"`
 	Routing      []Routing           `yaml:"routing"`
@@ -127,6 +128,28 @@ type Metrics struct {
 	Listen  string `yaml:"listen"`
 }
 
+type ControlPlane struct {
+	Enabled              bool            `yaml:"enabled"`
+	Listen               string          `yaml:"listen"`
+	ProtocolVersion      uint32          `yaml:"protocolVersion"`
+	HeartbeatInterval    Duration        `yaml:"heartbeatInterval"`
+	SessionTTL           Duration        `yaml:"sessionTTL"`
+	MaxStaleAge          Duration        `yaml:"maxStaleAge"`
+	MaxMessageBytes      int             `yaml:"maxMessageBytes"`
+	MaxRuntimeCounters   int             `yaml:"maxRuntimeCounters"`
+	MaxConcurrentStreams uint32          `yaml:"maxConcurrentStreams"`
+	TLS                  ControlPlaneTLS `yaml:"tls"`
+}
+
+type ControlPlaneTLS struct {
+	Mode         string `yaml:"mode"`
+	CertFile     string `yaml:"certFile"`
+	KeyFile      string `yaml:"keyFile"`
+	ClientCAFile string `yaml:"clientCAFile"`
+	TrustDomain  string `yaml:"trustDomain"`
+	Environment  string `yaml:"environment"`
+}
+
 type Storage struct {
 	PostgresURL string `yaml:"postgresURL"`
 	RedisURL    string `yaml:"redisURL"`

internal/config/config_test.go

@@ -352,6 +352,150 @@ func TestValidateMetricsListener(t *testing.T) {
 	}
 }
 
+func TestValidateControlPlane(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*Config)
+		want   string
+	}{
+		{
+			name: "missing listen",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.Listen = ""
+			},
+			want: "controlPlane listen",
+		},
+		{
+			name: "public plaintext",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.Listen = "0.0.0.0:8443"
+			},
+			want: "requires mtls",
+		},
+		{
+			name: "short session ttl",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.SessionTTL = Duration(29 * time.Second)
+			},
+			want: "sessionTTL",
+		},
+		{
+			name: "stale below heartbeat",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.MaxStaleAge = Duration(9 * time.Second)
+			},
+			want: "maxStaleAge",
+		},
+		{
+			name: "invalid protocol",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.ProtocolVersion = 0
+			},
+			want: "protocolVersion",
+		},
+		{
+			name: "invalid message limit",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.MaxMessageBytes = 64<<20 + 1
+			},
+			want: "maxMessageBytes",
+		},
+		{
+			name: "invalid counter limit",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.MaxRuntimeCounters = MaximumPoolSize + 1
+			},
+			want: "maxRuntimeCounters",
+		},
+		{
+			name: "zero concurrent streams",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.MaxConcurrentStreams = 0
+			},
+			want: "maxConcurrentStreams",
+		},
+		{
+			name: "unsupported tls mode",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validControlPlane()
+				cfg.ControlPlane.TLS.Mode = "serverTLS"
+			},
+			want: "tls.mode",
+		},
+		{
+			name: "missing mtls files",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validMTLSControlPlane()
+				cfg.ControlPlane.TLS.CertFile = ""
+			},
+			want: "certFile",
+		},
+		{
+			name: "trust domain with port",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validMTLSControlPlane()
+				cfg.ControlPlane.TLS.TrustDomain = "proxy.example:443"
+			},
+			want: "trustDomain",
+		},
+		{
+			name: "environment path",
+			mutate: func(cfg *Config) {
+				cfg.ControlPlane = validMTLSControlPlane()
+				cfg.ControlPlane.TLS.Environment = "prod/eu"
+			},
+			want: "environment",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := mustLoadValidConfig(t)
+			test.mutate(cfg)
+			if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("Validate() error = %v, want substring %q", err, test.want)
+			}
+		})
+	}
+}
+
+func validControlPlane() ControlPlane {
+	return ControlPlane{
+		Enabled:              true,
+		Listen:               "127.0.0.1:8443",
+		ProtocolVersion:      1,
+		HeartbeatInterval:    Duration(10 * time.Second),
+		SessionTTL:           Duration(30 * time.Second),
+		MaxStaleAge:          Duration(10 * time.Second),
+		MaxMessageBytes:      1 << 20,
+		MaxRuntimeCounters:   100_000,
+		MaxConcurrentStreams: 128,
+		TLS:                  ControlPlaneTLS{Mode: "disabled"},
+	}
+}
+
+func validMTLSControlPlane() ControlPlane {
+	controlPlane := validControlPlane()
+	controlPlane.Listen = "0.0.0.0:8443"
+	controlPlane.TLS = ControlPlaneTLS{
+		Mode:         "mtls",
+		CertFile:     "/run/secrets/controller-cert.pem",
+		KeyFile:      "/run/secrets/controller-key.pem",
+		ClientCAFile: "/run/secrets/worker-ca.pem",
+		TrustDomain:  "proxy.example",
+		Environment:  "production",
+	}
+	return controlPlane
+}
+
 func TestValidateRejectsInvalidConfigurationMatrix(t *testing.T) {
 	tests := []struct {
 		name   string

internal/config/validate.go

@@ -10,6 +10,11 @@ import (
 	"strings"
 )
 
+var (
+	controlPlaneIdentityPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)
+	trustDomainLabelPattern     = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`)
+)
+
 func Validate(cfg *Config) error {
 	if cfg == nil {
 		return fmt.Errorf("validate configuration: nil config")
@@ -38,6 +43,9 @@ func Validate(cfg *Config) error {
 			return err
 		}
 	}
+	if err := validateControlPlane(cfg.ControlPlane); err != nil {
+		return err
+	}
 	if fetchConfigured(cfg.Defaults.Fetch) {
 		if err := validateFetch("defaults.fetch", cfg.Defaults.Fetch); err != nil {
 			return err
@@ -101,6 +109,70 @@ func Validate(cfg *Config) error {
 	return nil
 }
 
+func validateControlPlane(item ControlPlane) error {
+	if !item.Enabled {
+		return nil
+	}
+	host, err := validateListenAddress("controlPlane", item.Listen)
+	if err != nil {
+		return err
+	}
+	if item.ProtocolVersion != 1 {
+		return fmt.Errorf("validate controlPlane protocolVersion: must be 1")
+	}
+	heartbeat := item.HeartbeatInterval.Value()
+	if heartbeat <= 0 {
+		return fmt.Errorf("validate controlPlane heartbeatInterval: must be greater than zero")
+	}
+	if item.SessionTTL.Value()/3 < heartbeat {
+		return fmt.Errorf("validate controlPlane sessionTTL: must be at least three heartbeat intervals")
+	}
+	if item.MaxStaleAge.Value() < heartbeat {
+		return fmt.Errorf("validate controlPlane maxStaleAge: must not be shorter than heartbeatInterval")
+	}
+	if item.MaxMessageBytes <= 0 || item.MaxMessageBytes > 64<<20 {
+		return fmt.Errorf("validate controlPlane maxMessageBytes: must be in [1, 67108864]")
+	}
+	if item.MaxRuntimeCounters <= 0 || item.MaxRuntimeCounters > MaximumPoolSize {
+		return fmt.Errorf("validate controlPlane maxRuntimeCounters: must be in [1, %d]", MaximumPoolSize)
+	}
+	if item.MaxConcurrentStreams == 0 {
+		return fmt.Errorf("validate controlPlane maxConcurrentStreams: must be positive")
+	}
+	switch item.TLS.Mode {
+	case "disabled":
+		if isPublicHost(host) {
+			return fmt.Errorf("validate controlPlane tls: non-loopback listen requires mtls")
+		}
+	case "mtls":
+		if item.TLS.CertFile == "" || item.TLS.KeyFile == "" || item.TLS.ClientCAFile == "" ||
+			item.TLS.TrustDomain == "" || item.TLS.Environment == "" {
+			return fmt.Errorf("validate controlPlane tls: mtls requires certFile, keyFile, clientCAFile, trustDomain and environment")
+		}
+		if !validTrustDomain(item.TLS.TrustDomain) {
+			return fmt.Errorf("validate controlPlane tls.trustDomain: must be a lowercase DNS name without port")
+		}
+		if !controlPlaneIdentityPattern.MatchString(item.TLS.Environment) {
+			return fmt.Errorf("validate controlPlane tls.environment: must be one URI path segment")
+		}
+	default:
+		return fmt.Errorf("validate controlPlane tls.mode: must be disabled or mtls")
+	}
+	return nil
+}
+
+func validTrustDomain(value string) bool {
+	if len(value) == 0 || len(value) > 253 {
+		return false
+	}
+	for _, label := range strings.Split(value, ".") {
+		if !trustDomainLabelPattern.MatchString(label) {
+			return false
+		}
+	}
+	return true
+}
+
 func validateListener(name string, listener Listener, security Security) error {
 	if !listener.Enabled {
 		return nil

internal/domain/workerruntime/contract_external_test.go

@@ -0,0 +1,24 @@
+package workerruntime_test
+
+import (
+	"testing"
+	"time"
+
+	"proxy-pool/internal/domain/workerruntime"
+	"proxy-pool/internal/domain/workerruntime/contracttest"
+)
+
+func TestMemoryStoreContract(t *testing.T) {
+	now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
+	contracttest.Run(t, func(t *testing.T) contracttest.Fixture {
+		t.Helper()
+		store, err := workerruntime.NewMemoryStore(func() time.Time { return now })
+		if err != nil {
+			t.Fatalf("NewMemoryStore(): %v", err)
+		}
+		return contracttest.Fixture{
+			Store: store, Reader: store,
+			Advance: func(duration time.Duration) { now = now.Add(duration) },
+		}
+	})
+}

internal/domain/workerruntime/contracttest/contract.go

@@ -0,0 +1,131 @@
+package contracttest
+
+import (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/domain/workerruntime"
+)
+
+type Fixture struct {
+	Store   workerruntime.ControlStore
+	Reader  workerruntime.RuntimeReader
+	Advance func(time.Duration)
+}
+
+type Factory func(*testing.T) Fixture
+
+// Run exercises the public control-store behavior shared by Memory and Redis.
+func Run(t *testing.T, factory Factory) {
+	t.Helper()
+	t.Run("acknowledged runtime lifecycle", func(t *testing.T) { runLifecycle(t, newFixture(t, factory)) })
+	t.Run("negative acknowledgement fences runtime", func(t *testing.T) { runNegativeAck(t, newFixture(t, factory)) })
+}
+
+func runLifecycle(t *testing.T, fixture Fixture) {
+	t.Helper()
+	ctx := context.Background()
+	open(t, fixture.Store)
+	epoch, err := fixture.Store.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		t.Fatalf("CurrentOwnershipEpoch(): %v", err)
+	}
+	reference := snapshot(7, epoch, "snapshot-7")
+	report := runtimeReport(1, 7, epoch)
+	if err := fixture.Store.ReplaceRuntime(ctx, report, time.Minute); !errors.Is(err, workerruntime.ErrSnapshotMismatch) {
+		t.Fatalf("ReplaceRuntime(before ACK) error = %v", err)
+	}
+	if err := fixture.Store.RecordIssuedSnapshot(ctx, reference, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(): %v", err)
+	}
+	ack := workerruntime.SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true}
+	if err := fixture.Store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(): %v", err)
+	}
+	if err := fixture.Store.ReplaceRuntime(ctx, report, time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(after ACK): %v", err)
+	}
+	assertFresh(t, fixture.Reader, epoch, true)
+	if err := fixture.Store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(replay): %v", err)
+	}
+	if err := fixture.Store.ReplaceRuntime(ctx, runtimeReport(0, 7, epoch), time.Minute); !errors.Is(err, workerruntime.ErrInvalidReport) {
+		t.Fatalf("ReplaceRuntime(invalid sequence) error = %v", err)
+	}
+	fixture.Advance(2 * time.Minute)
+	assertFresh(t, fixture.Reader, epoch, false)
+}
+
+func runNegativeAck(t *testing.T, fixture Fixture) {
+	t.Helper()
+	ctx := context.Background()
+	open(t, fixture.Store)
+	epoch, err := fixture.Store.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		t.Fatalf("CurrentOwnershipEpoch(): %v", err)
+	}
+	first := snapshot(7, epoch, "snapshot-7")
+	if err := fixture.Store.RecordIssuedSnapshot(ctx, first, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(first): %v", err)
+	}
+	if err := fixture.Store.AcknowledgeSnapshot(ctx, workerruntime.SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: first, Applied: true}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(first): %v", err)
+	}
+	second := snapshot(8, epoch, "snapshot-8")
+	if err := fixture.Store.RecordIssuedSnapshot(ctx, second, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(second): %v", err)
+	}
+	if err := fixture.Store.AcknowledgeSnapshot(ctx, workerruntime.SnapshotAcknowledgement{
+		WorkerID: "worker-a", SessionID: "session-a", Reference: second, ErrorCode: "apply_failed",
+	}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(negative): %v", err)
+	}
+	if err := fixture.Store.ReplaceRuntime(ctx, runtimeReport(1, 7, epoch), time.Minute); !errors.Is(err, workerruntime.ErrSnapshotMismatch) {
+		t.Fatalf("ReplaceRuntime(delayed): %v", err)
+	}
+}
+
+func newFixture(t *testing.T, factory Factory) Fixture {
+	t.Helper()
+	fixture := factory(t)
+	if fixture.Store == nil || fixture.Reader == nil || fixture.Advance == nil {
+		t.Fatal("contract fixture is incomplete")
+	}
+	return fixture
+}
+
+func open(t *testing.T, store workerruntime.ControlStore) {
+	t.Helper()
+	err := store.OpenSession(context.Background(), workerruntime.Session{
+		WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a", Zone: "zone-a", ProtocolVersion: 1,
+	}, time.Minute)
+	if err != nil {
+		t.Fatalf("OpenSession(): %v", err)
+	}
+}
+
+func snapshot(version, epoch uint64, value string) workerruntime.SnapshotReference {
+	return workerruntime.SnapshotReference{
+		WorkerID: "worker-a", Version: version, OwnershipEpoch: epoch, Checksum: sha256.Sum256([]byte(value)),
+	}
+}
+
+func runtimeReport(sequence, version, epoch uint64) workerruntime.Report {
+	return workerruntime.Report{
+		WorkerID: "worker-a", SessionID: "session-a", Sequence: sequence, SnapshotVersion: version,
+		OwnershipEpoch: epoch, ObservedAt: time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC),
+	}
+}
+
+func assertFresh(t *testing.T, reader workerruntime.RuntimeReader, epoch uint64, want bool) {
+	t.Helper()
+	snapshots, err := reader.ReadRuntime(context.Background(), []workerruntime.OwnedProxy{{
+		ProxyID: "proxy-a", WorkerID: "worker-a", OwnershipEpoch: epoch,
+	}})
+	if err != nil || len(snapshots) != 1 || snapshots[0].Fresh != want {
+		t.Fatalf("ReadRuntime() = %+v, %v; want Fresh=%t", snapshots, err, want)
+	}
+}

internal/domain/workerruntime/memory.go

@@ -3,18 +3,17 @@ package workerruntime
 import (
 	"context"
 	"crypto/sha256"
-	"encoding/json"
-	"sort"
-	"strings"
 	"sync"
 	"time"
 )
 
 type MemoryStore struct {
-	mu       sync.Mutex
+	mu         sync.Mutex
-	now      func() time.Time
+	now        func() time.Time
-	sessions map[string]memorySession
+	epoch      uint64
-	reports  map[string]memoryReport
+	sessions   map[string]memorySession
+	references map[string]memoryReference
+	reports    map[string]memoryReport
 }
 
 type memorySession struct {
@@ -22,6 +21,11 @@ type memorySession struct {
 	expiresAt time.Time
 }
 
+type memoryReference struct {
+	value     SnapshotReference
+	expiresAt time.Time
+}
+
 type memoryReport struct {
 	value     Report
 	digest    [sha256.Size]byte
@@ -30,6 +34,7 @@ type memoryReport struct {
 }
 
 var (
+	_ ControlStore  = (*MemoryStore)(nil)
 	_ SessionWriter = (*MemoryStore)(nil)
 	_ ReportWriter  = (*MemoryStore)(nil)
 	_ RuntimeReader = (*MemoryStore)(nil)
@@ -40,33 +45,184 @@ func NewMemoryStore(now func() time.Time) (*MemoryStore, error) {
 		return nil, ErrInvalidStore
 	}
 	return &MemoryStore{
-		now: now, sessions: make(map[string]memorySession), reports: make(map[string]memoryReport),
+		now: now, epoch: 1, sessions: make(map[string]memorySession),
+		references: make(map[string]memoryReference), reports: make(map[string]memoryReport),
 	}, nil
 }
 
-func (store *MemoryStore) ReplaceSession(ctx context.Context, session Session, ttl time.Duration) error {
+func (store *MemoryStore) CurrentOwnershipEpoch(ctx context.Context) (uint64, error) {
-	if ctx == nil || store == nil || !validSession(session) || ttl <= 0 {
+	if ctx == nil || store == nil {
+		return 0, ErrInvalidStore
+	}
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if store.epoch == 0 {
+		return 0, ErrInvalidStore
+	}
+	return store.epoch, nil
+}
+
+// OpenSession always replaces the previous Worker session and clears Runtime.
+func (store *MemoryStore) OpenSession(ctx context.Context, session Session, ttl time.Duration) error {
+	if ctx == nil || store == nil || ttl <= 0 {
 		return ErrInvalidSession
 	}
 	if err := ctx.Err(); err != nil {
 		return err
 	}
-	now := store.now().UTC()
+	normalized, err := NormalizeSession(session)
-	if now.IsZero() {
+	if err != nil {
-		return ErrInvalidStore
+		return err
+	}
+	now, err := store.currentTime()
+	if err != nil {
+		return err
+	}
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	delete(store.reports, normalized.WorkerID)
+	store.sessions[normalized.WorkerID] = memorySession{value: normalized, expiresAt: now.Add(ttl)}
+	return nil
+}
+
+func (store *MemoryStore) RecordIssuedSnapshot(ctx context.Context, reference SnapshotReference, ttl time.Duration) error {
+	if ctx == nil || store == nil || ttl <= 0 {
+		return ErrInvalidSnapshotReference
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	normalized, err := NormalizeSnapshotReference(reference)
+	if err != nil {
+		return err
+	}
+	now, err := store.currentTime()
+	if err != nil {
+		return err
+	}
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if normalized.OwnershipEpoch != store.epoch {
+		return ErrSnapshotMismatch
+	}
+	current, exists := store.references[normalized.WorkerID]
+	if exists && !current.expiresAt.After(now) {
+		delete(store.references, normalized.WorkerID)
+		exists = false
+	}
+	if exists {
+		switch compareSnapshotTuple(normalized, current.value) {
+		case -1:
+			return ErrStaleSnapshotReference
+		case 0:
+			if normalized.Checksum != current.value.Checksum {
+				return ErrConflictingSnapshotReference
+			}
+		}
+	}
+	store.references[normalized.WorkerID] = memoryReference{value: normalized, expiresAt: now.Add(ttl)}
+	return nil
+}
+
+func (store *MemoryStore) AcknowledgeSnapshot(ctx context.Context, acknowledgement SnapshotAcknowledgement, ttl time.Duration) error {
+	if ctx == nil || store == nil || ttl <= 0 {
+		return ErrInvalidAcknowledgement
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	normalized, err := NormalizeAcknowledgement(acknowledgement)
+	if err != nil {
+		return err
+	}
+	now, err := store.currentTime()
+	if err != nil {
+		return err
+	}
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	session, exists := store.sessions[normalized.WorkerID]
+	if !exists || !session.expiresAt.After(now) || session.value.SessionID != normalized.SessionID {
+		return ErrStaleSession
+	}
+	if session.value.AckedSnapshotVersion != 0 {
+		acknowledged := sessionReference(session.value)
+		switch compareSnapshotTuple(normalized.Reference, acknowledged) {
+		case -1:
+			return ErrStaleAcknowledgement
+		case 0:
+			if normalized.Reference.Checksum != acknowledged.Checksum {
+				return ErrSnapshotMismatch
+			}
+		}
+	}
+	reference, exists := store.references[normalized.WorkerID]
+	if !exists || !reference.expiresAt.After(now) {
+		return ErrSnapshotMismatch
+	}
+	switch compareSnapshotTuple(normalized.Reference, reference.value) {
+	case -1:
+		return ErrStaleAcknowledgement
+	case 1:
+		return ErrSnapshotMismatch
+	}
+	if normalized.Reference.Checksum != reference.value.Checksum {
+		return ErrSnapshotMismatch
+	}
+	if !normalized.Applied {
+		delete(store.reports, normalized.WorkerID)
+		session.value.RuntimeEnabled = false
+		session.expiresAt = now.Add(ttl)
+		store.sessions[normalized.WorkerID] = session
+		return nil
+	}
+	if session.value.AckedSnapshotVersion != 0 && compareSnapshotTuple(normalized.Reference, sessionReference(session.value)) == 0 {
+		if !session.value.RuntimeEnabled {
+			delete(store.reports, normalized.WorkerID)
+			session.value.RuntimeEnabled = true
+		}
+		session.expiresAt = now.Add(ttl)
+		store.sessions[normalized.WorkerID] = session
+		return nil
+	}
+	delete(store.reports, normalized.WorkerID)
+	session.value.AckedSnapshotVersion = normalized.Reference.Version
+	session.value.AckedOwnershipEpoch = normalized.Reference.OwnershipEpoch
+	session.value.AckedChecksum = normalized.Reference.Checksum
+	session.value.RuntimeEnabled = true
+	session.expiresAt = now.Add(ttl)
+	store.sessions[normalized.WorkerID] = session
+	return nil
+}
+
+// ReplaceSession is retained temporarily for the pre-control-plane adapters.
+func (store *MemoryStore) ReplaceSession(ctx context.Context, session Session, ttl time.Duration) error {
+	if ctx == nil || store == nil || !validLegacySession(session) || ttl <= 0 {
+		return ErrInvalidSession
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	now, err := store.currentTime()
+	if err != nil {
+		return err
 	}
 	store.mu.Lock()
 	defer store.mu.Unlock()
 	current, exists := store.sessions[session.WorkerID]
 	identityChanged := exists && (current.value.SessionID != session.SessionID || current.value.InstanceID != session.InstanceID)
 	expired := exists && !current.expiresAt.After(now)
-	if exists && !identityChanged && !expired && sessionBefore(session, current.value) {
+	if exists && !identityChanged && !expired && legacySessionBefore(session, current.value) {
 		return ErrStaleSession
 	}
-	ackAdvanced := exists && !identityChanged && !expired && sessionAfter(session, current.value)
+	ackAdvanced := exists && !identityChanged && !expired && legacySessionAfter(session, current.value)
 	if identityChanged || expired || ackAdvanced {
 		delete(store.reports, session.WorkerID)
 	}
+	session.RuntimeEnabled = true
 	store.sessions[session.WorkerID] = memorySession{value: session, expiresAt: now.Add(ttl)}
 	return nil
 }
@@ -78,44 +234,48 @@ func (store *MemoryStore) ReplaceRuntime(ctx context.Context, report Report, ttl
 	if err := ctx.Err(); err != nil {
 		return err
 	}
-	normalized, counterIndex, err := normalizeReport(report)
+	normalized, digest, err := NormalizeReport(report)
 	if err != nil {
 		return err
 	}
-	payload, err := json.Marshal(normalized)
+	now, err := store.currentTime()
 	if err != nil {
-		return ErrInvalidReport
+		return err
-	}
-	digest := sha256.Sum256(payload)
-	now := store.now().UTC()
-	if now.IsZero() {
-		return ErrInvalidStore
 	}
 	store.mu.Lock()
 	defer store.mu.Unlock()
-	session, exists := store.sessions[report.WorkerID]
+	session, exists := store.sessions[normalized.WorkerID]
-	if !exists || !session.expiresAt.After(now) || session.value.SessionID != report.SessionID {
+	if !exists || !session.expiresAt.After(now) || session.value.SessionID != normalized.SessionID {
 		return ErrStaleSession
 	}
-	if report.SnapshotVersion != session.value.AckedSnapshotVersion ||
+	if !session.value.RuntimeEnabled || normalized.SnapshotVersion != session.value.AckedSnapshotVersion ||
-		report.OwnershipEpoch != session.value.AckedOwnershipEpoch {
+		normalized.OwnershipEpoch != session.value.AckedOwnershipEpoch {
-		return ErrStaleReport
+		return ErrSnapshotMismatch
 	}
-	if current, exists := store.reports[report.WorkerID]; exists && current.value.SessionID == report.SessionID {
+	current, exists := store.reports[normalized.WorkerID]
+	if exists && !current.expiresAt.After(now) {
+		delete(store.reports, normalized.WorkerID)
+		exists = false
+	}
+	if exists && current.value.SessionID == normalized.SessionID {
 		switch {
 		case normalized.Sequence < current.value.Sequence:
 			return ErrStaleReport
 		case normalized.Sequence == current.value.Sequence && digest != current.digest:
 			return ErrConflictingReport
 		case normalized.Sequence == current.value.Sequence:
+			current.expiresAt = now.Add(ttl)
+			store.reports[normalized.WorkerID] = current
+			session.expiresAt = now.Add(ttl)
+			store.sessions[normalized.WorkerID] = session
 			return nil
 		}
 	}
-	store.reports[report.WorkerID] = memoryReport{
+	store.reports[normalized.WorkerID] = memoryReport{
-		value: normalized, digest: digest, expiresAt: now.Add(ttl), counters: counterIndex,
+		value: normalized, digest: digest, expiresAt: now.Add(ttl), counters: countersByProxy(normalized.Counters),
 	}
 	session.expiresAt = now.Add(ttl)
-	store.sessions[report.WorkerID] = session
+	store.sessions[normalized.WorkerID] = session
 	return nil
 }
 
@@ -128,7 +288,7 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
 	}
 	seen := make(map[string]struct{}, len(proxies))
 	for _, proxy := range proxies {
-		if !clean(proxy.ProxyID) || !clean(proxy.WorkerID) || proxy.OwnershipEpoch == 0 {
+		if !ValidIdentifier(proxy.ProxyID) || !ValidIdentifier(proxy.WorkerID) || proxy.OwnershipEpoch == 0 {
 			return nil, ErrInvalidQuery
 		}
 		key := proxy.WorkerID + "\x00" + proxy.ProxyID
@@ -137,9 +297,9 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
 		}
 		seen[key] = struct{}{}
 	}
-	now := store.now().UTC()
+	now, err := store.currentTime()
-	if now.IsZero() {
+	if err != nil {
-		return nil, ErrInvalidStore
+		return nil, err
 	}
 	store.mu.Lock()
 	defer store.mu.Unlock()
@@ -149,7 +309,7 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
 		session, sessionExists := store.sessions[proxy.WorkerID]
 		report, reportExists := store.reports[proxy.WorkerID]
 		if !sessionExists || !reportExists || !session.expiresAt.After(now) || !report.expiresAt.After(now) ||
-			report.value.SessionID != session.value.SessionID ||
+			!session.value.RuntimeEnabled || report.value.SessionID != session.value.SessionID ||
 			report.value.SnapshotVersion != session.value.AckedSnapshotVersion ||
 			report.value.OwnershipEpoch != session.value.AckedOwnershipEpoch ||
 			report.value.OwnershipEpoch < proxy.OwnershipEpoch {
@@ -165,47 +325,31 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
 	return result, nil
 }
 
-func normalizeReport(report Report) (Report, map[string]Counter, error) {
+func (store *MemoryStore) currentTime() (time.Time, error) {
-	if !clean(report.WorkerID) || !clean(report.SessionID) || report.Sequence == 0 ||
+	now := store.now().UTC()
-		report.SnapshotVersion == 0 || report.OwnershipEpoch == 0 || report.ObservedAt.IsZero() {
+	if now.IsZero() {
-		return Report{}, nil, ErrInvalidReport
+		return time.Time{}, ErrInvalidStore
 	}
-	normalized := report
+	return now, nil
-	normalized.ObservedAt = report.ObservedAt.UTC()
-	normalized.Counters = append([]Counter(nil), report.Counters...)
-	sort.Slice(normalized.Counters, func(left, right int) bool {
-		return normalized.Counters[left].ProxyID < normalized.Counters[right].ProxyID
-	})
-	index := make(map[string]Counter, len(normalized.Counters))
-	for _, counter := range normalized.Counters {
-		if !clean(counter.ProxyID) || counter.Active < 0 || counter.Reserved < 0 {
-			return Report{}, nil, ErrInvalidReport
-		}
-		if _, exists := index[counter.ProxyID]; exists {
-			return Report{}, nil, ErrInvalidReport
-		}
-		index[counter.ProxyID] = counter
-	}
-	return normalized, index, nil
 }
 
-func validSession(session Session) bool {
+func countersByProxy(counters []Counter) map[string]Counter {
-	return clean(session.WorkerID) && clean(session.InstanceID) && clean(session.SessionID) &&
+	indexed := make(map[string]Counter, len(counters))
+	for _, counter := range counters {
+		indexed[counter.ProxyID] = counter
+	}
+	return indexed
+}
+
+func validLegacySession(session Session) bool {
+	return ValidIdentifier(session.WorkerID) && ValidIdentifier(session.InstanceID) && ValidIdentifier(session.SessionID) &&
 		session.AckedSnapshotVersion > 0 && session.AckedOwnershipEpoch > 0
 }
 
-func sessionBefore(left, right Session) bool {
+func legacySessionBefore(left, right Session) bool {
-	return left.AckedOwnershipEpoch < right.AckedOwnershipEpoch ||
+	return compareSnapshotTuple(sessionReference(left), sessionReference(right)) < 0
-		(left.AckedOwnershipEpoch == right.AckedOwnershipEpoch &&
-			left.AckedSnapshotVersion < right.AckedSnapshotVersion)
 }
 
-func sessionAfter(left, right Session) bool {
+func legacySessionAfter(left, right Session) bool {
-	return left.AckedOwnershipEpoch > right.AckedOwnershipEpoch ||
+	return compareSnapshotTuple(sessionReference(left), sessionReference(right)) > 0
-		(left.AckedOwnershipEpoch == right.AckedOwnershipEpoch &&
-			left.AckedSnapshotVersion > right.AckedSnapshotVersion)
-}
-
-func clean(value string) bool {
-	return value != "" && strings.TrimSpace(value) == value
 }

internal/domain/workerruntime/memory_test.go

@@ -2,11 +2,125 @@ package workerruntime
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
 	"testing"
 	"time"
 )
 
+func TestMemoryStoreRequiresAcknowledgedSnapshotForRuntime(t *testing.T) {
+	now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
+	store := newRuntimeStore(t, &now)
+	ctx := context.Background()
+	openControlSession(t, store, "session-a", time.Minute)
+	epoch, err := store.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		t.Fatalf("CurrentOwnershipEpoch(): %v", err)
+	}
+	report := controlReport(now, "session-a", 1, 7, epoch)
+	if err := store.ReplaceRuntime(ctx, report, time.Minute); !errors.Is(err, ErrSnapshotMismatch) {
+		t.Fatalf("ReplaceRuntime(before ACK) error = %v, want ErrSnapshotMismatch", err)
+	}
+	reference := controlReference(7, epoch, "snapshot-7")
+	if err := store.RecordIssuedSnapshot(ctx, reference, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(): %v", err)
+	}
+	if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{
+		WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true,
+	}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(): %v", err)
+	}
+	if err := store.ReplaceRuntime(ctx, report, time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(after ACK): %v", err)
+	}
+}
+
+func TestMemoryStoreNegativeAcknowledgementFencesDelayedRuntime(t *testing.T) {
+	now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
+	store := newRuntimeStore(t, &now)
+	ctx := context.Background()
+	openControlSession(t, store, "session-a", time.Minute)
+	epoch, err := store.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		t.Fatalf("CurrentOwnershipEpoch(): %v", err)
+	}
+	first := controlReference(7, epoch, "snapshot-7")
+	if err := store.RecordIssuedSnapshot(ctx, first, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(first): %v", err)
+	}
+	if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: first, Applied: true}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(first): %v", err)
+	}
+	if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 1, 7, epoch), time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(first): %v", err)
+	}
+
+	second := controlReference(8, epoch, "snapshot-8")
+	if err := store.RecordIssuedSnapshot(ctx, second, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(second): %v", err)
+	}
+	if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: second, Applied: false, ErrorCode: "apply_failed"}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(negative): %v", err)
+	}
+	if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 2, 7, epoch), time.Minute); !errors.Is(err, ErrSnapshotMismatch) {
+		t.Fatalf("ReplaceRuntime(delayed): %v, want ErrSnapshotMismatch", err)
+	}
+	if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: second, Applied: true}, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(recover): %v", err)
+	}
+	if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 2, 8, epoch), time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(recovered): %v", err)
+	}
+}
+
+func TestMemoryStoreAcknowledgementReplayPreservesRuntimeFence(t *testing.T) {
+	now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
+	store := newRuntimeStore(t, &now)
+	ctx := context.Background()
+	openControlSession(t, store, "session-a", time.Minute)
+	epoch, err := store.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		t.Fatalf("CurrentOwnershipEpoch(): %v", err)
+	}
+	reference := controlReference(7, epoch, "snapshot-7")
+	if err := store.RecordIssuedSnapshot(ctx, reference, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(): %v", err)
+	}
+	ack := SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true}
+	if err := store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(first): %v", err)
+	}
+	if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 100, 7, epoch), time.Minute); err != nil {
+		t.Fatalf("ReplaceRuntime(): %v", err)
+	}
+	if err := store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
+		t.Fatalf("AcknowledgeSnapshot(replay): %v", err)
+	}
+	got, err := store.ReadRuntime(ctx, []OwnedProxy{{ProxyID: "proxy-a", WorkerID: "worker-a", OwnershipEpoch: epoch}})
+	if err != nil || len(got) != 1 || !got[0].Fresh {
+		t.Fatalf("ReadRuntime(after ACK replay) = %+v, %v", got, err)
+	}
+	if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 99, 7, epoch), time.Minute); !errors.Is(err, ErrStaleReport) {
+		t.Fatalf("ReplaceRuntime(stale): %v, want ErrStaleReport", err)
+	}
+}
+
+func TestMemoryStoreRejectsConflictingSnapshotReference(t *testing.T) {
+	now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
+	store := newRuntimeStore(t, &now)
+	ctx := context.Background()
+	epoch, err := store.CurrentOwnershipEpoch(ctx)
+	if err != nil {
+		t.Fatalf("CurrentOwnershipEpoch(): %v", err)
+	}
+	if err := store.RecordIssuedSnapshot(ctx, controlReference(7, epoch, "first"), time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(first): %v", err)
+	}
+	if err := store.RecordIssuedSnapshot(ctx, controlReference(7, epoch, "second"), time.Minute); !errors.Is(err, ErrConflictingSnapshotReference) {
+		t.Fatalf("RecordIssuedSnapshot(conflict): %v, want ErrConflictingSnapshotReference", err)
+	}
+}
+
 func TestMemoryStoreReplacesSparseRuntimeAndClearsMissingCounters(t *testing.T) {
 	now := time.Date(2026, 7, 30, 15, 0, 0, 0, time.UTC)
 	store := newRuntimeStore(t, &now)
@@ -103,13 +217,13 @@ func TestMemoryStoreRejectsRuntimeBeyondAcknowledgedSnapshot(t *testing.T) {
 		WorkerID: "worker-a", SessionID: "session-a", Sequence: 1,
 		SnapshotVersion: 4, OwnershipEpoch: 9, ObservedAt: now,
 	}
-	if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrStaleReport) {
+	if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrSnapshotMismatch) {
-		t.Fatalf("ReplaceRuntime(ahead snapshot) error = %v, want ErrStaleReport", err)
+		t.Fatalf("ReplaceRuntime(ahead snapshot) error = %v, want ErrSnapshotMismatch", err)
 	}
 	report.SnapshotVersion = 3
 	report.OwnershipEpoch = 10
-	if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrStaleReport) {
+	if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrSnapshotMismatch) {
-		t.Fatalf("ReplaceRuntime(ahead epoch) error = %v, want ErrStaleReport", err)
+		t.Fatalf("ReplaceRuntime(ahead epoch) error = %v, want ErrSnapshotMismatch", err)
 	}
 }
 
@@ -154,3 +268,26 @@ func registerRuntimeSession(t *testing.T, store *MemoryStore, sessionID string,
 		t.Fatalf("ReplaceSession(): %v", err)
 	}
 }
+
+func openControlSession(t *testing.T, store *MemoryStore, sessionID string, ttl time.Duration) {
+	t.Helper()
+	if err := store.OpenSession(context.Background(), Session{
+		WorkerID: "worker-a", InstanceID: "instance-a", SessionID: sessionID, Zone: "zone-a", ProtocolVersion: 1,
+		Labels: map[string]string{"region": "test"},
+	}, ttl); err != nil {
+		t.Fatalf("OpenSession(): %v", err)
+	}
+}
+
+func controlReference(version, epoch uint64, content string) SnapshotReference {
+	return SnapshotReference{
+		WorkerID: "worker-a", Version: version, OwnershipEpoch: epoch, Checksum: sha256.Sum256([]byte(content)),
+	}
+}
+
+func controlReport(now time.Time, sessionID string, sequence, version, epoch uint64) Report {
+	return Report{
+		WorkerID: "worker-a", SessionID: sessionID, Sequence: sequence, SnapshotVersion: version,
+		OwnershipEpoch: epoch, ObservedAt: now, Counters: []Counter{{ProxyID: "proxy-a", Active: 1}},
+	}
+}

internal/domain/workerruntime/runtime.go

@@ -2,26 +2,53 @@ package workerruntime
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
 	"time"
 )
 
 var (
-	ErrInvalidStore      = errors.New("invalid worker runtime store")
+	ErrInvalidStore                 = errors.New("invalid worker runtime store")
-	ErrInvalidSession    = errors.New("invalid worker runtime session")
+	ErrInvalidSession               = errors.New("invalid worker runtime session")
-	ErrInvalidReport     = errors.New("invalid worker runtime report")
+	ErrInvalidReport                = errors.New("invalid worker runtime report")
-	ErrInvalidQuery      = errors.New("invalid worker runtime query")
+	ErrInvalidQuery                 = errors.New("invalid worker runtime query")
-	ErrStaleSession      = errors.New("stale worker runtime session")
+	ErrStaleSession                 = errors.New("stale worker runtime session")
-	ErrStaleReport       = errors.New("stale worker runtime report")
+	ErrStaleReport                  = errors.New("stale worker runtime report")
-	ErrConflictingReport = errors.New("conflicting worker runtime report")
+	ErrConflictingReport            = errors.New("conflicting worker runtime report")
+	ErrInvalidSnapshotReference     = errors.New("invalid worker snapshot reference")
+	ErrInvalidAcknowledgement       = errors.New("invalid worker snapshot acknowledgement")
+	ErrSnapshotMismatch             = errors.New("worker snapshot does not match acknowledged state")
+	ErrStaleSnapshotReference       = errors.New("stale worker snapshot reference")
+	ErrConflictingSnapshotReference = errors.New("conflicting worker snapshot reference")
+	ErrStaleAcknowledgement         = errors.New("stale worker snapshot acknowledgement")
 )
 
 type Session struct {
 	WorkerID             string
 	InstanceID           string
 	SessionID            string
+	Zone                 string
+	ProtocolVersion      uint32
+	Labels               map[string]string
 	AckedSnapshotVersion uint64
 	AckedOwnershipEpoch  uint64
+	AckedChecksum        [sha256.Size]byte
+	RuntimeEnabled       bool
+}
+
+type SnapshotReference struct {
+	WorkerID       string
+	Version        uint64
+	OwnershipEpoch uint64
+	Checksum       [sha256.Size]byte
+}
+
+type SnapshotAcknowledgement struct {
+	WorkerID  string
+	SessionID string
+	Reference SnapshotReference
+	Applied   bool
+	ErrorCode string
 }
 
 type Counter struct {
@@ -61,6 +88,14 @@ type SessionWriter interface {
 	ReplaceSession(context.Context, Session, time.Duration) error
 }
 
+type ControlStore interface {
+	CurrentOwnershipEpoch(context.Context) (uint64, error)
+	OpenSession(context.Context, Session, time.Duration) error
+	RecordIssuedSnapshot(context.Context, SnapshotReference, time.Duration) error
+	AcknowledgeSnapshot(context.Context, SnapshotAcknowledgement, time.Duration) error
+	ReplaceRuntime(context.Context, Report, time.Duration) error
+}
+
 type ReportWriter interface {
 	ReplaceRuntime(context.Context, Report, time.Duration) error
 }

internal/domain/workerruntime/validation.go

@@ -0,0 +1,138 @@
+package workerruntime
+
+import (
+	"crypto/sha256"
+	"encoding/json"
+	"regexp"
+	"sort"
+	"strings"
+)
+
+const (
+	maximumLabels          = 32
+	maximumLabelKeyBytes   = 64
+	maximumLabelValueBytes = 256
+	maximumLabelTotalBytes = 4 << 10
+)
+
+var identifierPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`)
+
+// ValidIdentifier accepts stable Worker, Session and Proxy identifiers.
+func ValidIdentifier(value string) bool {
+	return identifierPattern.MatchString(value)
+}
+
+// NormalizeLabels validates and deep-copies the bounded Worker label set.
+func NormalizeLabels(labels map[string]string) (map[string]string, error) {
+	if len(labels) > maximumLabels {
+		return nil, ErrInvalidSession
+	}
+	normalized := make(map[string]string, len(labels))
+	total := 0
+	for key, value := range labels {
+		if !ValidIdentifier(key) || len(key) > maximumLabelKeyBytes || value == "" ||
+			strings.TrimSpace(value) != value || strings.IndexByte(value, 0) >= 0 ||
+			len(value) > maximumLabelValueBytes {
+			return nil, ErrInvalidSession
+		}
+		total += len(key) + len(value)
+		if total > maximumLabelTotalBytes {
+			return nil, ErrInvalidSession
+		}
+		normalized[key] = value
+	}
+	return normalized, nil
+}
+
+// NormalizeSession prepares a new, not-yet-acknowledged session for storage.
+func NormalizeSession(session Session) (Session, error) {
+	if !ValidIdentifier(session.WorkerID) || !ValidIdentifier(session.InstanceID) ||
+		!ValidIdentifier(session.SessionID) || !ValidIdentifier(session.Zone) ||
+		session.ProtocolVersion == 0 || session.AckedSnapshotVersion != 0 ||
+		session.AckedOwnershipEpoch != 0 || !checksumIsZero(session.AckedChecksum) ||
+		session.RuntimeEnabled {
+		return Session{}, ErrInvalidSession
+	}
+	labels, err := NormalizeLabels(session.Labels)
+	if err != nil {
+		return Session{}, err
+	}
+	session.Labels = labels
+	return session, nil
+}
+
+func NormalizeSnapshotReference(reference SnapshotReference) (SnapshotReference, error) {
+	if !ValidIdentifier(reference.WorkerID) || reference.Version == 0 || reference.OwnershipEpoch == 0 ||
+		checksumIsZero(reference.Checksum) {
+		return SnapshotReference{}, ErrInvalidSnapshotReference
+	}
+	return reference, nil
+}
+
+func NormalizeAcknowledgement(acknowledgement SnapshotAcknowledgement) (SnapshotAcknowledgement, error) {
+	if !ValidIdentifier(acknowledgement.WorkerID) || !ValidIdentifier(acknowledgement.SessionID) ||
+		(acknowledgement.ErrorCode != "" && !ValidIdentifier(acknowledgement.ErrorCode)) {
+		return SnapshotAcknowledgement{}, ErrInvalidAcknowledgement
+	}
+	reference, err := NormalizeSnapshotReference(acknowledgement.Reference)
+	if err != nil || reference.WorkerID != acknowledgement.WorkerID {
+		return SnapshotAcknowledgement{}, ErrInvalidAcknowledgement
+	}
+	acknowledgement.Reference = reference
+	return acknowledgement, nil
+}
+
+// NormalizeReport returns the canonical sparse replacement and its digest.
+func NormalizeReport(report Report) (Report, [sha256.Size]byte, error) {
+	if !ValidIdentifier(report.WorkerID) || !ValidIdentifier(report.SessionID) || report.Sequence == 0 ||
+		report.SnapshotVersion == 0 || report.OwnershipEpoch == 0 || report.ObservedAt.IsZero() {
+		return Report{}, [sha256.Size]byte{}, ErrInvalidReport
+	}
+	normalized := report
+	normalized.ObservedAt = report.ObservedAt.UTC()
+	normalized.Counters = append([]Counter(nil), report.Counters...)
+	sort.Slice(normalized.Counters, func(left, right int) bool {
+		return normalized.Counters[left].ProxyID < normalized.Counters[right].ProxyID
+	})
+	seen := make(map[string]struct{}, len(normalized.Counters))
+	for _, counter := range normalized.Counters {
+		if !ValidIdentifier(counter.ProxyID) || counter.Active < 0 || counter.Reserved < 0 {
+			return Report{}, [sha256.Size]byte{}, ErrInvalidReport
+		}
+		if _, exists := seen[counter.ProxyID]; exists {
+			return Report{}, [sha256.Size]byte{}, ErrInvalidReport
+		}
+		seen[counter.ProxyID] = struct{}{}
+	}
+	payload, err := json.Marshal(normalized)
+	if err != nil {
+		return Report{}, [sha256.Size]byte{}, ErrInvalidReport
+	}
+	return normalized, sha256.Sum256(payload), nil
+}
+
+func checksumIsZero(checksum [sha256.Size]byte) bool {
+	return checksum == [sha256.Size]byte{}
+}
+
+func compareSnapshotTuple(left, right SnapshotReference) int {
+	switch {
+	case left.OwnershipEpoch < right.OwnershipEpoch:
+		return -1
+	case left.OwnershipEpoch > right.OwnershipEpoch:
+		return 1
+	case left.Version < right.Version:
+		return -1
+	case left.Version > right.Version:
+		return 1
+	default:
+		return 0
+	}
+}
+
+func sessionReference(session Session) SnapshotReference {
+	return SnapshotReference{
+		WorkerID: session.WorkerID, Version: session.AckedSnapshotVersion,
+		OwnershipEpoch: session.AckedOwnershipEpoch, Checksum: session.AckedChecksum,
+	}
+}

internal/domain/workerruntime/validation_test.go

@@ -0,0 +1,63 @@
+package workerruntime
+
+import (
+	"crypto/sha256"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestNormalizeLabelsClonesAndBoundsValues(t *testing.T) {
+	source := map[string]string{"region": "cn-north"}
+	labels, err := NormalizeLabels(source)
+	if err != nil {
+		t.Fatalf("NormalizeLabels(): %v", err)
+	}
+	labels["region"] = "changed"
+	if source["region"] != "cn-north" {
+		t.Fatal("NormalizeLabels() aliases the source map")
+	}
+	if !ValidIdentifier("worker-a:1") || ValidIdentifier("worker a") || ValidIdentifier("") {
+		t.Fatal("ValidIdentifier() accepted or rejected an invalid value")
+	}
+	if _, err := NormalizeLabels(map[string]string{" region": "cn"}); !errors.Is(err, ErrInvalidSession) {
+		t.Fatalf("NormalizeLabels(invalid key) error = %v", err)
+	}
+}
+
+func TestNormalizeReportUsesStableCounterOrdering(t *testing.T) {
+	now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.FixedZone("UTC+8", 8*60*60))
+	base := Report{
+		WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, SnapshotVersion: 7, OwnershipEpoch: 3,
+		ObservedAt: now, Counters: []Counter{{ProxyID: "proxy-b", Active: 2}, {ProxyID: "proxy-a", Reserved: 1}},
+	}
+	normalized, digest, err := NormalizeReport(base)
+	if err != nil || normalized.ObservedAt.Location() != time.UTC || normalized.Counters[0].ProxyID != "proxy-a" {
+		t.Fatalf("NormalizeReport() = %+v, %x, %v", normalized, digest, err)
+	}
+	base.Counters[0], base.Counters[1] = base.Counters[1], base.Counters[0]
+	_, replayDigest, err := NormalizeReport(base)
+	if err != nil || digest != replayDigest {
+		t.Fatalf("NormalizeReport(reordered) digest = %x, %v; want %x", replayDigest, err, digest)
+	}
+	if _, _, err := NormalizeReport(Report{}); !errors.Is(err, ErrInvalidReport) {
+		t.Fatalf("NormalizeReport(invalid) error = %v", err)
+	}
+}
+
+func TestNormalizeSnapshotReferenceAndAcknowledgement(t *testing.T) {
+	reference := SnapshotReference{
+		WorkerID: "worker-a", Version: 7, OwnershipEpoch: 3, Checksum: sha256.Sum256([]byte("snapshot")),
+	}
+	if _, err := NormalizeSnapshotReference(reference); err != nil {
+		t.Fatalf("NormalizeSnapshotReference(): %v", err)
+	}
+	acknowledgement := SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: reference, ErrorCode: "apply_failed"}
+	if _, err := NormalizeAcknowledgement(acknowledgement); err != nil {
+		t.Fatalf("NormalizeAcknowledgement(): %v", err)
+	}
+	acknowledgement.Reference.WorkerID = "worker-b"
+	if _, err := NormalizeAcknowledgement(acknowledgement); !errors.Is(err, ErrInvalidAcknowledgement) {
+		t.Fatalf("NormalizeAcknowledgement(worker mismatch) error = %v", err)
+	}
+}

scripts/generate-proto.ps1

@@ -0,0 +1,62 @@
+param(
+    [string]$Protoc = "",
+    [string]$IncludePath = $env:PROTOC_INCLUDE,
+    [string]$OutputRoot = ""
+)
+
+$ErrorActionPreference = "Stop"
+
+$repositoryRoot = Split-Path -Parent $PSScriptRoot
+$protoRoot = Join-Path $repositoryRoot "api/proto"
+$source = Join-Path $protoRoot "controlplane/v1/controlplane.proto"
+$toolRoot = Join-Path $repositoryRoot ".tmp-proto/tools"
+New-Item -ItemType Directory -Force -Path $toolRoot | Out-Null
+
+if ([string]::IsNullOrWhiteSpace($Protoc)) {
+    $Protoc = & (Join-Path $PSScriptRoot "install-protoc.ps1")
+}
+
+$runningWindows = [System.Runtime.InteropServices.RuntimeInformation]::IsOSPlatform(
+    [System.Runtime.InteropServices.OSPlatform]::Windows
+)
+$suffix = if ($runningWindows) { ".exe" } else { "" }
+$protocGenGo = Join-Path $toolRoot ("protoc-gen-go" + $suffix)
+$protocGenGoGRPC = Join-Path $toolRoot ("protoc-gen-go-grpc" + $suffix)
+
+go build -o $protocGenGo google.golang.org/protobuf/cmd/protoc-gen-go
+if ($LASTEXITCODE -ne 0) { throw "build protoc-gen-go failed" }
+go build -o $protocGenGoGRPC google.golang.org/grpc/cmd/protoc-gen-go-grpc
+if ($LASTEXITCODE -ne 0) { throw "build protoc-gen-go-grpc failed" }
+
+$protocCommand = Get-Command $Protoc -ErrorAction Stop
+$protocVersion = (& $protocCommand.Source --version).Trim()
+if ($protocVersion -ne "libprotoc 35.0") { throw "protoc 35.0 is required" }
+
+if ([string]::IsNullOrWhiteSpace($IncludePath)) {
+    $installationRoot = Split-Path (Split-Path $protocCommand.Source -Parent) -Parent
+    $IncludePath = @(
+        (Join-Path $installationRoot "include"),
+        "/usr/include",
+        "/usr/local/include"
+    ) |
+        Where-Object { Test-Path (Join-Path $_ "google/protobuf/timestamp.proto") } |
+        Select-Object -First 1
+}
+if ([string]::IsNullOrWhiteSpace($OutputRoot)) { $OutputRoot = $repositoryRoot }
+if ([string]::IsNullOrWhiteSpace($IncludePath) -or
+    -not (Test-Path (Join-Path $IncludePath "google/protobuf/timestamp.proto"))) {
+    throw "protoc well-known type include directory was not found"
+}
+New-Item -ItemType Directory -Force -Path $OutputRoot | Out-Null
+
+& $protocCommand.Source `
+    "--proto_path=$protoRoot" `
+    "--proto_path=$IncludePath" `
+    "--plugin=protoc-gen-go=$protocGenGo" `
+    "--plugin=protoc-gen-go-grpc=$protocGenGoGRPC" `
+    "--go_out=$OutputRoot" `
+    "--go_opt=module=proxy-pool" `
+    "--go-grpc_out=$OutputRoot" `
+    "--go-grpc_opt=module=proxy-pool" `
+    $source
+if ($LASTEXITCODE -ne 0) { throw "protobuf Go generation failed" }

scripts/install-protoc.ps1

@@ -0,0 +1,61 @@
+param()
+
+$ErrorActionPreference = "Stop"
+
+$repositoryRoot = Split-Path -Parent $PSScriptRoot
+$runtime = [System.Runtime.InteropServices.RuntimeInformation]
+$runningWindows = $runtime::IsOSPlatform([System.Runtime.InteropServices.OSPlatform]::Windows)
+$runningLinux = $runtime::IsOSPlatform([System.Runtime.InteropServices.OSPlatform]::Linux)
+
+if ($runtime::OSArchitecture -ne [System.Runtime.InteropServices.Architecture]::X64) {
+    throw "protoc bootstrap supports x64 only"
+}
+
+if ($runningWindows) {
+    $asset = "protoc-35.0-win64.zip"
+    $expectedHash = "d1cede9e308cc3eb072392af1c02ccae4bdd3d2f374ec2970dbd8cdfdaa91363"
+    $executableName = "protoc.exe"
+}
+elseif ($runningLinux) {
+    $asset = "protoc-35.0-linux-x86_64.zip"
+    $expectedHash = "a45cda0989c17dd950db55f6fbe1e5814c50fda08e87aa422980ac1f89dddbbc"
+    $executableName = "protoc"
+}
+else {
+    throw "protoc bootstrap supports Windows x64 and Linux x64 only"
+}
+
+$cacheDirectory = Join-Path $repositoryRoot ".tmp-proto/downloads"
+$installationDirectory = Join-Path $repositoryRoot ".tmp-proto/protoc-35.0"
+$archivePath = Join-Path $cacheDirectory $asset
+$downloadURL = "https://github.com/protocolbuffers/protobuf/releases/download/v35.0/$asset"
+New-Item -ItemType Directory -Force -Path $cacheDirectory | Out-Null
+
+$needsDownload = -not (Test-Path -LiteralPath $archivePath)
+if (-not $needsDownload) {
+    $actualHash = (Get-FileHash -LiteralPath $archivePath -Algorithm SHA256).Hash.ToLowerInvariant()
+    $needsDownload = $actualHash -ne $expectedHash
+}
+if ($needsDownload) {
+    Invoke-WebRequest -Uri $downloadURL -OutFile $archivePath
+}
+
+$actualHash = (Get-FileHash -LiteralPath $archivePath -Algorithm SHA256).Hash.ToLowerInvariant()
+if ($actualHash -ne $expectedHash) {
+    throw "protoc archive SHA-256 mismatch"
+}
+
+New-Item -ItemType Directory -Force -Path $installationDirectory | Out-Null
+Expand-Archive -LiteralPath $archivePath -DestinationPath $installationDirectory -Force
+
+$protocPath = Join-Path $installationDirectory "bin/$executableName"
+if (-not (Test-Path -LiteralPath $protocPath)) {
+    throw "protoc executable was not extracted"
+}
+
+$version = (& $protocPath --version).Trim()
+if ($version -ne "libprotoc 35.0") {
+    throw "protoc version mismatch: $version"
+}
+
+(Resolve-Path -LiteralPath $protocPath).Path

scripts/verify-proto.ps1

@@ -1,5 +1,5 @@
 param(
-    [string]$Protoc = "protoc",
+    [string]$Protoc = "",
     [string]$IncludePath = $env:PROTOC_INCLUDE,
     [string]$OutputPath = ""
 )
@@ -10,7 +10,14 @@ $repositoryRoot = Split-Path -Parent $PSScriptRoot
 $protoRoot = Join-Path $repositoryRoot "api/proto"
 $source = Join-Path $protoRoot "controlplane/v1/controlplane.proto"
 
+if ([string]::IsNullOrWhiteSpace($Protoc)) {
+    $Protoc = & (Join-Path $PSScriptRoot "install-protoc.ps1")
+}
 $protocCommand = Get-Command $Protoc -ErrorAction Stop
+$protocVersion = (& $protocCommand.Source --version).Trim()
+if ($protocVersion -ne "libprotoc 35.0") {
+    throw "protoc 35.0 is required"
+}
 if ([string]::IsNullOrWhiteSpace($IncludePath)) {
     $installationRoot = Split-Path (Split-Path $protocCommand.Source -Parent) -Parent
     $candidates = @(
@@ -49,3 +56,25 @@ if ($descriptor.Length -le 0) {
     throw "protoc produced an empty descriptor set"
 }
 Write-Host "descriptor: $($descriptor.FullName) ($($descriptor.Length) bytes)"
+
+$outputRoot = Join-Path $repositoryRoot ".tmp-proto/generated"
+& (Join-Path $PSScriptRoot "generate-proto.ps1") `
+    -Protoc $protocCommand.Source `
+    -IncludePath $IncludePath `
+    -OutputRoot $outputRoot
+
+$generated = @(
+    "gen/controlplane/v1/controlplane.pb.go",
+    "gen/controlplane/v1/controlplane_grpc.pb.go"
+)
+foreach ($relative in $generated) {
+    $committed = Join-Path $repositoryRoot $relative
+    $candidate = Join-Path $outputRoot $relative
+    if (-not (Test-Path -LiteralPath $committed) -or
+        -not (Test-Path -LiteralPath $candidate) -or
+        -not [System.Linq.Enumerable]::SequenceEqual(
+            [System.IO.File]::ReadAllBytes($committed),
+            [System.IO.File]::ReadAllBytes($candidate))) {
+        throw "generated protobuf drift: $relative"
+    }
+}

scripts/verify.ps1

@@ -22,12 +22,7 @@ if ($unformatted) {
 Invoke-Step "go vet" { go vet ./... }
 Invoke-Step "unit tests" { go test -timeout 60s ./... }
 
-if (Get-Command protoc -ErrorAction SilentlyContinue) {
+Invoke-Step "protobuf contracts" { & (Join-Path $PSScriptRoot "verify-proto.ps1") }
-    Invoke-Step "protobuf descriptor" { & (Join-Path $PSScriptRoot "verify-proto.ps1") }
-}
-else {
-    Write-Host "==> protobuf descriptor skipped: protoc is not installed"
-}
 
 if ((go env CGO_ENABLED) -eq "1") {
     Invoke-Step "race tests" { go test -race -timeout 60s ./internal/... }