feat: support egress probe task contracts

internal/checker/controlplane/runner_test.go

@@ -41,6 +41,29 @@ func TestRunnerExecutesBoundedTaskBatchAndReportsFacts(t *testing.T) {
 	}
 }
 
+func TestRunnerOmitsEgressProbeURLFromGlobalObservation(t *testing.T) {
+	now := time.Date(2026, 8, 1, 13, 30, 0, 0, time.UTC)
+	task := checkerTask("task-egress", now)
+	task.Level = controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS
+	task.TargetUrl = "https://egress.example/identity"
+	client := &clientStub{stream: &taskStreamStub{tasks: []*controlplanev1.CheckTask{task}}}
+	runner, err := NewRunner(client, &executorStub{}, Options{
+		CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 1,
+		SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS},
+		ReportBatchSize: 1, RetryDelay: time.Millisecond, Now: func() time.Time { return now },
+	})
+	if err != nil {
+		t.Fatalf("NewRunner(): %v", err)
+	}
+	if err := runner.RunOnce(context.Background()); err != nil {
+		t.Fatalf("RunOnce(): %v", err)
+	}
+	observation := client.batches[0].GetObservations()[0]
+	if observation.GetLevel() != controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS || observation.GetTargetUrl() != "" || observation.GetRoutingName() != "" {
+		t.Fatalf("Observation = %+v, want global EGRESS fact without target profile", observation)
+	}
+}
+
 func checkerTask(id string, now time.Time) *controlplanev1.CheckTask {
 	return &controlplanev1.CheckTask{
 		TaskId: id, ProxyId: id + "-proxy", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP,

internal/checker/probe/probe.go

@@ -136,6 +136,9 @@ func prepare(task *controlplanev1.CheckTask, now time.Time) (time.Time, string,
 		}
 		target = task.GetTargetUrl()
 	case controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS:
+		if task.GetRoutingName() != "" || task.GetTargetUrl() == "" {
+			return time.Time{}, "", nil, errors.New("invalid egress task")
+		}
 		target = task.GetTargetUrl()
 	default:
 		return time.Time{}, "", nil, errors.New("unsupported task level")

internal/checker/probe/probe_test.go

@@ -60,6 +60,20 @@ func TestExecutorTargetReportsHTTPFailureAsFact(t *testing.T) {
 	}
 }
 
+func TestExecutorRejectsEgressTaskWithRoutingProfile(t *testing.T) {
+	proxy := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) {
+		response.WriteHeader(http.StatusOK)
+	}))
+	defer proxy.Close()
+	task := validTask(t, proxy.URL, controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS)
+	task.RoutingName = "route-a"
+	task.TargetUrl = "https://egress.example/identity"
+	result := NewExecutor().Execute(context.Background(), task)
+	if result.Success || result.FailureClass != FailureInvalidTask {
+		t.Fatalf("Execute(EGRESS with routing) = %+v", result)
+	}
+}
+
 func TestExecutorReportsUnsupportedProtocolAsFact(t *testing.T) {
 	now := time.Now().UTC()
 	result := NewExecutor().Execute(context.Background(), &controlplanev1.CheckTask{

internal/controller/health/grpc_handler.go

@@ -239,13 +239,23 @@ func wireCheckTask(task LeasedTask, now time.Time) (*controlplanev1.CheckTask, e
 	if !ok {
 		return nil, ErrInvalidLeasedTask
 	}
-	if task.Level == healthDomain.LevelTarget {
+	switch task.Level {
+	case healthDomain.LevelBasic:
+		if task.RoutingName != "" || task.TargetURL != "" {
+			return nil, ErrInvalidLeasedTask
+		}
+	case healthDomain.LevelEgress:
+		targetURL, err := healthDomain.NormalizeEgressTarget(task.TargetURL)
+		if task.RoutingName != "" || err != nil || targetURL != task.TargetURL {
+			return nil, ErrInvalidLeasedTask
+		}
+	case healthDomain.LevelTarget:
 		if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{
 			RoutingName: task.RoutingName, TargetURL: task.TargetURL,
 		}); err != nil {
 			return nil, ErrInvalidLeasedTask
 		}
-	} else if task.RoutingName != "" || task.TargetURL != "" {
+	default:
 		return nil, ErrInvalidLeasedTask
 	}
 	return &controlplanev1.CheckTask{

internal/controller/health/grpc_handler_test.go

@@ -173,6 +173,60 @@ func TestGRPCHandlerStreamsLeasedTasksAndFencesReportedFacts(t *testing.T) {
 	}
 }
 
+func TestGRPCHandlerStreamsEgressProbeURLAndAcceptsGlobalFact(t *testing.T) {
+	now := time.Date(2026, 8, 1, 12, 30, 0, 0, time.UTC)
+	broker, err := NewMemoryTaskBroker(MemoryTaskBrokerOptions{
+		LeaseTTL: time.Minute, Now: func() time.Time { return now },
+		Material: TaskMaterialResolverFunc(func(context.Context, Candidate) (TaskMaterial, error) {
+			return TaskMaterial{Protocol: proxyDomain.SchemeHTTP, Host: "proxy.example", Port: 8080}, nil
+		}),
+	})
+	if err != nil {
+		t.Fatalf("NewMemoryTaskBroker(): %v", err)
+	}
+	const egressURL = "https://egress.example/identity"
+	if offered, offerErr := broker.Offer(context.Background(), []PlannedTask{{
+		Candidate: Candidate{ProxyID: "proxy-a", State: proxyDomain.StateAvailable, Level: healthDomain.LevelEgress, TargetURL: egressURL, DueAt: now},
+		Deadline:  now.Add(10 * time.Second), Attempts: 1,
+	}}); offerErr != nil || offered != 1 {
+		t.Fatalf("Offer() = (%d, %v)", offered, offerErr)
+	}
+	global := &recordingGlobalStore{}
+	reducer, err := NewReducer(global, &recordingTargetStore{}, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil })
+	if err != nil {
+		t.Fatalf("NewReducer(): %v", err)
+	}
+	handler, err := NewGRPCHandler(reducer, &recordingCheckerIdentity{}, GRPCHandlerOptions{
+		MaxObservationsPerBatch: 1, MaxTasksPerClaim: 1, TaskBroker: broker, Now: func() time.Time { return now },
+	})
+	if err != nil {
+		t.Fatalf("NewGRPCHandler(): %v", err)
+	}
+	client, closeClient := newCheckerGRPCClient(t, handler)
+	defer closeClient()
+
+	stream, err := client.StreamCheckTasks(context.Background(), &controlplanev1.StreamCheckTasksRequest{
+		CheckerId: "checker-a", InstanceId: "instance-a", MaxInFlight: 1,
+		SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS},
+	})
+	if err != nil {
+		t.Fatalf("StreamCheckTasks(): %v", err)
+	}
+	task, err := stream.Recv()
+	if err != nil || task.GetLevel() != controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS || task.GetTargetUrl() != egressURL || task.GetRoutingName() != "" {
+		t.Fatalf("Recv() = (%+v, %v)", task, err)
+	}
+	observation := grpcHealthObservation(task.GetTaskId(), controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS, now.Add(time.Second))
+	observation.LeaseToken = task.GetLeaseToken()
+	response, err := client.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{
+		CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{observation},
+	})
+	if err != nil || response.GetAccepted() != 1 || len(global.commands) != 1 ||
+		global.commands[0].Observation.TargetURL != "" || global.commands[0].Observation.RoutingName != "" {
+		t.Fatalf("ReportObservations() = (%+v, %v), commands=%+v", response, err, global.commands)
+	}
+}
+
 func newCheckerGRPCClient(t *testing.T, handler controlplanev1.CheckerControlPlaneServer) (controlplanev1.CheckerControlPlaneClient, func()) {
 	t.Helper()
 	listener := bufconn.Listen(1 << 20)

internal/controller/health/scheduler.go

@@ -142,10 +142,15 @@ func (planner *Planner) validateCandidate(candidate Candidate) (Priority, bool,
 		return 0, false, ErrInvalidScheduleRequest
 	}
 	switch candidate.Level {
-	case healthDomain.LevelBasic, healthDomain.LevelEgress:
+	case healthDomain.LevelBasic:
 		if candidate.RoutingName != "" || candidate.TargetURL != "" {
 			return 0, false, ErrInvalidScheduleRequest
 		}
+	case healthDomain.LevelEgress:
+		targetURL, err := healthDomain.NormalizeEgressTarget(candidate.TargetURL)
+		if candidate.RoutingName != "" || err != nil || targetURL != candidate.TargetURL {
+			return 0, false, ErrInvalidScheduleRequest
+		}
 	case healthDomain.LevelTarget:
 		if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{
 			RoutingName: candidate.RoutingName, TargetURL: candidate.TargetURL,

internal/controller/health/scheduler_test.go

@@ -19,7 +19,7 @@ func TestPlannerPrioritizesAndBoundsOverdueCandidates(t *testing.T) {
 	now := time.Date(2026, 7, 31, 17, 0, 0, 0, time.UTC)
 	plans, err := planner.Plan(now, 1, 10, []Candidate{
 		{ProxyID: "available", State: proxyDomain.StateAvailable, Level: healthDomain.LevelBasic, DueAt: now.Add(-time.Second)},
-		{ProxyID: "suspect", State: proxyDomain.StateSuspect, Level: healthDomain.LevelEgress, DueAt: now.Add(-2 * time.Second)},
+		{ProxyID: "suspect", State: proxyDomain.StateSuspect, Level: healthDomain.LevelEgress, TargetURL: "https://egress.example/identity", DueAt: now.Add(-2 * time.Second)},
 		{ProxyID: "fetched", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now.Add(-3 * time.Second)},
 		{ProxyID: "future", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now.Add(time.Second)},
 	})

internal/controller/health/task_broker.go

@@ -352,10 +352,15 @@ func validatePlannedTask(task PlannedTask, now time.Time) error {
 		return ErrInvalidLeasedTask
 	}
 	switch task.Candidate.Level {
-	case healthDomain.LevelBasic, healthDomain.LevelEgress:
+	case healthDomain.LevelBasic:
 		if task.Candidate.RoutingName != "" || task.Candidate.TargetURL != "" {
 			return ErrInvalidLeasedTask
 		}
+	case healthDomain.LevelEgress:
+		targetURL, err := healthDomain.NormalizeEgressTarget(task.Candidate.TargetURL)
+		if task.Candidate.RoutingName != "" || err != nil || targetURL != task.Candidate.TargetURL {
+			return ErrInvalidLeasedTask
+		}
 	case healthDomain.LevelTarget:
 		if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{
 			RoutingName: task.Candidate.RoutingName, TargetURL: task.Candidate.TargetURL,
@@ -403,8 +408,17 @@ func newLeaseToken() (string, error) {
 }
 
 func taskMatchesObservation(task PlannedTask, observation healthDomain.Observation) bool {
-	return task.Candidate.ProxyID == observation.ProxyID && task.Candidate.Level == observation.Level &&
-		task.Candidate.RoutingName == observation.RoutingName && task.Candidate.TargetURL == observation.TargetURL
+	if task.Candidate.ProxyID != observation.ProxyID || task.Candidate.Level != observation.Level {
+		return false
+	}
+	switch task.Candidate.Level {
+	case healthDomain.LevelBasic, healthDomain.LevelEgress:
+		return observation.RoutingName == "" && observation.TargetURL == ""
+	case healthDomain.LevelTarget:
+		return task.Candidate.RoutingName == observation.RoutingName && task.Candidate.TargetURL == observation.TargetURL
+	default:
+		return false
+	}
 }
 
 func validTaskIdentifier(value string) bool {

internal/domain/health/health.go

@@ -145,6 +145,22 @@ func NormalizeTargetProfile(value TargetProfile) (TargetProfile, error) {
 	return value, nil
 }
 
+// NormalizeEgressTarget canonicalizes the HTTP endpoint used to observe a
+// proxy's public egress identity. The endpoint is task execution material,
+// never part of the resulting global health fact.
+func NormalizeEgressTarget(value string) (string, error) {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return "", ErrInvalidObservation
+	}
+	parsed, err := url.Parse(value)
+	if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" ||
+		(parsed.Scheme != "http" && parsed.Scheme != "https") {
+		return "", ErrInvalidObservation
+	}
+	return parsed.String(), nil
+}
+
 func (value TargetProfile) Key() string {
 	return value.RoutingName + "\x00" + value.TargetURL
 }

internal/domain/health/health_test.go

@@ -86,6 +86,16 @@ func TestNormalizeObservationRejectsCrossLevelFields(t *testing.T) {
 	}
 }
 
+func TestNormalizeEgressTargetAcceptsOnlyHTTPOrHTTPSURL(t *testing.T) {
+	target, err := NormalizeEgressTarget("https://egress.example/identity?format=json")
+	if err != nil || target != "https://egress.example/identity?format=json" {
+		t.Fatalf("NormalizeEgressTarget(valid) = (%q, %v)", target, err)
+	}
+	if _, err := NormalizeEgressTarget("ftp://egress.example/identity"); !errors.Is(err, ErrInvalidObservation) {
+		t.Fatalf("NormalizeEgressTarget(ftp) 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 {