README.md

@@ -71,18 +71,18 @@ flowchart LR
 
 ## 当前完成度
 
-截至 **2026-07-31**，实施计划检查项为 **53 / 74（71.6%）**。详情见
+截至 **2026-07-30**，实施计划检查项为 **52 / 74（70.3%）**。详情见
 [实施计划](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 契约。
-- **待完成**：Worker Snapshot 下发流、Gateway 快照客户端、Outcome 上报、Checker
+- **待完成**：WorkerControlPlane gRPC session/snapshot/ACK 闭环、Checker 调度与
-  调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性集群压测。
+  健康状态链、Gateway 进程与快照客户端、完整 Routing 运行链，以及 loadgen 和
+  代表性集群压测。
 
 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表
 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。

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

@@ -66,7 +66,6 @@ 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)
 ```
 
@@ -104,8 +103,6 @@ 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,23 +12,6 @@ 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.
-- [x] Implement the Worker heartbeat receiving path and session lifecycle.
+- [ ] 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-31）：已实现共享 `platform/httpapi`、Distribution
+当前进度（2026-07-30）：已实现共享 `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,16 +203,11 @@ Distribution/Admin 服务构造、错误合并和资源关闭。生产 Provider
 权威管理状态动态装配 Upstream，并与 HTTP Runtime 通过公用 lifecycle Group 联动
 停机；Admin disable 会取消 Runtime，reload 在提交前预检并在发布后替换运行实例。
 组合 fixture 已验证隔离 Redis namespace 下的选主、Provider HTTP 调用、模板解析
-和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、`/readyz` 与基础
+和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、
-Prometheus 运行时指标，三监听器隔离已通过测试；业务指标仍待
+`/readyz` 与基础 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,11 +46,6 @@
 - `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
@@ -89,11 +84,9 @@ 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 Register/ACK/Runtime gRPC 接收路径、Redis 服务端 TTL 会话生命周期、
+5. Worker heartbeat gRPC 接收路径；Redis Provider Leader、分布式请求额度、
-   SPIFFE mTLS 校验和 Controller 生命周期接线已完成；Redis Provider Leader、
+   Distribution Client 限制和 Provider 状态丢失重建已完成。
-   分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。
+6. Worker 网络快照流；Redis ownership drain/ACK/过期回收已完成。
-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 已运行 Worker Register/ACK/Runtime gRPC 接收，Gateway/Checker/Loadgen 构建产物及 Snapshot 下发待实现 |
+| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 命令已实现，Gateway/Checker/Loadgen 构建产物待实现 |
 | 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 已测试；Worker session/ACK/sequence/TTL/ownership fence 与 WorkerControlPlane Register/ACK/Runtime 接收已通过内存、Redis 8.2 和 gRPC 测试；Snapshot 下发、目标健康和 Gateway reserve 策略仍待完成 |
+| 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-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,10 +19,8 @@ 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"
@@ -61,7 +59,6 @@ type ports struct {
 	coordinator      provider.Coordinator
 	credentials      credentials.Store
 	providerResults  provider.ResultRecorder
-	workerStore      workerruntime.ControlStore
 	close            func() error
 }
 
@@ -77,29 +74,15 @@ 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 runWithWorkerFactory(ctx, options, &productionInfrastructure{
+	return run(ctx, options, &productionInfrastructure{
 		holderID: options.HolderID, namespace: options.RedisNamespace,
-	}, productionRuntimeFactory{}, productionWorkerRuntimeFactory{})
+	}, productionRuntimeFactory{})
 }
 
 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(workerFactory) {
+		nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) {
 		return ErrInvalidOptions
 	}
 	if err := ctx.Err(); err != nil {
@@ -217,7 +200,7 @@ func runWithWorkerFactory(
 		dependencies.MetricsHandler = handler
 	}
 
-	runners := make([]lifecycle.Runner, 0, 3)
+	runners := make([]lifecycle.Runner, 0, 2)
 	if hasHTTPRuntime(loaded.Value) {
 		runner, err := factory.New(configurationStore.Current(), dependencies, controllerRuntime.Options{HTTP: options.HTTP})
 		if err != nil {
@@ -228,29 +211,6 @@ func runWithWorkerFactory(
 		}
 		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 {
@@ -338,9 +298,3 @@ 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,9 +4,7 @@ package bootstrap
 
 import (
 	"context"
-	"crypto/sha256"
 	"errors"
-	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -24,14 +22,7 @@ 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) {
@@ -89,95 +80,6 @@ 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
@@ -304,62 +206,3 @@ 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,12 +12,10 @@ 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"
 )
@@ -172,52 +170,6 @@ 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)
@@ -380,22 +332,6 @@ 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 }
@@ -517,22 +453,3 @@ 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 || configuration.ControlPlane.Enabled {
+	if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled {
 		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: runtimeCounterCapacity(configuration),
+			MaxRuntimeCounters: credentialCapacity(configuration),
 			MaxInventoryScan:   maxInventoryScan(configuration),
 			CleanupLimit:       redisCleanupLimit,
 		})
@@ -131,7 +131,6 @@ 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 {
@@ -204,7 +203,7 @@ func selectMetricsReadiness(
 	configuration *config.Config,
 	admin, activity platformMetrics.ReadinessChecker,
 ) platformMetrics.ReadinessChecker {
-	if configuration.Distribution.Enabled || configuration.ControlPlane.Enabled || hasEnabledUpstream(configuration) {
+	if configuration.Distribution.Enabled || hasEnabledUpstream(configuration) {
 		return activity
 	}
 	if configuration.Admin.Enabled {
@@ -276,14 +275,6 @@ 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,13 +50,6 @@ 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) {
@@ -131,19 +124,6 @@ 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) }
@@ -163,14 +143,6 @@ 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

@@ -1,136 +0,0 @@
-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

@@ -1,103 +0,0 @@
-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

@@ -1,72 +0,0 @@
-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

@@ -1,84 +0,0 @@
-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

@@ -1,169 +0,0 @@
-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

@@ -1,86 +0,0 @@
-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

@@ -1,177 +0,0 @@
-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

@@ -1,120 +0,0 @@
-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)
-	}
-}