feat: add atomic global health reducer

docs/development/implementation-plan.md

@@ -273,6 +273,12 @@ Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重
 - [ ] Ensure target failures affect only the target profile.
 - [ ] Add fixture target server and deterministic clock/scheduler tests.
 
+当前进度（2026-07-31）：已新增不依赖 gRPC/Redis 的 `domain/health` Observation、
+全局/Target Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄接口在
+Memory 和 Redis Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、
+AVAILABLE/SUSPECT 的阈值降级、精确重放、冲突拒绝与成功恢复。TARGET 仍只在纯领域
+Profile 中归并，尚未接入持久化、调度器、Checker RPC 或独立进程，因此本任务保持未完成。
+
 ## Task 12: Machine-readable Contracts
 
 **Files:** `api/openapi/proxy-pool.yaml`, `api/proto/controlplane/v1/controlplane.proto`,

docs/requirements/completion-audit.md

@@ -57,7 +57,9 @@
 Routing 一起纳入 Snapshot checksum，仅在 mTLS 和 Gateway 当前内存 View 中处理。Gateway
 Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、
 Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/
-摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
+摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer
+与 Memory/Redis 原子状态提交基础，但任务调度、Checker RPC、TARGET Profile 持久化和
+独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
   `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller
   在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用
   `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot

docs/requirements/traceability.md

@@ -79,9 +79,9 @@
 
 | ID | 最终需求 | 来源 | 验证证据 |
 |---|---|---|---|
-| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | Protobuf/配置契约已定义；目标 Profile 与健康 Reducer 待实现 |
+| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 持久化和 Routing 消费待实现 |
 | HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置校验已完成；有界调度器、抖动和分级频率测试待实现 |
-| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Proxy 状态迁移骨架已完成；连续失败 Reducer 与活动池原子提交待实现 |
+| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | 全局 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/adapters/redisactivity/codec.go

@@ -32,6 +32,10 @@ type proxyRecord struct {
 	UsableUntilMS          int64             `json:"usableUntilMs"`
 	LastCheckedAtMS        int64             `json:"lastCheckedAtMs,omitempty"`
 	LastSuccessAtMS        int64             `json:"lastSuccessAtMs,omitempty"`
+	ConsecutiveFailures    int               `json:"consecutiveFailures,omitempty"`
+	LastHealthTaskID       string            `json:"lastHealthTaskId,omitempty"`
+	LastHealthDigest       string            `json:"lastHealthDigest,omitempty"`
+	LastHealthObservedAtMS int64             `json:"lastHealthObservedAtMs,omitempty"`
 	LatencyNS              int64             `json:"latencyNs"`
 	MaxConcurrency         int64             `json:"maxConcurrency"`
 	State                  string            `json:"state"`
@@ -204,11 +208,16 @@ func validateProxyRecord(record proxyRecord) error {
 		record.Port <= 0 || record.Port > 65_535 || record.SourceUpstream == "" ||
 		record.CreatedAtMS <= 0 || record.ExpiresAtMS <= 0 || record.UsableUntilMS <= 0 ||
 		record.UsableUntilMS > record.ExpiresAtMS || record.LastCheckedAtMS < 0 ||
-		record.LastSuccessAtMS < 0 || record.LatencyNS < 0 || record.MaxConcurrency < 0 ||
+		record.LastSuccessAtMS < 0 || record.ConsecutiveFailures < 0 || record.LatencyNS < 0 || record.MaxConcurrency < 0 ||
 		!validScheme(record.Scheme) || !validProxyState(record.State) ||
 		record.OwnerIndexKey == "" || !strings.Contains(record.OwnerIndexKey, "{activity}") {
 		return ErrInvalidRecord
 	}
+	if (record.LastHealthTaskID == "") != (record.LastHealthDigest == "") ||
+		(record.LastHealthDigest != "" && (!validDigest(record.LastHealthDigest) || record.LastHealthObservedAtMS <= 0)) ||
+		record.LastHealthObservedAtMS < 0 {
+		return ErrInvalidRecord
+	}
 	for _, key := range record.IndexKeys {
 		if key == "" || !strings.Contains(key, "{activity}") {
 			return ErrInvalidRecord

internal/adapters/redisactivity/global_health_integration_test.go

@@ -0,0 +1,74 @@
+//go:build integration
+
+package redisactivity
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+func TestRedisAppliesGlobalHealthObservationAtomically(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)
+	}
+	if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(checking): %v", err)
+	}
+	first := redisGlobalObservation("task-1", false, now.Add(2*time.Second))
+	failed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
+		Observation: first, MaxConsecutiveFailures: 3,
+	})
+	if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 {
+		t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err)
+	}
+	replayed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
+		Observation: first, MaxConsecutiveFailures: 3,
+	})
+	if err != nil || replayed.GlobalHealth != failed.GlobalHealth {
+		t.Fatalf("ApplyGlobalObservation(replay) = %+v, %v", replayed, err)
+	}
+	conflicting := first
+	conflicting.Success = true
+	conflicting.FailureClass = ""
+	if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
+		Observation: conflicting, MaxConsecutiveFailures: 3,
+	}); !errors.Is(err, healthDomain.ErrConflictingObservation) {
+		t.Fatalf("ApplyGlobalObservation(conflict) error = %v, want ErrConflictingObservation", err)
+	}
+	if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(3 * time.Second), NextState: proxyDomain.StateChecking,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(recheck): %v", err)
+	}
+	recovered, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
+		Observation: redisGlobalObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3,
+	})
+	if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.GlobalHealth.ConsecutiveFailures != 0 ||
+		recovered.GlobalHealth.LastObservedAt != now.Add(4*time.Second) {
+		t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", recovered, err)
+	}
+}
+
+func redisGlobalObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation {
+	observation := healthDomain.Observation{
+		TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic,
+		Success: success, Latency: 10 * time.Millisecond, ObservedAt: observedAt,
+	}
+	if !success {
+		observation.FailureClass = "timeout"
+	}
+	return observation
+}

internal/adapters/redisactivity/health.go

@@ -2,11 +2,14 @@ package redisactivity
 
 import (
 	"context"
+	"encoding/hex"
 
 	"proxy-pool/internal/domain/activitypool"
+	healthDomain "proxy-pool/internal/domain/health"
 )
 
 var _ activitypool.HealthStore = (*Adapter)(nil)
+var _ activitypool.GlobalHealthStore = (*Adapter)(nil)
 
 func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpdate) (activitypool.Entry, error) {
 	if ctx == nil {
@@ -56,3 +59,71 @@ func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpd
 		return activitypool.Entry{}, invalidScriptReply("unexpected health status")
 	}
 }
+
+// ApplyGlobalObservation reduces one BASIC or EGRESS observation inside the
+// same Redis Lua transaction that updates state indexes and inventory.
+func (a *Adapter) ApplyGlobalObservation(
+	ctx context.Context,
+	command activitypool.GlobalHealthCommand,
+) (activitypool.Entry, error) {
+	if ctx == nil {
+		return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate
+	}
+	if err := ctx.Err(); err != nil {
+		return activitypool.Entry{}, err
+	}
+	normalized, err := healthDomain.NormalizeObservation(command.Observation)
+	if err != nil {
+		return activitypool.Entry{}, err
+	}
+	if a == nil || normalized.Level == healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 {
+		return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate
+	}
+	digest, err := healthDomain.ObservationDigest(normalized)
+	if err != nil {
+		return activitypool.Entry{}, err
+	}
+	operationID, err := newOperationID()
+	if err != nil {
+		return activitypool.Entry{}, err
+	}
+	success := "0"
+	if normalized.Success {
+		success = "1"
+	}
+	result, err := runScript(ctx, a.client, healthScript, []string{
+		a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
+		a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
+		a.keys.operation(operationID),
+	}, normalized.ObservedAt.UnixMilli(), "", int64(normalized.Latency),
+		a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), normalized.ProxyID,
+		"global", success, command.MaxConsecutiveFailures, normalized.TaskID, hex.EncodeToString(digest[:]))
+	if err != nil {
+		return activitypool.Entry{}, err
+	}
+	var reply healthScriptReply
+	if err := decodeScriptResult(result, &reply); err != nil {
+		return activitypool.Entry{}, err
+	}
+	switch reply.Status {
+	case scriptNotFound:
+		return activitypool.Entry{}, activitypool.ErrActivityNotFound
+	case scriptStale:
+		return activitypool.Entry{}, healthDomain.ErrStaleObservation
+	case scriptConflict:
+		return activitypool.Entry{}, healthDomain.ErrConflictingObservation
+	case scriptInvalid:
+		return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate
+	case scriptOK:
+		if reply.Record == "" {
+			return activitypool.Entry{}, invalidScriptReply("global health reply omitted record")
+		}
+		record, err := decodeProxyRecord(reply.Record)
+		if err != nil {
+			return activitypool.Entry{}, invalidScriptReply("global health reply contained an invalid record")
+		}
+		return proxyRecordEntry(record), nil
+	default:
+		return activitypool.Entry{}, invalidScriptReply("unexpected global health status")
+	}
+}

internal/adapters/redisactivity/scripts/health.lua

@@ -15,6 +15,11 @@ local latency_ns = tonumber(ARGV[3])
 local cleanup_limit = tonumber(ARGV[4])
 local operation_ttl_ms = tonumber(ARGV[5])
 local proxy_id = ARGV[6]
+local mode = ARGV[7] or 'transition'
+local success = ARGV[8]
+local max_consecutive_failures = tonumber(ARGV[9])
+local task_id = ARGV[10]
+local observation_digest = ARGV[11]
 
 local committed = redis.call('GET', operation_key)
 if committed then
@@ -194,14 +199,61 @@ if tonumber(record.expiresAtMs) <= checked_at_ms then
 end
 
 local last_checked_at_ms = tonumber(record.lastCheckedAtMs or '0')
-if checked_at_ms < last_checked_at_ms then
+if mode == 'global' then
+    if (success ~= '0' and success ~= '1') or not max_consecutive_failures or
+        max_consecutive_failures <= 0 or type(task_id) ~= 'string' or task_id == '' or
+        type(observation_digest) ~= 'string' or string.len(observation_digest) ~= 64 then
+        return finish({status = 'invalid'})
+    end
+    local last_task_id = record.lastHealthTaskId or ''
+    local last_observation_digest = record.lastHealthDigest or ''
+    if last_task_id == task_id then
+        if last_observation_digest == observation_digest then
+            return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs))
+        end
+        return finish({status = 'conflict'})
+    end
+    if checked_at_ms <= last_checked_at_ms then
         return finish({status = 'stale'})
-end
+    end
-if checked_at_ms == last_checked_at_ms then
+    local consecutive_failures = tonumber(record.consecutiveFailures or '0')
+    if not consecutive_failures or consecutive_failures < 0 then
+        return finish({status = 'invalid'})
+    end
+    if success == '1' then
+        if record.state ~= 'CHECKING' and record.state ~= 'AVAILABLE' and record.state ~= 'SUSPECT' then
+            return finish({status = 'invalid'})
+        end
+        next_state = 'AVAILABLE'
+        consecutive_failures = 0
+    else
+        consecutive_failures = consecutive_failures + 1
+        if record.state == 'CHECKING' then
+            next_state = 'UNHEALTHY'
+        elseif record.state == 'AVAILABLE' or record.state == 'SUSPECT' then
+            if consecutive_failures >= max_consecutive_failures then
+                next_state = 'UNHEALTHY'
+            else
+                next_state = 'SUSPECT'
+            end
+        else
+            return finish({status = 'invalid'})
+        end
+    end
+    record.consecutiveFailures = consecutive_failures
+    record.lastHealthTaskId = task_id
+    record.lastHealthDigest = observation_digest
+    record.lastHealthObservedAtMs = checked_at_ms
+else
+    if checked_at_ms < last_checked_at_ms then
+        return finish({status = 'stale'})
+    end
+    if checked_at_ms == last_checked_at_ms then
         if record.state ~= next_state then
             return finish({status = 'stale'})
         end
         return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs))
+    end
 end
 if record.state ~= next_state and not (transitions[record.state] and transitions[record.state][next_state]) then
     return finish({status = 'invalid'})

internal/adapters/redisactivity/scripts/upsert.lua

@@ -227,6 +227,10 @@ for _, candidate in ipairs(candidates) do
             incoming.state = current.state
             incoming.lastCheckedAtMs = current.lastCheckedAtMs
             incoming.lastSuccessAtMs = current.lastSuccessAtMs
+            incoming.consecutiveFailures = current.consecutiveFailures
+            incoming.lastHealthTaskId = current.lastHealthTaskId
+            incoming.lastHealthDigest = current.lastHealthDigest
+            incoming.lastHealthObservedAtMs = current.lastHealthObservedAtMs
             incoming.latencyNs = current.latencyNs
             incoming.ownerWorkerId = current.ownerWorkerId
             local encoded = cjson.encode(incoming)

internal/adapters/redisactivity/upsert.go

@@ -11,6 +11,7 @@ import (
 
 	"proxy-pool/internal/domain/activitypool"
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	healthDomain "proxy-pool/internal/domain/health"
 	proxyDomain "proxy-pool/internal/domain/proxy"
 	"proxy-pool/internal/platform/credentials"
 )
@@ -252,9 +253,19 @@ func proxyRecordEntry(record proxyRecord) activitypool.Entry {
 		value := time.UnixMilli(record.LastSuccessAtMS).UTC()
 		proxy.LastSuccessAt = &value
 	}
+	global := healthDomain.GlobalState{State: proxyDomain.State(record.State)}
+	if record.LastHealthTaskID != "" {
+		digest, err := hex.DecodeString(record.LastHealthDigest)
+		if err == nil && len(digest) == sha256.Size && record.LastHealthObservedAtMS > 0 {
+			copy(global.LastObservationDigest[:], digest)
+			global.LastTaskID = record.LastHealthTaskID
+			global.LastObservedAt = time.UnixMilli(record.LastHealthObservedAtMS).UTC()
+			global.ConsecutiveFailures = record.ConsecutiveFailures
+		}
+	}
 	return activitypool.Entry{
 		Proxy: proxy, UsableUntil: usableUntil,
-		OwnerWorkerID: record.OwnerWorkerID, State: proxyDomain.State(record.State),
+		OwnerWorkerID: record.OwnerWorkerID, State: proxyDomain.State(record.State), GlobalHealth: global,
 	}
 }
 

internal/domain/activitypool/global_health_test.go

@@ -0,0 +1,102 @@
+package activitypool
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	healthDomain "proxy-pool/internal/domain/health"
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+func TestMemoryPoolAppliesGlobalObservationAtomically(t *testing.T) {
+	now := time.Date(2026, 7, 31, 11, 0, 0, 0, time.UTC)
+	pool := seededHealthPool(t, now)
+	if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(checking): %v", err)
+	}
+	failed, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{
+		Observation: healthObservation("task-1", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 3,
+	})
+	if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 {
+		t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err)
+	}
+	if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(3 * time.Second), NextState: proxyDomain.StateChecking,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(recheck): %v", err)
+	}
+	available, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{
+		Observation: healthObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3,
+	})
+	if err != nil || available.State != proxyDomain.StateAvailable || available.GlobalHealth.ConsecutiveFailures != 0 ||
+		available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(now.Add(4*time.Second)) {
+		t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", available, err)
+	}
+}
+
+func TestMemoryPoolGlobalObservationReplayDoesNotIncreaseFailureStreak(t *testing.T) {
+	now := time.Date(2026, 7, 31, 11, 0, 0, 0, time.UTC)
+	pool := seededHealthPool(t, now)
+	if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{
+		ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
+	}); err != nil {
+		t.Fatalf("ApplyHealth(checking): %v", err)
+	}
+	command := GlobalHealthCommand{Observation: healthObservation("task-1", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 3}
+	first, err := pool.ApplyGlobalObservation(context.Background(), command)
+	if err != nil {
+		t.Fatalf("ApplyGlobalObservation(first): %v", err)
+	}
+	replayed, err := pool.ApplyGlobalObservation(context.Background(), command)
+	if err != nil || replayed.GlobalHealth != first.GlobalHealth {
+		t.Fatalf("ApplyGlobalObservation(replay) = %+v, %v", replayed, err)
+	}
+	conflicting := command
+	conflicting.Observation.Success = true
+	conflicting.Observation.FailureClass = ""
+	if _, err := pool.ApplyGlobalObservation(context.Background(), conflicting); !errors.Is(err, healthDomain.ErrConflictingObservation) {
+		t.Fatalf("ApplyGlobalObservation(conflict) error = %v, want ErrConflictingObservation", err)
+	}
+}
+
+func TestMemoryPoolRejectsTargetObservationFromGlobalStore(t *testing.T) {
+	now := time.Date(2026, 7, 31, 11, 0, 0, 0, time.UTC)
+	pool := seededHealthPool(t, now)
+	_, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{Observation: healthDomain.Observation{
+		TaskID: "task-target", ProxyID: "proxy-a", Level: healthDomain.LevelTarget, RoutingName: "route-a",
+		TargetURL: "https://target.example/", Success: false, FailureClass: "target_403", ObservedAt: now.Add(time.Second),
+	}, MaxConsecutiveFailures: 2})
+	if !errors.Is(err, ErrInvalidHealthUpdate) {
+		t.Fatalf("ApplyGlobalObservation(target) error = %v, want ErrInvalidHealthUpdate", err)
+	}
+}
+
+func seededHealthPool(t *testing.T, now time.Time) *MemoryPool {
+	t.Helper()
+	pool := NewMemoryPool()
+	_, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
+		ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
+		Proxies: []proxyDomain.Proxy{{
+			ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
+		}},
+	})
+	if err != nil {
+		t.Fatalf("UpsertFetched(): %v", err)
+	}
+	return pool
+}
+
+func healthObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation {
+	observation := healthDomain.Observation{
+		TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic,
+		Success: success, Latency: 10 * time.Millisecond, ObservedAt: observedAt,
+	}
+	if !success {
+		observation.FailureClass = "timeout"
+	}
+	return observation
+}

internal/domain/activitypool/pool.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	healthDomain "proxy-pool/internal/domain/health"
 	ownershipDomain "proxy-pool/internal/domain/ownership"
 	proxyDomain "proxy-pool/internal/domain/proxy"
 )
@@ -77,6 +78,18 @@ type HealthStore interface {
 	ApplyHealth(context.Context, HealthUpdate) (Entry, error)
 }
 
+// GlobalHealthCommand carries a BASIC or EGRESS fact to the authoritative
+// activity pool. TARGET observations use a separate profile store and never
+// enter this command.
+type GlobalHealthCommand struct {
+	Observation            healthDomain.Observation
+	MaxConsecutiveFailures int
+}
+
+type GlobalHealthStore interface {
+	ApplyGlobalObservation(context.Context, GlobalHealthCommand) (Entry, error)
+}
+
 type InventoryReader interface {
 	Inventory(context.Context, string, time.Time) (Inventory, error)
 }
@@ -94,6 +107,7 @@ type Entry struct {
 	UsableUntil   time.Time
 	OwnerWorkerID string
 	State         proxyDomain.State
+	GlobalHealth  healthDomain.GlobalState
 }
 
 type MemoryPool struct {
@@ -115,6 +129,7 @@ type idempotencyEntry struct {
 var (
 	_ Upserter                   = (*MemoryPool)(nil)
 	_ HealthStore                = (*MemoryPool)(nil)
+	_ GlobalHealthStore          = (*MemoryPool)(nil)
 	_ InventoryReader            = (*MemoryPool)(nil)
 	_ StateInventoryReader       = (*MemoryPool)(nil)
 	_ Maintainer                 = (*MemoryPool)(nil)
@@ -221,6 +236,7 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch
 				UsableUntil:   usableUntil,
 				OwnerWorkerID: current.OwnerWorkerID,
 				State:         current.State,
+				GlobalHealth:  current.GlobalHealth,
 			}
 			continue
 		}
@@ -236,6 +252,7 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch
 			Proxy:        cloneProxy(candidate),
 			UsableUntil:  usableUntil,
 			State:        candidate.State,
+			GlobalHealth: healthDomain.GlobalState{State: candidate.State},
 		}
 		p.keyByID[candidate.ID] = key
 		if managedActivityState(candidate.State) {
@@ -284,6 +301,7 @@ func (p *MemoryPool) ApplyHealth(ctx context.Context, update HealthUpdate) (Entr
 	checkedAt := update.CheckedAt.UTC()
 	entry.State = update.NextState
 	entry.Proxy.State = update.NextState
+	entry.GlobalHealth.State = update.NextState
 	entry.Proxy.LastCheckedAt = &checkedAt
 	entry.Proxy.Latency = update.Latency
 	if update.NextState == proxyDomain.StateAvailable {
@@ -295,6 +313,65 @@ func (p *MemoryPool) ApplyHealth(ctx context.Context, update HealthUpdate) (Entr
 	return entry, nil
 }
 
+// ApplyGlobalObservation atomically reduces a BASIC or EGRESS Checker fact
+// against the current Proxy health state and publishes the resulting state.
+func (p *MemoryPool) ApplyGlobalObservation(
+	ctx context.Context,
+	command GlobalHealthCommand,
+) (Entry, error) {
+	if ctx == nil {
+		return Entry{}, ErrInvalidHealthUpdate
+	}
+	if err := ctx.Err(); err != nil {
+		return Entry{}, err
+	}
+	normalized, err := healthDomain.NormalizeObservation(command.Observation)
+	if err != nil {
+		return Entry{}, err
+	}
+	if normalized.Level == healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 {
+		return Entry{}, ErrInvalidHealthUpdate
+	}
+	if p == nil {
+		return Entry{}, ErrInvalidHealthUpdate
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return Entry{}, err
+	}
+	p.purgeExpiredLocked(normalized.ObservedAt)
+	entry, ok := p.entryByIDLocked(normalized.ProxyID)
+	if !ok {
+		return Entry{}, ErrActivityNotFound
+	}
+	current := entry.GlobalHealth
+	if current.State == "" {
+		current.State = entry.State
+	}
+	if current.State != entry.State {
+		return Entry{}, ErrInvalidHealthUpdate
+	}
+	next, err := healthDomain.ReduceGlobal(current, normalized, command.MaxConsecutiveFailures)
+	if err != nil {
+		return Entry{}, err
+	}
+	if entry.State != next.State && !proxyDomain.CanTransition(entry.State, next.State) {
+		return Entry{}, ErrInvalidHealthUpdate
+	}
+	entry.State = next.State
+	entry.Proxy.State = next.State
+	entry.Proxy.LastCheckedAt = timePointer(normalized.ObservedAt)
+	entry.Proxy.Latency = normalized.Latency
+	if normalized.Success {
+		entry.Proxy.LastSuccessAt = timePointer(normalized.ObservedAt)
+	}
+	entry.GlobalHealth = next
+	p.setEntryByIDLocked(normalized.ProxyID, entry)
+	entry.Proxy = cloneProxy(entry.Proxy)
+	return entry, nil
+}
+
 func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) {
 	result := Inventory{UpstreamID: upstreamID}
 	if ctx == nil {
@@ -477,6 +554,7 @@ func (p *MemoryPool) Extract(ctx context.Context, command extractionDomain.Comma
 		entry := p.entries[key]
 		entry.State = proxyDomain.StateExtracted
 		entry.Proxy.State = proxyDomain.StateExtracted
+		entry.GlobalHealth.State = proxyDomain.StateExtracted
 		p.entries[key] = entry
 		result.Items = append(result.Items, extractionCandidate(entry))
 	}
@@ -488,6 +566,11 @@ func (p *MemoryPool) Extract(ctx context.Context, command extractionDomain.Comma
 	return result, nil
 }
 
+func timePointer(value time.Time) *time.Time {
+	canonical := value.UTC()
+	return &canonical
+}
+
 func (p *MemoryPool) rememberExtractionLocked(
 	idempotencyKey string,
 	command extractionDomain.Command,

internal/domain/health/health.go

@@ -0,0 +1,364 @@
+// Package health defines transport-free facts and deterministic health
+// reduction rules shared by the Controller and Checker integration layers.
+package health
+
+import (
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"net/netip"
+	"net/url"
+	"strings"
+	"time"
+
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+var (
+	ErrInvalidObservation      = errors.New("invalid health observation")
+	ErrInvalidGlobalState      = errors.New("invalid global health state")
+	ErrInvalidTargetState      = errors.New("invalid target health state")
+	ErrNonGlobalObservation    = errors.New("target health observation cannot change global proxy health")
+	ErrNonTargetObservation    = errors.New("global health observation cannot change target profile health")
+	ErrStaleObservation        = errors.New("stale health observation")
+	ErrConflictingObservation  = errors.New("conflicting health observation replay")
+	ErrInvalidFailureThreshold = errors.New("invalid health failure threshold")
+	ErrInvalidCheckPreparation = errors.New("invalid health check preparation")
+)
+
+type Level string
+
+const (
+	LevelBasic  Level = "BASIC"
+	LevelEgress Level = "EGRESS"
+	LevelTarget Level = "TARGET"
+)
+
+type TargetStatus string
+
+const (
+	TargetUnknown   TargetStatus = "UNKNOWN"
+	TargetAvailable TargetStatus = "AVAILABLE"
+	TargetSuspect   TargetStatus = "SUSPECT"
+	TargetUnhealthy TargetStatus = "UNHEALTHY"
+)
+
+// Observation is an immutable result produced by a Checker. It has no
+// authority to modify a Proxy until the Controller reduces and commits it.
+type Observation struct {
+	TaskID           string
+	ProxyID          string
+	Level            Level
+	RoutingName      string
+	TargetURL        string
+	Success          bool
+	FailureClass     string
+	Latency          time.Duration
+	ObservedEgressIP string
+	ObservedAt       time.Time
+}
+
+type TargetProfile struct {
+	RoutingName string
+	TargetURL   string
+}
+
+// GlobalState is the small authoritative state needed to reduce BASIC and
+// EGRESS observations. The persistence adapter owns storing it atomically.
+type GlobalState struct {
+	State                 proxyDomain.State
+	ConsecutiveFailures   int
+	LastTaskID            string
+	LastObservedAt        time.Time
+	LastObservationDigest [sha256.Size]byte
+}
+
+// TargetState is isolated by (proxy, routing, target URL). A target failure
+// never changes a Proxy's GlobalState.
+type TargetState struct {
+	Status                TargetStatus
+	ConsecutiveFailures   int
+	LastTaskID            string
+	LastObservedAt        time.Time
+	LastSuccessAt         time.Time
+	Latency               time.Duration
+	LastObservationDigest [sha256.Size]byte
+}
+
+// NormalizeObservation validates and canonicalizes an observation before it
+// crosses an idempotency or persistence boundary.
+func NormalizeObservation(value Observation) (Observation, error) {
+	value.TaskID = strings.TrimSpace(value.TaskID)
+	value.ProxyID = strings.TrimSpace(value.ProxyID)
+	value.RoutingName = strings.TrimSpace(value.RoutingName)
+	value.TargetURL = strings.TrimSpace(value.TargetURL)
+	value.FailureClass = strings.TrimSpace(value.FailureClass)
+	value.ObservedEgressIP = strings.TrimSpace(value.ObservedEgressIP)
+	if !validIdentifier(value.TaskID) || !validIdentifier(value.ProxyID) || value.ObservedAt.IsZero() || value.Latency < 0 {
+		return Observation{}, ErrInvalidObservation
+	}
+	if value.Success && value.FailureClass != "" {
+		return Observation{}, ErrInvalidObservation
+	}
+	if value.ObservedEgressIP != "" {
+		if _, err := netip.ParseAddr(value.ObservedEgressIP); err != nil {
+			return Observation{}, ErrInvalidObservation
+		}
+	}
+	switch value.Level {
+	case LevelBasic, LevelEgress:
+		if value.RoutingName != "" || value.TargetURL != "" {
+			return Observation{}, ErrInvalidObservation
+		}
+	case LevelTarget:
+		profile, err := NormalizeTargetProfile(TargetProfile{RoutingName: value.RoutingName, TargetURL: value.TargetURL})
+		if err != nil {
+			return Observation{}, err
+		}
+		value.RoutingName = profile.RoutingName
+		value.TargetURL = profile.TargetURL
+	default:
+		return Observation{}, ErrInvalidObservation
+	}
+	value.ObservedAt = value.ObservedAt.UTC()
+	return value, nil
+}
+
+func NormalizeTargetProfile(value TargetProfile) (TargetProfile, error) {
+	value.RoutingName = strings.TrimSpace(value.RoutingName)
+	value.TargetURL = strings.TrimSpace(value.TargetURL)
+	if !validIdentifier(value.RoutingName) || value.TargetURL == "" {
+		return TargetProfile{}, ErrInvalidObservation
+	}
+	parsed, err := url.Parse(value.TargetURL)
+	if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" {
+		return TargetProfile{}, ErrInvalidObservation
+	}
+	value.TargetURL = parsed.String()
+	return value, nil
+}
+
+func (value TargetProfile) Key() string {
+	return value.RoutingName + "\x00" + value.TargetURL
+}
+
+func ObservationDigest(value Observation) ([sha256.Size]byte, error) {
+	normalized, err := NormalizeObservation(value)
+	if err != nil {
+		return [sha256.Size]byte{}, err
+	}
+	payload := strings.Join([]string{
+		normalized.TaskID, normalized.ProxyID, string(normalized.Level), normalized.RoutingName,
+		normalized.TargetURL, fmt.Sprintf("%t", normalized.Success), normalized.FailureClass,
+		normalized.Latency.String(), normalized.ObservedEgressIP, normalized.ObservedAt.Format(time.RFC3339Nano),
+	}, "\x00")
+	return sha256.Sum256([]byte(payload)), nil
+}
+
+// BeginGlobalCheck moves only states that are deliberately unavailable during
+// probing into CHECKING. AVAILABLE and SUSPECT remain serving states while a
+// periodic probe is in flight.
+func BeginGlobalCheck(current GlobalState) (proxyDomain.State, error) {
+	if err := validateGlobalState(current); err != nil {
+		return "", err
+	}
+	switch current.State {
+	case proxyDomain.StateFetched, proxyDomain.StateUnhealthy:
+		return proxyDomain.StateChecking, nil
+	case proxyDomain.StateChecking, proxyDomain.StateAvailable, proxyDomain.StateSuspect:
+		return current.State, nil
+	default:
+		return "", ErrInvalidCheckPreparation
+	}
+}
+
+// ReduceGlobal reduces one BASIC or EGRESS fact. It is intentionally pure so
+// Memory and Redis adapters can apply the same result atomically.
+func ReduceGlobal(current GlobalState, observation Observation, maxConsecutiveFailures int) (GlobalState, error) {
+	if maxConsecutiveFailures <= 0 {
+		return GlobalState{}, ErrInvalidFailureThreshold
+	}
+	if err := validateGlobalState(current); err != nil {
+		return GlobalState{}, err
+	}
+	normalized, err := NormalizeObservation(observation)
+	if err != nil {
+		return GlobalState{}, err
+	}
+	if normalized.Level == LevelTarget {
+		return GlobalState{}, ErrNonGlobalObservation
+	}
+	digest, err := ObservationDigest(normalized)
+	if err != nil {
+		return GlobalState{}, err
+	}
+	if err := checkObservationOrder(current.LastTaskID, current.LastObservedAt, current.LastObservationDigest, normalized, digest); err != nil {
+		if errors.Is(err, errExactReplay) {
+			return current, nil
+		}
+		return GlobalState{}, err
+	}
+	next := current
+	next.LastTaskID = normalized.TaskID
+	next.LastObservedAt = normalized.ObservedAt
+	next.LastObservationDigest = digest
+	if normalized.Success {
+		switch current.State {
+		case proxyDomain.StateChecking, proxyDomain.StateAvailable, proxyDomain.StateSuspect:
+			next.State = proxyDomain.StateAvailable
+			next.ConsecutiveFailures = 0
+			return next, nil
+		default:
+			return GlobalState{}, ErrInvalidGlobalState
+		}
+	}
+	switch current.State {
+	case proxyDomain.StateChecking:
+		next.ConsecutiveFailures++
+		next.State = proxyDomain.StateUnhealthy
+		return next, nil
+	case proxyDomain.StateAvailable, proxyDomain.StateSuspect:
+		next.ConsecutiveFailures++
+		if next.ConsecutiveFailures >= maxConsecutiveFailures {
+			next.State = proxyDomain.StateUnhealthy
+		} else {
+			next.State = proxyDomain.StateSuspect
+		}
+		return next, nil
+	default:
+		return GlobalState{}, ErrInvalidGlobalState
+	}
+}
+
+// ReduceTarget reduces one TARGET fact into only its target profile state.
+// Callers must keep this result separate from activity-pool global state.
+func ReduceTarget(current TargetState, observation Observation, maxConsecutiveFailures int) (TargetState, error) {
+	if maxConsecutiveFailures <= 0 {
+		return TargetState{}, ErrInvalidFailureThreshold
+	}
+	if current.Status == "" {
+		current.Status = TargetUnknown
+	}
+	if err := validateTargetState(current); err != nil {
+		return TargetState{}, err
+	}
+	normalized, err := NormalizeObservation(observation)
+	if err != nil {
+		return TargetState{}, err
+	}
+	if normalized.Level != LevelTarget {
+		return TargetState{}, ErrNonTargetObservation
+	}
+	digest, err := ObservationDigest(normalized)
+	if err != nil {
+		return TargetState{}, err
+	}
+	if err := checkObservationOrder(current.LastTaskID, current.LastObservedAt, current.LastObservationDigest, normalized, digest); err != nil {
+		if errors.Is(err, errExactReplay) {
+			return current, nil
+		}
+		return TargetState{}, err
+	}
+	next := current
+	next.LastTaskID = normalized.TaskID
+	next.LastObservedAt = normalized.ObservedAt
+	next.LastObservationDigest = digest
+	next.Latency = normalized.Latency
+	if normalized.Success {
+		next.Status = TargetAvailable
+		next.ConsecutiveFailures = 0
+		next.LastSuccessAt = normalized.ObservedAt
+		return next, nil
+	}
+	next.ConsecutiveFailures++
+	if next.ConsecutiveFailures >= maxConsecutiveFailures {
+		next.Status = TargetUnhealthy
+	} else {
+		next.Status = TargetSuspect
+	}
+	return next, nil
+}
+
+var errExactReplay = errors.New("exact health observation replay")
+
+func checkObservationOrder(
+	lastTaskID string,
+	lastObservedAt time.Time,
+	lastDigest [sha256.Size]byte,
+	observation Observation,
+	digest [sha256.Size]byte,
+) error {
+	if lastObservedAt.IsZero() {
+		return nil
+	}
+	if observation.ObservedAt.Before(lastObservedAt) {
+		return ErrStaleObservation
+	}
+	if observation.TaskID == lastTaskID {
+		if digest == lastDigest {
+			return errExactReplay
+		}
+		return ErrConflictingObservation
+	}
+	if observation.ObservedAt.Equal(lastObservedAt) {
+		return ErrStaleObservation
+	}
+	return nil
+}
+
+func validateGlobalState(value GlobalState) error {
+	if value.ConsecutiveFailures < 0 {
+		return ErrInvalidGlobalState
+	}
+	switch value.State {
+	case proxyDomain.StateFetched, proxyDomain.StateChecking, proxyDomain.StateAvailable,
+		proxyDomain.StateSuspect, proxyDomain.StateUnhealthy:
+	default:
+		return ErrInvalidGlobalState
+	}
+	if value.LastObservedAt.IsZero() {
+		if value.LastTaskID != "" || value.ConsecutiveFailures != 0 || value.LastObservationDigest != ([sha256.Size]byte{}) {
+			return ErrInvalidGlobalState
+		}
+		return nil
+	}
+	if !validIdentifier(value.LastTaskID) || value.LastObservationDigest == ([sha256.Size]byte{}) {
+		return ErrInvalidGlobalState
+	}
+	return nil
+}
+
+func validateTargetState(value TargetState) error {
+	if value.ConsecutiveFailures < 0 || value.Latency < 0 {
+		return ErrInvalidTargetState
+	}
+	switch value.Status {
+	case TargetUnknown, TargetAvailable, TargetSuspect, TargetUnhealthy:
+	default:
+		return ErrInvalidTargetState
+	}
+	if value.LastObservedAt.IsZero() {
+		if value.Status != TargetUnknown || value.LastTaskID != "" || value.ConsecutiveFailures != 0 ||
+			!value.LastSuccessAt.IsZero() || value.LastObservationDigest != ([sha256.Size]byte{}) {
+			return ErrInvalidTargetState
+		}
+		return nil
+	}
+	if !validIdentifier(value.LastTaskID) || value.LastObservationDigest == ([sha256.Size]byte{}) ||
+		(!value.LastSuccessAt.IsZero() && value.LastSuccessAt.After(value.LastObservedAt)) {
+		return ErrInvalidTargetState
+	}
+	return nil
+}
+
+func validIdentifier(value string) bool {
+	if value == "" || len(value) > 256 {
+		return false
+	}
+	for _, character := range value {
+		if character <= ' ' || character == '\x7f' {
+			return false
+		}
+	}
+	return true
+}

internal/domain/health/health_test.go

@@ -0,0 +1,95 @@
+package health
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	proxyDomain "proxy-pool/internal/domain/proxy"
+)
+
+func TestReduceGlobalAppliesSuspectUnhealthyAndRecovery(t *testing.T) {
+	now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
+	current := GlobalState{State: proxyDomain.StateAvailable}
+	first, err := ReduceGlobal(current, globalObservation("task-1", false, now), 2)
+	if err != nil || first.State != proxyDomain.StateSuspect || first.ConsecutiveFailures != 1 {
+		t.Fatalf("first failure = %+v, %v; want SUSPECT / 1", first, err)
+	}
+	second, err := ReduceGlobal(first, globalObservation("task-2", false, now.Add(time.Second)), 2)
+	if err != nil || second.State != proxyDomain.StateUnhealthy || second.ConsecutiveFailures != 2 {
+		t.Fatalf("second failure = %+v, %v; want UNHEALTHY / 2", second, err)
+	}
+	if next, err := BeginGlobalCheck(second); err != nil || next != proxyDomain.StateChecking {
+		t.Fatalf("BeginGlobalCheck(unhealthy) = %s, %v; want CHECKING", next, err)
+	}
+	recovering := second
+	recovering.State = proxyDomain.StateChecking
+	recovered, err := ReduceGlobal(recovering, globalObservation("task-3", true, now.Add(2*time.Second)), 2)
+	if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.ConsecutiveFailures != 0 {
+		t.Fatalf("recovery = %+v, %v; want AVAILABLE / 0", recovered, err)
+	}
+}
+
+func TestReduceGlobalMarksFailedInitialCheckUnhealthy(t *testing.T) {
+	now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
+	next, err := ReduceGlobal(GlobalState{State: proxyDomain.StateChecking}, globalObservation("task-1", false, now), 3)
+	if err != nil || next.State != proxyDomain.StateUnhealthy || next.ConsecutiveFailures != 1 {
+		t.Fatalf("initial failure = %+v, %v; want UNHEALTHY / 1", next, err)
+	}
+}
+
+func TestReduceGlobalRejectsStaleAndConflictingReplays(t *testing.T) {
+	now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
+	first, err := ReduceGlobal(GlobalState{State: proxyDomain.StateAvailable}, globalObservation("task-1", false, now), 2)
+	if err != nil {
+		t.Fatalf("ReduceGlobal(first): %v", err)
+	}
+	replayed, err := ReduceGlobal(first, globalObservation("task-1", false, now), 2)
+	if err != nil || replayed != first {
+		t.Fatalf("ReduceGlobal(replay) = %+v, %v; want unchanged", replayed, err)
+	}
+	if _, err := ReduceGlobal(first, globalObservation("task-1", true, now), 2); !errors.Is(err, ErrConflictingObservation) {
+		t.Fatalf("conflicting replay error = %v, want ErrConflictingObservation", err)
+	}
+	if _, err := ReduceGlobal(first, globalObservation("task-2", true, now.Add(-time.Second)), 2); !errors.Is(err, ErrStaleObservation) {
+		t.Fatalf("stale observation error = %v, want ErrStaleObservation", err)
+	}
+}
+
+func TestReduceTargetNeverChangesGlobalState(t *testing.T) {
+	now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
+	observation := Observation{
+		TaskID: "task-target-1", ProxyID: "proxy-a", Level: LevelTarget, RoutingName: "route-a",
+		TargetURL: "https://target.example/path?volatile=true", Success: false, FailureClass: "target_403",
+		Latency: 20 * time.Millisecond, ObservedAt: now,
+	}
+	target, err := ReduceTarget(TargetState{}, observation, 2)
+	if err != nil || target.Status != TargetSuspect || target.ConsecutiveFailures != 1 {
+		t.Fatalf("ReduceTarget(first) = %+v, %v; want SUSPECT / 1", target, err)
+	}
+	if _, err := ReduceGlobal(GlobalState{State: proxyDomain.StateAvailable}, observation, 2); !errors.Is(err, ErrNonGlobalObservation) {
+		t.Fatalf("ReduceGlobal(target) error = %v, want ErrNonGlobalObservation", err)
+	}
+	profile, err := NormalizeTargetProfile(TargetProfile{RoutingName: observation.RoutingName, TargetURL: observation.TargetURL})
+	if err != nil || profile.TargetURL != "https://target.example/path?volatile=true" {
+		t.Fatalf("NormalizeTargetProfile() = %+v, %v", profile, err)
+	}
+}
+
+func TestNormalizeObservationRejectsCrossLevelFields(t *testing.T) {
+	now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
+	_, err := NormalizeObservation(Observation{
+		TaskID: "task-a", ProxyID: "proxy-a", Level: LevelBasic, RoutingName: "route-a", Success: true, ObservedAt: now,
+	})
+	if !errors.Is(err, ErrInvalidObservation) {
+		t.Fatalf("NormalizeObservation() error = %v, want ErrInvalidObservation", err)
+	}
+}
+
+func globalObservation(taskID string, success bool, observedAt time.Time) Observation {
+	observation := Observation{TaskID: taskID, ProxyID: "proxy-a", Level: LevelBasic, Success: success, ObservedAt: observedAt}
+	if !success {
+		observation.FailureClass = "timeout"
+	}
+	return observation
+}