feat: wire checker observations into controller

docs/development/implementation-plan.md

@@ -278,8 +278,9 @@ Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄
 Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、AVAILABLE/SUSPECT 的
 阈值降级、精确重放、冲突拒绝与成功恢复。TARGET Profile 也已在 Memory 和 Redis 中独立
 归并，以哈希键保存并随代理 TTL 过期，绝不写入 Proxy 全局状态或选择索引。调度器、
-Controller 到 Checker 的 RPC、任务调度和独立进程尚未接入；Controller 已新增公用 Reducer
-作为 Observation 到原子 Store 的唯一归并边界，因此本任务保持未完成。
+Checker Observation 上报 RPC 已复用既有控制面监听接入 Controller：SPIFFE `checker` 身份、
+每批上限、配置阈值解析、代理归属查询与 Reducer 均已闭环。`StreamCheckTasks` 的租约调度、
+Checker 独立进程与探测器尚未实现，因此本任务保持未完成。
 
 ## Task 12: Machine-readable Contracts
 

docs/requirements/completion-audit.md

@@ -60,7 +60,8 @@ Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controll
 摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer
 与 Memory/Redis 原子状态提交基础；TARGET Profile 以独立、随代理 TTL 过期的 Redis
 记录归并，不改写 Proxy 全局状态。Controller 公用 Reducer 已作为 Observation 的唯一状态
-归并边界；任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
+归并边界，Checker Observation RPC 已在同一控制面监听以独立 SPIFFE 身份接入；任务调度和
+独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
   `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller
   在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用
   `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot
@@ -110,8 +111,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。
 6. Worker 基础网络快照流、Proxy/Gateway Routing/凭据 Snapshot payload、Gateway Snapshot
    客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成；
    Outcome 上报已完成基础观测链；Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。
-7. Checker 调度、探测器和健康 reducer；其中全局与 TARGET Profile 的 Memory/Redis
-   原子归并已完成。
+7. Checker 调度与探测器；全局与 TARGET Profile 的 Memory/Redis 原子归并、Controller
+   Reducer 和 Observation 上报 RPC 已完成。
 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。
 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。
 10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS

docs/requirements/traceability.md

@@ -7,7 +7,7 @@
 
 | ID | 最终需求 | 来源 | 验证证据 |
 |---|---|---|---|
-| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物待实现 |
+| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物与任务流待实现 |
 | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 |
 | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 |
 | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 |
@@ -80,7 +80,7 @@
 | ID | 最终需求 | 来源 | 验证证据 |
 |---|---|---|---|
 | HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 在 Memory 和 Redis 独立、随代理 TTL 归并，不改写 Proxy 全局状态；Routing 消费待实现 |
-| HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置校验已完成；有界调度器、抖动和分级频率测试待实现 |
+| HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置有效合并、URL 校验和 Observation 上报已完成；有界调度器、抖动和分级频率测试待实现 |
 | HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复；任务调度与 REMOVE 编排待实现 |
 | SEC-001 | API 认证与 Proxy 认证分离，Secret 统一脱敏 | 7528-8111, 8904-8945 | Config 脱敏、Provider Store -> SecretRef -> Gateway Resolver 跨包测试与格式化泄漏回归测试 |
 | SEC-002 | 非回环监听无保护时严格模式启动失败 | 8112-8441 | 配置校验测试 |

internal/adapters/redisactivity/scripts.go

@@ -49,6 +49,11 @@ type targetHealthScriptReply struct {
 	Target string       `json:"target,omitempty"`
 }
 
+type upstreamLookupScriptReply struct {
+	Status   scriptStatus `json:"status"`
+	Upstream string       `json:"upstream,omitempty"`
+}
+
 type extractScriptReply struct {
 	Status        scriptStatus `json:"status"`
 	RequestDigest string       `json:"requestDigest"`
@@ -113,6 +118,9 @@ var healthSource string
 //go:embed scripts/target_health.lua
 var targetHealthSource string
 
+//go:embed scripts/upstream_lookup.lua
+var upstreamLookupSource string
+
 //go:embed scripts/extract.lua
 var extractSource string
 
@@ -138,6 +146,7 @@ var (
 	upsertScript         = redis.NewScript(upsertSource)
 	healthScript         = redis.NewScript(healthSource)
 	targetHealthScript   = redis.NewScript(targetHealthSource)
+	upstreamLookupScript = redis.NewScript(upstreamLookupSource)
 	extractScript        = redis.NewScript(extractSource)
 	ownershipScript      = redis.NewScript(ownershipSource)
 	sweepScript          = redis.NewScript(sweepSource)

internal/adapters/redisactivity/scripts/upstream_lookup.lua

@@ -0,0 +1,22 @@
+local records_key = KEYS[1]
+
+local now_ms = tonumber(ARGV[1])
+local proxy_id = ARGV[2]
+
+if not now_ms or now_ms <= 0 or type(proxy_id) ~= 'string' or proxy_id == '' then
+    return cjson.encode({status = 'invalid'})
+end
+
+local raw = redis.call('HGET', records_key, proxy_id)
+if not raw then
+    return cjson.encode({status = 'not_found'})
+end
+local decoded, record = pcall(cjson.decode, raw)
+if not decoded or type(record) ~= 'table' or type(record.sourceUpstream) ~= 'string' or
+    record.sourceUpstream == '' or type(record.expiresAtMs) ~= 'number' then
+    return cjson.encode({status = 'invalid'})
+end
+if record.expiresAtMs <= now_ms then
+    return cjson.encode({status = 'not_found'})
+end
+return cjson.encode({status = 'ok', upstream = record.sourceUpstream})

internal/adapters/redisactivity/upstream_lookup.go

@@ -0,0 +1,45 @@
+package redisactivity
+
+import (
+	"context"
+	"time"
+
+	"proxy-pool/internal/domain/activitypool"
+)
+
+var _ activitypool.ProxyUpstreamReader = (*Adapter)(nil)
+
+// UpstreamForProxy is a Controller-only metadata lookup. It returns neither
+// the proxy record nor its credentials and is never used by the Gateway path.
+func (a *Adapter) UpstreamForProxy(ctx context.Context, proxyID string, now time.Time) (string, error) {
+	if ctx == nil {
+		return "", activitypool.ErrInvalidProxyLookup
+	}
+	if err := ctx.Err(); err != nil {
+		return "", err
+	}
+	if a == nil || proxyID == "" || now.IsZero() {
+		return "", activitypool.ErrInvalidProxyLookup
+	}
+	result, err := runScript(ctx, a.client, upstreamLookupScript, []string{a.keys.records}, now.UnixMilli(), proxyID)
+	if err != nil {
+		return "", err
+	}
+	var reply upstreamLookupScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return "", err
+	}
+	switch reply.Status {
+	case scriptOK:
+		if reply.Upstream == "" {
+			return "", invalidScriptReply("upstream lookup reply omitted upstream")
+		}
+		return reply.Upstream, nil
+	case scriptNotFound:
+		return "", activitypool.ErrActivityNotFound
+	case scriptInvalid:
+		return "", activitypool.ErrInvalidProxyLookup
+	default:
+		return "", invalidScriptReply("unexpected upstream lookup status")
+	}
+}

internal/adapters/redisactivity/upstream_lookup_integration_test.go

@@ -0,0 +1,31 @@
+//go:build integration
+
+package redisactivity
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/domain/activitypool"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+func TestRedisLooksUpOnlyLiveProxyUpstream(t *testing.T) {
+	fixture := newRedisTestFixture(t)
+	now := redisTestNow()
+	if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
+		ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
+		Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
+	}); err != nil {
+		t.Fatalf("UpsertFetched(): %v", err)
+	}
+	upstream, err := fixture.Adapter.UpstreamForProxy(context.Background(), "proxy-a", now.Add(time.Second))
+	if err != nil || upstream != "provider-a" {
+		t.Fatalf("UpstreamForProxy(live) = (%q, %v)", upstream, err)
+	}
+	if _, err := fixture.Adapter.UpstreamForProxy(context.Background(), "proxy-a", now.Add(2*time.Minute)); !errors.Is(err, activitypool.ErrActivityNotFound) {
+		t.Fatalf("UpstreamForProxy(expired) error = %v, want ErrActivityNotFound", err)
+	}
+}

internal/controller/bootstrap/bootstrap.go

@@ -15,6 +15,7 @@ import (
 	"proxy-pool/internal/controller/admin"
 	"proxy-pool/internal/controller/distribution"
 	"proxy-pool/internal/controller/extraction"
+	controllerHealth "proxy-pool/internal/controller/health"
 	"proxy-pool/internal/controller/operations"
 	"proxy-pool/internal/controller/pool"
 	"proxy-pool/internal/controller/provider"
@@ -49,6 +50,9 @@ type Options struct {
 type activityStore interface {
 	extractionDomain.Store
 	activitypool.Upserter
+	activitypool.GlobalHealthStore
+	activitypool.TargetHealthStore
+	activitypool.ProxyUpstreamReader
 	pool.InventoryReader
 	activitypool.StateInventoryReader
 }
@@ -79,7 +83,7 @@ type runtimeFactory interface {
 }
 
 type workerRuntimeFactory interface {
-	New(config.ControlPlane, worker.Service) (controllerRunner, error)
+	New(config.ControlPlane, worker.Service, worker.ServerOptions) (controllerRunner, error)
 }
 
 func Run(ctx context.Context, options Options) error {
@@ -230,7 +234,7 @@ func runWithWorkerFactory(
 		runners = append(runners, runner)
 	}
 	if loaded.Value.ControlPlane.Enabled {
-		if nilInterface(opened.workerStore) {
+		if nilInterface(opened.workerStore) || nilInterface(opened.activity) {
 			return errors.Join(ErrStartup, ErrInvalidOptions)
 		}
 		var routingSources []worker.RoutingStateReader
@@ -262,7 +266,25 @@ func runWithWorkerFactory(
 		if serviceErr != nil {
 			return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr)
 		}
-		runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service)
+		threshold, thresholdErr := controllerHealth.NewConfiguredFailureThresholdResolver(configurationStore, opened.activity, options.Now)
+		if thresholdErr != nil {
+			return fmt.Errorf("%w: build Checker failure threshold resolver: %w", ErrStartup, thresholdErr)
+		}
+		reducer, reducerErr := controllerHealth.NewReducer(opened.activity, opened.activity, threshold)
+		if reducerErr != nil {
+			return fmt.Errorf("%w: build Checker health reducer: %w", ErrStartup, reducerErr)
+		}
+		checkerIdentity, identityErr := newCheckerIdentity(loaded.Value.ControlPlane)
+		if identityErr != nil {
+			return fmt.Errorf("%w: build Checker identity authorizer: %w", ErrStartup, identityErr)
+		}
+		checkerHandler, handlerErr := controllerHealth.NewGRPCHandler(reducer, checkerIdentity, controllerHealth.DefaultGRPCHandlerOptions())
+		if handlerErr != nil {
+			return fmt.Errorf("%w: build Checker control handler: %w", ErrStartup, handlerErr)
+		}
+		serverOptions := worker.DefaultServerOptions()
+		serverOptions.Checker = checkerHandler
+		runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service, serverOptions)
 		if runnerErr != nil {
 			return fmt.Errorf("%w: build Worker control server: %w", ErrStartup, runnerErr)
 		}
@@ -279,6 +301,17 @@ func runWithWorkerFactory(
 	return group.Run(ctx)
 }
 
+func newCheckerIdentity(controlPlane config.ControlPlane) (controllerHealth.CheckerIdentityAuthorizer, error) {
+	switch controlPlane.TLS.Mode {
+	case "disabled":
+		return worker.AllowLoopbackIdentity{}, nil
+	case "mtls":
+		return worker.NewSPIFFEIdentityAuthorizer(controlPlane.TLS.TrustDomain, controlPlane.TLS.Environment)
+	default:
+		return nil, ErrInvalidOptions
+	}
+}
+
 func prepareProviderConfiguration(
 	ctx context.Context,
 	configuration *config.Config,
@@ -361,6 +394,10 @@ func (productionRuntimeFactory) New(
 
 type productionWorkerRuntimeFactory struct{}
 
-func (productionWorkerRuntimeFactory) New(controlPlane config.ControlPlane, service worker.Service) (controllerRunner, error) {
-	return worker.NewServer(controlPlane, service, worker.DefaultServerOptions())
+func (productionWorkerRuntimeFactory) New(
+	controlPlane config.ControlPlane,
+	service worker.Service,
+	options worker.ServerOptions,
+) (controllerRunner, error) {
+	return worker.NewServer(controlPlane, service, options)
 }

internal/controller/bootstrap/bootstrap_integration_test.go

@@ -326,12 +326,13 @@ type integrationWorkerRuntimeFactory struct {
 func (factory *integrationWorkerRuntimeFactory) New(
 	controlPlane config.ControlPlane,
 	service controllerWorker.Service,
+	options controllerWorker.ServerOptions,
 ) (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())
+	server, err := controllerWorker.NewServer(controlPlane, service, options)
 	if err != nil {
 		_ = listener.Close()
 		return nil, err

internal/controller/bootstrap/bootstrap_test.go

@@ -16,6 +16,7 @@ import (
 	"proxy-pool/internal/domain/activitypool"
 	"proxy-pool/internal/domain/adminstate"
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	healthDomain "proxy-pool/internal/domain/health"
 	"proxy-pool/internal/domain/upstream"
 	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
@@ -213,8 +214,8 @@ func TestRunStartsWorkerControlPlaneWithoutHTTPRuntime(t *testing.T) {
 	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)
+	if workerFactory.controlPlane.Listen != "127.0.0.1:0" || workerFactory.service == nil || workerFactory.options.Checker == nil {
+		t.Fatalf("worker runtime factory = controlPlane:%+v service:%T checker:%T", workerFactory.controlPlane, workerFactory.service, workerFactory.options.Checker)
 	}
 }
 
@@ -383,6 +384,7 @@ func (factory runtimeFactoryFunc) New(
 type recordingWorkerRuntimeFactory struct {
 	controlPlane config.ControlPlane
 	service      controllerWorker.Service
+	options      controllerWorker.ServerOptions
 	runner       controllerRunner
 	err          error
 }
@@ -390,9 +392,11 @@ type recordingWorkerRuntimeFactory struct {
 func (factory *recordingWorkerRuntimeFactory) New(
 	controlPlane config.ControlPlane,
 	service controllerWorker.Service,
+	options controllerWorker.ServerOptions,
 ) (controllerRunner, error) {
 	factory.controlPlane = controlPlane
 	factory.service = service
+	factory.options = options
 	return factory.runner, factory.err
 }
 
@@ -426,6 +430,24 @@ func (*stubActivityStore) UpsertFetched(
 	return activitypool.UpsertResult{Accepted: len(batch.Proxies), Inserted: len(batch.Proxies)}, nil
 }
 
+func (*stubActivityStore) ApplyGlobalObservation(
+	context.Context,
+	activitypool.GlobalHealthCommand,
+) (activitypool.Entry, error) {
+	return activitypool.Entry{}, nil
+}
+
+func (*stubActivityStore) ApplyTargetObservation(
+	context.Context,
+	activitypool.TargetHealthCommand,
+) (healthDomain.TargetState, error) {
+	return healthDomain.TargetState{Status: healthDomain.TargetUnknown}, nil
+}
+
+func (*stubActivityStore) UpstreamForProxy(context.Context, string, time.Time) (string, error) {
+	return "provider-a", nil
+}
+
 func (*stubActivityStore) ReadInventory(
 	context.Context,
 	string,

internal/controller/health/grpc_handler.go

@@ -120,7 +120,7 @@ func rejectedObservationError(err error) bool {
 	return errors.Is(err, healthDomain.ErrInvalidObservation) || errors.Is(err, healthDomain.ErrInvalidFailureThreshold) ||
 		errors.Is(err, healthDomain.ErrStaleObservation) || errors.Is(err, healthDomain.ErrConflictingObservation) ||
 		errors.Is(err, activitypool.ErrInvalidHealthUpdate) || errors.Is(err, activitypool.ErrActivityNotFound) ||
-		errors.Is(err, ErrInvalidThreshold)
+		errors.Is(err, ErrInvalidThreshold) || errors.Is(err, ErrUnconfiguredUpstream)
 }
 
 func healthGRPCError(err error) error {

internal/controller/health/policy.go

@@ -0,0 +1,65 @@
+package health
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
+)
+
+var (
+	ErrInvalidPolicyResolver = errors.New("invalid health policy resolver")
+	ErrUnconfiguredUpstream  = errors.New("health observation references an unconfigured upstream")
+)
+
+// ConfigurationSource returns a detached, validated configuration snapshot.
+// config.Store implements this interface without exposing its publication
+// mechanism to the health package.
+type ConfigurationSource interface {
+	Current() *config.Config
+}
+
+// NewConfiguredFailureThresholdResolver resolves the policy against the
+// proxy's authoritative upstream at apply time. It keeps Checker input from
+// choosing its own failure threshold and follows Controller config reloads.
+func NewConfiguredFailureThresholdResolver(
+	configuration ConfigurationSource,
+	proxies activitypool.ProxyUpstreamReader,
+	now func() time.Time,
+) (FailureThresholdResolver, error) {
+	if nilInterface(configuration) || nilInterface(proxies) || now == nil {
+		return nil, ErrInvalidPolicyResolver
+	}
+	return func(ctx context.Context, observation healthDomain.Observation) (int, error) {
+		if ctx == nil {
+			return 0, ErrInvalidPolicyResolver
+		}
+		if err := ctx.Err(); err != nil {
+			return 0, err
+		}
+		current := configuration.Current()
+		if current == nil {
+			return 0, ErrInvalidPolicyResolver
+		}
+		currentTime := now()
+		if currentTime.IsZero() {
+			return 0, ErrInvalidPolicyResolver
+		}
+		upstreamID, err := proxies.UpstreamForProxy(ctx, observation.ProxyID, currentTime.UTC())
+		if err != nil {
+			return 0, err
+		}
+		upstream, exists := current.Upstreams[upstreamID]
+		if !exists || !upstream.Enabled {
+			return 0, ErrUnconfiguredUpstream
+		}
+		check := config.EffectiveCheck(current.Defaults.Check, upstream.Check)
+		if check.MaxConsecutiveFailures <= 0 {
+			return 0, ErrInvalidThreshold
+		}
+		return check.MaxConsecutiveFailures, nil
+	}, nil
+}

internal/controller/health/policy_test.go

@@ -0,0 +1,85 @@
+package health
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
+)
+
+func TestConfiguredFailureThresholdResolverUsesCurrentProxyUpstream(t *testing.T) {
+	now := time.Date(2026, 7, 31, 16, 0, 0, 0, time.UTC)
+	source := &configurationSourceStub{configuration: &config.Config{
+		Defaults: config.Defaults{Check: config.Check{MaxConsecutiveFailures: 2}},
+		Upstreams: map[string]config.Upstream{
+			"provider-a": {Enabled: true, Check: config.Check{MaxConsecutiveFailures: 4}},
+			"provider-b": {Enabled: true},
+		},
+	}}
+	proxies := &proxyUpstreamReaderStub{upstreams: map[string]string{"proxy-a": "provider-a", "proxy-b": "provider-b"}}
+	resolver, err := NewConfiguredFailureThresholdResolver(source, proxies, func() time.Time { return now })
+	if err != nil {
+		t.Fatalf("NewConfiguredFailureThresholdResolver(): %v", err)
+	}
+	if threshold, err := resolver(context.Background(), reducerObservation(healthDomain.LevelBasic, now)); err != nil || threshold != 4 {
+		t.Fatalf("resolver(provider-a) = (%d, %v)", threshold, err)
+	}
+	fact := reducerObservation(healthDomain.LevelTarget, now)
+	fact.ProxyID = "proxy-b"
+	if threshold, err := resolver(context.Background(), fact); err != nil || threshold != 2 {
+		t.Fatalf("resolver(provider-b default) = (%d, %v)", threshold, err)
+	}
+	if proxies.now != now {
+		t.Fatalf("lookup time = %v, want %v", proxies.now, now)
+	}
+}
+
+func TestConfiguredFailureThresholdResolverRejectsUnknownAndInvalidInputs(t *testing.T) {
+	now := time.Date(2026, 7, 31, 16, 0, 0, 0, time.UTC)
+	source := &configurationSourceStub{configuration: &config.Config{Upstreams: map[string]config.Upstream{}}}
+	proxies := &proxyUpstreamReaderStub{upstreams: map[string]string{"proxy-a": "missing"}}
+	resolver, err := NewConfiguredFailureThresholdResolver(source, proxies, func() time.Time { return now })
+	if err != nil {
+		t.Fatalf("NewConfiguredFailureThresholdResolver(): %v", err)
+	}
+	if _, err := resolver(context.Background(), reducerObservation(healthDomain.LevelBasic, now)); !errors.Is(err, ErrUnconfiguredUpstream) {
+		t.Fatalf("resolver(unknown) error = %v, want ErrUnconfiguredUpstream", err)
+	}
+	if resolver, err := NewConfiguredFailureThresholdResolver(nil, proxies, func() time.Time { return now }); err == nil || resolver != nil {
+		t.Fatalf("NewConfiguredFailureThresholdResolver(nil configuration) = (%v, %v)", resolver, err)
+	}
+	var typedNilReader *proxyUpstreamReaderStub
+	if resolver, err := NewConfiguredFailureThresholdResolver(source, typedNilReader, func() time.Time { return now }); err == nil || resolver != nil {
+		t.Fatalf("NewConfiguredFailureThresholdResolver(typed nil reader) = (%v, %v)", resolver, err)
+	}
+}
+
+type configurationSourceStub struct {
+	configuration *config.Config
+}
+
+func (source *configurationSourceStub) Current() *config.Config {
+	return source.configuration
+}
+
+type proxyUpstreamReaderStub struct {
+	upstreams map[string]string
+	now       time.Time
+	err       error
+}
+
+func (reader *proxyUpstreamReaderStub) UpstreamForProxy(_ context.Context, proxyID string, now time.Time) (string, error) {
+	reader.now = now
+	if reader.err != nil {
+		return "", reader.err
+	}
+	upstream, exists := reader.upstreams[proxyID]
+	if !exists {
+		return "", activitypool.ErrActivityNotFound
+	}
+	return upstream, nil
+}

internal/controller/worker/identity.go

@@ -10,7 +10,7 @@ import (
 	"google.golang.org/grpc/peer"
 )
 
-var ErrUnauthorizedIdentity = errors.New("worker identity is not authorized")
+var ErrUnauthorizedIdentity = errors.New("control-plane 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.
@@ -20,6 +20,10 @@ func (AllowLoopbackIdentity) Authorize(context.Context, string) error {
 	return nil
 }
 
+func (AllowLoopbackIdentity) AuthorizeChecker(context.Context, string) error {
+	return nil
+}
+
 type SPIFFEIdentityAuthorizer struct {
 	trustDomain string
 	environment string
@@ -33,7 +37,15 @@ func NewSPIFFEIdentityAuthorizer(trustDomain, environment string) (*SPIFFEIdenti
 }
 
 func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, workerID string) error {
-	if authorizer == nil || workerID == "" {
+	return authorizer.authorize(ctx, "worker", workerID)
+}
+
+func (authorizer *SPIFFEIdentityAuthorizer) AuthorizeChecker(ctx context.Context, checkerID string) error {
+	return authorizer.authorize(ctx, "checker", checkerID)
+}
+
+func (authorizer *SPIFFEIdentityAuthorizer) authorize(ctx context.Context, resourceType, resourceID string) error {
+	if authorizer == nil || (resourceType != "worker" && resourceType != "checker") || resourceID == "" {
 		return ErrUnauthorizedIdentity
 	}
 	peerInfo, ok := peer.FromContext(ctx)
@@ -45,8 +57,8 @@ func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, worke
 		return ErrUnauthorizedIdentity
 	}
 
-	expectedPath := "/" + authorizer.environment + "/worker/" + workerID
-	workerIdentityCount := 0
+	expectedPath := "/" + authorizer.environment + "/" + resourceType + "/" + resourceID
+	identityCount := 0
 	for _, chain := range tlsInfo.State.VerifiedChains {
 		if len(chain) == 0 || chain[0] == nil {
 			continue
@@ -56,16 +68,16 @@ func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, worke
 				continue
 			}
 			segments := strings.Split(strings.Trim(uri.Path, "/"), "/")
-			if len(segments) != 3 || segments[1] != "worker" {
+			if len(segments) != 3 || segments[1] != resourceType {
 				continue
 			}
-			workerIdentityCount++
-			if workerIdentityCount > 1 || uri.RawQuery != "" || uri.Fragment != "" || uri.Path != expectedPath {
+			identityCount++
+			if identityCount > 1 || uri.RawQuery != "" || uri.Fragment != "" || uri.Path != expectedPath {
 				return ErrUnauthorizedIdentity
 			}
 		}
 	}
-	if workerIdentityCount != 1 {
+	if identityCount != 1 {
 		return ErrUnauthorizedIdentity
 	}
 	return nil

internal/controller/worker/identity_test.go

@@ -66,10 +66,29 @@ func TestSPIFFEIdentityAuthorizerRejectsMultipleWorkerURIs(t *testing.T) {
 	}
 }
 
+func TestSPIFFEIdentityAuthorizerAuthorizesCheckerOnlyForMatchingResource(t *testing.T) {
+	authorizer, err := NewSPIFFEIdentityAuthorizer("proxy.example", "prod")
+	if err != nil {
+		t.Fatalf("NewSPIFFEIdentityAuthorizer(): %v", err)
+	}
+	if err := authorizer.AuthorizeChecker(tlsPeerContext(t, "spiffe://proxy.example/prod/checker/checker-a"), "checker-a"); err != nil {
+		t.Fatalf("AuthorizeChecker(matching checker) = %v", err)
+	}
+	if err := authorizer.AuthorizeChecker(tlsPeerContext(t, "spiffe://proxy.example/prod/worker/checker-a"), "checker-a"); err == nil {
+		t.Fatal("AuthorizeChecker(worker identity) error = nil")
+	}
+	if err := authorizer.Authorize(tlsPeerContext(t, "spiffe://proxy.example/prod/checker/checker-a"), "checker-a"); err == nil {
+		t.Fatal("Authorize(checker identity) error = nil")
+	}
+}
+
 func TestAllowLoopbackIdentity(t *testing.T) {
 	if err := (AllowLoopbackIdentity{}).Authorize(context.Background(), "worker-a"); err != nil {
 		t.Fatalf("Authorize(): %v", err)
 	}
+	if err := (AllowLoopbackIdentity{}).AuthorizeChecker(context.Background(), "checker-a"); err != nil {
+		t.Fatalf("AuthorizeChecker(): %v", err)
+	}
 }
 
 func tlsPeerContext(t *testing.T, identityURI string) context.Context {

internal/controller/worker/server.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"reflect"
 	"strings"
 	"time"
 
@@ -24,6 +25,7 @@ var ErrInvalidServer = errors.New("invalid worker control server configuration")
 type ServerOptions struct {
 	ShutdownTimeout time.Duration
 	Snapshots       SnapshotSource
+	Checker         controlplanev1.CheckerControlPlaneServer
 }
 
 func DefaultServerOptions() ServerOptions {
@@ -74,9 +76,25 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server
 	)
 	grpcServer := grpc.NewServer(serverOptions...)
 	controlplanev1.RegisterWorkerControlPlaneServer(grpcServer, NewGRPCHandler(service, identity, snapshots))
+	if !nilService(options.Checker) {
+		controlplanev1.RegisterCheckerControlPlaneServer(grpcServer, options.Checker)
+	}
 	return &Server{listen: controlPlane.Listen, grpcServer: grpcServer, shutdownTimeout: options.ShutdownTimeout}, nil
 }
 
+func nilService(value any) bool {
+	if value == nil {
+		return true
+	}
+	reflected := reflect.ValueOf(value)
+	switch reflected.Kind() {
+	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
+		return reflected.IsNil()
+	default:
+		return false
+	}
+}
+
 func (server *Server) Run(ctx context.Context) error {
 	if server == nil || server.grpcServer == nil || server.listen == "" {
 		return ErrInvalidServer

internal/controller/worker/server_test.go

@@ -70,6 +70,56 @@ func TestServerServesAndStopsOnContextCancellation(t *testing.T) {
 	}
 }
 
+func TestServerRegistersCheckerServiceOnTheExistingControlPlaneListener(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()
+	checker := &checkerServiceStub{}
+	server, err := NewServer(controlPlane, &grpcServiceStub{}, ServerOptions{ShutdownTimeout: time.Second, Checker: checker})
+	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)
+	}
+	response, err := controlplanev1.NewCheckerControlPlaneClient(connection).ReportObservations(context.Background(), &controlplanev1.ObservationBatch{
+		CheckerId: "checker-a",
+	})
+	if err != nil || response.GetAccepted() != 1 || checker.checkerID != "checker-a" {
+		t.Fatalf("ReportObservations() = (%+v, %v); checker=%q", response, err, checker.checkerID)
+	}
+	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")
+	}
+}
+
+type checkerServiceStub struct {
+	controlplanev1.UnimplementedCheckerControlPlaneServer
+	checkerID string
+}
+
+func (stub *checkerServiceStub) ReportObservations(_ context.Context, request *controlplanev1.ObservationBatch) (*controlplanev1.ReportObservationsResponse, error) {
+	stub.checkerID = request.GetCheckerId()
+	return &controlplanev1.ReportObservationsResponse{Accepted: 1}, nil
+}
+
 func validServerControlPlane() config.ControlPlane {
 	return config.ControlPlane{
 		Enabled:              true,

internal/domain/activitypool/pool.go

@@ -20,6 +20,7 @@ const defaultIdempotencyTTL = 5 * time.Minute
 var (
 	ErrInvalidBatch        = errors.New("invalid activity pool batch")
 	ErrInvalidHealthUpdate = errors.New("invalid activity pool health update")
+	ErrInvalidProxyLookup  = errors.New("invalid activity pool proxy lookup")
 	ErrActivityNotFound    = errors.New("activity pool proxy not found")
 	ErrStaleHealthUpdate   = errors.New("stale activity pool health update")
 	ErrInvalidInventory    = errors.New("invalid activity pool inventory query")
@@ -101,6 +102,13 @@ type TargetHealthStore interface {
 	ApplyTargetObservation(context.Context, TargetHealthCommand) (healthDomain.TargetState, error)
 }
 
+// ProxyUpstreamReader returns only the authoritative upstream that owns a
+// still-live proxy. It intentionally does not expose proxy endpoints or
+// credentials to Controller policy resolution.
+type ProxyUpstreamReader interface {
+	UpstreamForProxy(context.Context, string, time.Time) (string, error)
+}
+
 type InventoryReader interface {
 	Inventory(context.Context, string, time.Time) (Inventory, error)
 }
@@ -148,6 +156,7 @@ var (
 	_ HealthStore                = (*MemoryPool)(nil)
 	_ GlobalHealthStore          = (*MemoryPool)(nil)
 	_ TargetHealthStore          = (*MemoryPool)(nil)
+	_ ProxyUpstreamReader        = (*MemoryPool)(nil)
 	_ InventoryReader            = (*MemoryPool)(nil)
 	_ StateInventoryReader       = (*MemoryPool)(nil)
 	_ Maintainer                 = (*MemoryPool)(nil)
@@ -435,6 +444,26 @@ func (p *MemoryPool) ApplyTargetObservation(
 	return next, nil
 }
 
+func (p *MemoryPool) UpstreamForProxy(ctx context.Context, proxyID string, now time.Time) (string, error) {
+	if ctx == nil {
+		return "", ErrInvalidProxyLookup
+	}
+	if err := ctx.Err(); err != nil {
+		return "", err
+	}
+	if p == nil || proxyID == "" || now.IsZero() {
+		return "", ErrInvalidProxyLookup
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.purgeExpiredLocked(now)
+	entry, exists := p.entryByIDLocked(proxyID)
+	if !exists || entry.Proxy.SourceUpstream == "" {
+		return "", ErrActivityNotFound
+	}
+	return entry.Proxy.SourceUpstream, nil
+}
+
 func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) {
 	result := Inventory{UpstreamID: upstreamID}
 	if ctx == nil {

internal/domain/activitypool/upstream_lookup_test.go

@@ -0,0 +1,23 @@
+package activitypool
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestMemoryPoolLooksUpOnlyLiveProxyUpstream(t *testing.T) {
+	now := time.Date(2026, 7, 31, 15, 0, 0, 0, time.UTC)
+	pool := seededHealthPool(t, now)
+	upstream, err := pool.UpstreamForProxy(context.Background(), "proxy-a", now.Add(time.Second))
+	if err != nil || upstream != "provider-a" {
+		t.Fatalf("UpstreamForProxy(live) = (%q, %v)", upstream, err)
+	}
+	if _, err := pool.UpstreamForProxy(context.Background(), "proxy-a", now.Add(2*time.Minute)); !errors.Is(err, ErrActivityNotFound) {
+		t.Fatalf("UpstreamForProxy(expired) error = %v, want ErrActivityNotFound", err)
+	}
+	if _, err := pool.UpstreamForProxy(context.Background(), "", now); !errors.Is(err, ErrInvalidProxyLookup) {
+		t.Fatalf("UpstreamForProxy(empty) error = %v, want ErrInvalidProxyLookup", err)
+	}
+}