feat: add controller health reducer

docs/development/implementation-plan.md

@@ -278,7 +278,8 @@ Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄
 Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、AVAILABLE/SUSPECT 的
 阈值降级、精确重放、冲突拒绝与成功恢复。TARGET Profile 也已在 Memory 和 Redis 中独立
 归并，以哈希键保存并随代理 TTL 过期，绝不写入 Proxy 全局状态或选择索引。调度器、
-Checker RPC 与独立进程尚未接入，因此本任务保持未完成。
+Controller 到 Checker 的 RPC、任务调度和独立进程尚未接入；Controller 已新增公用 Reducer
+作为 Observation 到原子 Store 的唯一归并边界，因此本任务保持未完成。
 
 ## Task 12: Machine-readable Contracts
 

docs/requirements/completion-audit.md

@@ -59,7 +59,8 @@ Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot
 Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/
 摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer
 与 Memory/Redis 原子状态提交基础；TARGET Profile 以独立、随代理 TTL 过期的 Redis
-记录归并，不改写 Proxy 全局状态。任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
+记录归并，不改写 Proxy 全局状态。Controller 公用 Reducer 已作为 Observation 的唯一状态
+归并边界；任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
   `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller
   在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用
   `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot

docs/requirements/traceability.md

@@ -81,7 +81,7 @@
 |---|---|---|---|
 | 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-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | 全局 Reducer 已在 Memory/Redis 活动池原子提交连续失败、精确重放和成功恢复；任务调度与 REMOVE 编排待实现 |
+| 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 | 配置校验测试 |
 | OPS-001 | 配置校验后构建不可变快照并原子替换 | 8959-8999 | 100k 索引、版本/epoch 与并发 Apply/Acquire 测试 |

internal/config/effective_check.go

@@ -0,0 +1,41 @@
+package config
+
+// EffectiveCheck overlays one upstream's non-zero check fields on the global
+// defaults. A non-nil URL slice is an explicit override, including an empty
+// slice that intentionally disables URL-based EGRESS checks for that upstream.
+func EffectiveCheck(defaults, override Check) Check {
+	effective := defaults
+	if override.Interval != 0 {
+		effective.Interval = override.Interval
+	}
+	if override.Jitter != 0 {
+		effective.Jitter = override.Jitter
+	}
+	if override.MaxInFlight != 0 {
+		effective.MaxInFlight = override.MaxInFlight
+	}
+	if override.Timeout != 0 {
+		effective.Timeout = override.Timeout
+	}
+	if override.MaxAttempts != 0 {
+		effective.MaxAttempts = override.MaxAttempts
+	}
+	if override.MaxConsecutiveFailures != 0 {
+		effective.MaxConsecutiveFailures = override.MaxConsecutiveFailures
+	}
+	if override.URLs != nil {
+		effective.URLs = cloneCheckURLs(override.URLs)
+	} else {
+		effective.URLs = cloneCheckURLs(defaults.URLs)
+	}
+	return effective
+}
+
+func cloneCheckURLs(source []string) []string {
+	if source == nil {
+		return nil
+	}
+	result := make([]string, len(source))
+	copy(result, source)
+	return result
+}

internal/config/effective_check_test.go

@@ -0,0 +1,61 @@
+package config
+
+import (
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestEffectiveCheckOverlaysScalarsAndClonesURLs(t *testing.T) {
+	defaults := Check{
+		Interval: Duration(time.Minute), Jitter: 20, MaxInFlight: 12, Timeout: Duration(5 * time.Second),
+		MaxAttempts: 3, MaxConsecutiveFailures: 2, URLs: []string{"https://egress.example/check"},
+	}
+	override := Check{Timeout: Duration(8 * time.Second), MaxAttempts: 5}
+	effective := EffectiveCheck(defaults, override)
+	if effective.Interval != defaults.Interval || effective.Jitter != defaults.Jitter ||
+		effective.MaxInFlight != defaults.MaxInFlight || effective.Timeout != override.Timeout ||
+		effective.MaxAttempts != override.MaxAttempts ||
+		effective.MaxConsecutiveFailures != defaults.MaxConsecutiveFailures ||
+		len(effective.URLs) != 1 || effective.URLs[0] != defaults.URLs[0] {
+		t.Fatalf("EffectiveCheck() = %+v", effective)
+	}
+	effective.URLs[0] = "https://changed.example/check"
+	if defaults.URLs[0] == effective.URLs[0] {
+		t.Fatal("EffectiveCheck() aliases default URLs")
+	}
+
+	effective = EffectiveCheck(defaults, Check{URLs: []string{}})
+	if effective.URLs == nil || len(effective.URLs) != 0 {
+		t.Fatalf("EffectiveCheck(explicit empty URLs) = %+v", effective)
+	}
+}
+
+func TestValidateUsesEffectiveCheckAndRejectsUnsafeURLs(t *testing.T) {
+	cfg := mustLoadValidConfig(t)
+	defaults := cfg.Upstreams["provider-a"].Check
+	cfg.Defaults.Check = defaults
+	upstream := cfg.Upstreams["provider-a"]
+	upstream.Check = Check{Timeout: Duration(7 * time.Second)}
+	cfg.Upstreams["provider-a"] = upstream
+	if err := Validate(cfg); err != nil {
+		t.Fatalf("Validate(effective check) = %v", err)
+	}
+
+	for _, invalid := range []string{
+		"target.example/check", "ftp://target.example/check", "https://user:pass@target.example/check",
+		"https://target.example/check#fragment", " https://target.example/check", "https://target.example/check",
+	} {
+		cfg := mustLoadValidConfig(t)
+		upstream := cfg.Upstreams["provider-a"]
+		if strings.HasPrefix(invalid, "https://target.example/check") && invalid == "https://target.example/check" {
+			upstream.Check.URLs = []string{invalid, invalid}
+		} else {
+			upstream.Check.URLs = []string{invalid}
+		}
+		cfg.Upstreams["provider-a"] = upstream
+		if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "check.urls") {
+			t.Fatalf("Validate(check URL %q) error = %v, want check.urls", invalid, err)
+		}
+	}
+}

internal/config/validate.go

@@ -62,7 +62,11 @@ func Validate(cfg *Config) error {
 		if upstream.Enabled {
 			enabledUpstreams++
 		}
-		if err := validateUpstream(name, upstream); err != nil {
+		effective := upstream
+		if upstream.Enabled {
+			effective.Check = EffectiveCheck(cfg.Defaults.Check, upstream.Check)
+		}
+		if err := validateUpstream(name, effective); err != nil {
 			return err
 		}
 	}
@@ -577,6 +581,22 @@ func validateCheck(scope string, check Check) error {
 	if err := requirePositive(scope+".maxConsecutiveFailures", check.MaxConsecutiveFailures); err != nil {
 		return err
 	}
+	seenURLs := make(map[string]struct{}, len(check.URLs))
+	for index, rawURL := range check.URLs {
+		if rawURL == "" || strings.TrimSpace(rawURL) != rawURL {
+			return fmt.Errorf("validate %s.urls[%d]: must be an absolute http or https URL", scope, index)
+		}
+		parsed, err := url.Parse(rawURL)
+		if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" ||
+			(parsed.Scheme != "http" && parsed.Scheme != "https") {
+			return fmt.Errorf("validate %s.urls[%d]: must be an absolute http or https URL", scope, index)
+		}
+		canonical := parsed.String()
+		if _, exists := seenURLs[canonical]; exists {
+			return fmt.Errorf("validate %s.urls[%d]: duplicate URL", scope, index)
+		}
+		seenURLs[canonical] = struct{}{}
+	}
 	return nil
 }
 

internal/controller/health/reducer.go

@@ -0,0 +1,103 @@
+// Package health owns Controller-side reduction of Checker facts. It never
+// probes, schedules, or persists raw observations.
+package health
+
+import (
+	"context"
+	"errors"
+	"reflect"
+
+	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
+)
+
+var (
+	ErrInvalidReducer   = errors.New("invalid health reducer")
+	ErrInvalidThreshold = errors.New("invalid health failure threshold")
+)
+
+// FailureThresholdResolver selects the configured consecutive-failure limit
+// for one normalized fact. It keeps policy lookup outside the Checker and
+// lets a Controller use different limits for different upstreams.
+type FailureThresholdResolver func(context.Context, healthDomain.Observation) (int, error)
+
+// Result contains only the reduced state written by the corresponding store.
+// Exactly one of Global or Target is non-nil for an accepted observation.
+type Result struct {
+	Global *activitypool.Entry
+	Target *healthDomain.TargetState
+}
+
+// Reducer is the Controller's narrow authority boundary for checker facts.
+// The stores commit their own atomic idempotency and ordering semantics.
+type Reducer struct {
+	global    activitypool.GlobalHealthStore
+	target    activitypool.TargetHealthStore
+	threshold FailureThresholdResolver
+}
+
+func NewReducer(
+	global activitypool.GlobalHealthStore,
+	target activitypool.TargetHealthStore,
+	threshold FailureThresholdResolver,
+) (*Reducer, error) {
+	if nilInterface(global) || nilInterface(target) || threshold == nil {
+		return nil, ErrInvalidReducer
+	}
+	return &Reducer{global: global, target: target, threshold: threshold}, nil
+}
+
+func nilInterface(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
+	}
+}
+
+// Apply normalizes one immutable Checker fact, resolves its configured policy,
+// and commits it through exactly one atomic activity-pool store method.
+func (r *Reducer) Apply(ctx context.Context, observation healthDomain.Observation) (Result, error) {
+	if ctx == nil || r == nil || r.global == nil || r.target == nil || r.threshold == nil {
+		return Result{}, ErrInvalidReducer
+	}
+	if err := ctx.Err(); err != nil {
+		return Result{}, err
+	}
+	normalized, err := healthDomain.NormalizeObservation(observation)
+	if err != nil {
+		return Result{}, err
+	}
+	maxConsecutiveFailures, err := r.threshold(ctx, normalized)
+	if err != nil {
+		return Result{}, err
+	}
+	if maxConsecutiveFailures <= 0 {
+		return Result{}, ErrInvalidThreshold
+	}
+	switch normalized.Level {
+	case healthDomain.LevelBasic, healthDomain.LevelEgress:
+		entry, err := r.global.ApplyGlobalObservation(ctx, activitypool.GlobalHealthCommand{
+			Observation: normalized, MaxConsecutiveFailures: maxConsecutiveFailures,
+		})
+		if err != nil {
+			return Result{}, err
+		}
+		return Result{Global: &entry}, nil
+	case healthDomain.LevelTarget:
+		state, err := r.target.ApplyTargetObservation(ctx, activitypool.TargetHealthCommand{
+			Observation: normalized, MaxConsecutiveFailures: maxConsecutiveFailures,
+		})
+		if err != nil {
+			return Result{}, err
+		}
+		return Result{Target: &state}, nil
+	default:
+		return Result{}, healthDomain.ErrInvalidObservation
+	}
+}

internal/controller/health/reducer_test.go

@@ -0,0 +1,96 @@
+package health
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
+)
+
+func TestReducerRoutesObservationsToTheirIsolatedStores(t *testing.T) {
+	global := &recordingGlobalStore{}
+	target := &recordingTargetStore{}
+	var policyFacts []healthDomain.Observation
+	reducer, err := NewReducer(global, target, func(_ context.Context, observation healthDomain.Observation) (int, error) {
+		policyFacts = append(policyFacts, observation)
+		return 3, nil
+	})
+	if err != nil {
+		t.Fatalf("NewReducer(): %v", err)
+	}
+	now := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC)
+	globalResult, err := reducer.Apply(context.Background(), reducerObservation(healthDomain.LevelBasic, now))
+	if err != nil || globalResult.Global == nil || globalResult.Target != nil || len(global.commands) != 1 || len(target.commands) != 0 {
+		t.Fatalf("Apply(BASIC) = %+v, %v; global=%d target=%d", globalResult, err, len(global.commands), len(target.commands))
+	}
+	if global.commands[0].MaxConsecutiveFailures != 3 || global.commands[0].Observation.Level != healthDomain.LevelBasic {
+		t.Fatalf("global command = %+v", global.commands[0])
+	}
+
+	targetResult, err := reducer.Apply(context.Background(), reducerObservation(healthDomain.LevelTarget, now.Add(time.Second)))
+	if err != nil || targetResult.Global != nil || targetResult.Target == nil || len(global.commands) != 1 || len(target.commands) != 1 {
+		t.Fatalf("Apply(TARGET) = %+v, %v; global=%d target=%d", targetResult, err, len(global.commands), len(target.commands))
+	}
+	if target.commands[0].MaxConsecutiveFailures != 3 || target.commands[0].Observation.Level != healthDomain.LevelTarget {
+		t.Fatalf("target command = %+v", target.commands[0])
+	}
+	if len(policyFacts) != 2 || policyFacts[0].ObservedAt.Location() != time.UTC || policyFacts[1].RoutingName != "route-a" {
+		t.Fatalf("policy observations = %+v", policyFacts)
+	}
+}
+
+func TestReducerRejectsInvalidPolicyAndDoesNotCallStores(t *testing.T) {
+	global := &recordingGlobalStore{}
+	target := &recordingTargetStore{}
+	reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 0, nil })
+	if err != nil {
+		t.Fatalf("NewReducer(): %v", err)
+	}
+	if _, err := reducer.Apply(context.Background(), reducerObservation(healthDomain.LevelEgress, time.Now())); !errors.Is(err, ErrInvalidThreshold) {
+		t.Fatalf("Apply(zero threshold) error = %v, want ErrInvalidThreshold", err)
+	}
+	if len(global.commands) != 0 || len(target.commands) != 0 {
+		t.Fatalf("stores called with invalid policy: global=%d target=%d", len(global.commands), len(target.commands))
+	}
+
+	if reducer, err := NewReducer(nil, target, func(context.Context, healthDomain.Observation) (int, error) { return 1, nil }); err == nil || reducer != nil {
+		t.Fatalf("NewReducer(nil global) = (%v, %v)", reducer, err)
+	}
+	var typedNilGlobal *recordingGlobalStore
+	if reducer, err := NewReducer(typedNilGlobal, target, func(context.Context, healthDomain.Observation) (int, error) { return 1, nil }); err == nil || reducer != nil {
+		t.Fatalf("NewReducer(typed nil global) = (%v, %v)", reducer, err)
+	}
+}
+
+func reducerObservation(level healthDomain.Level, observedAt time.Time) healthDomain.Observation {
+	observation := healthDomain.Observation{
+		TaskID: "task-a", ProxyID: "proxy-a", Level: level, Success: true,
+		Latency: time.Millisecond, ObservedAt: observedAt,
+	}
+	if level == healthDomain.LevelTarget {
+		observation.RoutingName = "route-a"
+		observation.TargetURL = "https://target.example/check"
+	}
+	return observation
+}
+
+type recordingGlobalStore struct {
+	commands []activitypool.GlobalHealthCommand
+}
+
+func (store *recordingGlobalStore) ApplyGlobalObservation(_ context.Context, command activitypool.GlobalHealthCommand) (activitypool.Entry, error) {
+	store.commands = append(store.commands, command)
+	return activitypool.Entry{}, nil
+}
+
+type recordingTargetStore struct {
+	commands []activitypool.TargetHealthCommand
+}
+
+func (store *recordingTargetStore) ApplyTargetObservation(_ context.Context, command activitypool.TargetHealthCommand) (healthDomain.TargetState, error) {
+	store.commands = append(store.commands, command)
+	return healthDomain.TargetState{Status: healthDomain.TargetAvailable}, nil
+}